欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

Go異步任務解決方案之Asynq庫詳解

 更新時間:2023年02月15日 12:38:19   作者:王中陽Go  
需要在Go應用程序中異步處理任務? Asynq,簡單高效的任務隊列實現(xiàn),下面這篇文章主要給大家介紹了關(guān)于Go異步任務解決方案之Asynq庫的相關(guān)資料,需要的朋友可以參考下

今天為大家介紹一個Go處理異步任務的解決方案:Asynq,是一個 Go 庫,用于排隊任務并與 worker 異步處理它們。它由Redis提供支持,旨在實現(xiàn)可擴展且易于上手。

一、概述

Asynq 是一個 Go 庫,用于對任務進行排隊并與工作人員異步處理它們。

  • Asynq 工作原理的高級概述:
    • 客戶端將任務放入隊列
    • 服務器從隊列中拉出任務并為每個任務啟動一個工作 goroutine
    • 多個工作人員同時處理任務
  • git庫地址:

https://github.com/hibiken/asynq

二、快速開始

1. 準備工作

確保已安裝并運行了redis

redis 版本大于5.0

redis-server

目錄結(jié)構(gòu)

.
├── conf
│   └── redis.conf
└── docker-compose.yml

docker-compose.yml

version: '3.8'
services:
  myredis:
    container_name: myredis
    image: redis:6.2.5 #6.0.6
    restart: always
    ports:
      - 6379:6379
    privileged: true
    environment:
      # 時區(qū)上海
      TZ: Asia/Shanghai
    command: redis-server /etc/redis/redis.conf --appendonly yes
    volumes:
      - $PWD/data:/data
      - $PWD/conf/redis.conf:/etc/redis/redis.conf
   # networks:
    #  - myweb

#networks:
 # myweb:
  #  driver: bridge

conf/redis.conf

#開啟保護
protected-mode yes
#開啟遠程連接 
#bind 127.0.0.1 
#自定義密碼
requirepass 123456 
port 6379
timeout 0
# 900s內(nèi)至少一次寫操作則執(zhí)行bgsave進行RDB持久化
save 900 1 
save 300 10
save 60 10000
# rdbcompression ;默認值是yes。對于存儲到磁盤中的快照,可以設置是否進行壓縮存儲。如果是的話,redis會采用LZF算法進行壓縮。如果你不想消耗CPU來進行壓縮的話,可以設置為關(guān)閉此功能,但是存儲在磁盤上的快照會比較大。
rdbcompression yes
# dbfilename :設置快照的文件名,默認是 dump.rdb
dbfilename dump.rdb
# dir:設置快照文件的存放路徑,這個配置項一定是個目錄,而不能是文件名。使用上面的 dbfilename 作為保存的文件名。
dir /data
# 默認redis使用的是rdb方式持久化,這種方式在許多應用中已經(jīng)足夠用了。但是redis如果中途宕機,會導致可能有幾分鐘的數(shù)據(jù)丟失,根據(jù)save來策略進行持久化,Append Only File是另一種持久化方式, 可以提供更好的持久化特性。Redis會把每次寫入的數(shù)據(jù)在接收后都寫入appendonly.aof文件,每次啟動時Redis都會先把這個文件的數(shù)據(jù)讀入內(nèi)存里,先忽略RDB文件。默認值為no。
appendonly yes
# appendfilename :aof文件名,默認是"appendonly.aof"
# appendfsync:aof持久化策略的配置;no表示不執(zhí)行fsync,由操作系統(tǒng)保證數(shù)據(jù)同步到磁盤,速度最快;always表示每次寫入都執(zhí)行fsync,以保證數(shù)據(jù)同步到磁盤;everysec表示每秒執(zhí)行一次fsync,可能會導致丟失這1s數(shù)據(jù)
appendfsync everysec

啟動 redis 服務

docker-compose up -d

2. 安裝asynq軟件包

go get -u github.com/hibiken/asynq

3. 創(chuàng)建項目asynq_task

目錄結(jié)構(gòu):

.
|-- README.md
|-- cmd
|   `-- main.go  # 啟動消費者監(jiān)聽
|-- go.mod
|-- go.sum
|-- test.go # 生產(chǎn)者 發(fā)送測試數(shù)據(jù)
`-- test_delivery
    |-- client 
    |   `-- client.go # 生產(chǎn)者 具體發(fā)送測試數(shù)據(jù)的邏輯
    `-- test_delivery.go  # 消費者,執(zhí)行任務具體處理邏輯

Redis連接項

Asynq 使用 Redis 作為消息代理。

client.gomain.go 都需要連接到 Redis 進行寫入和讀取。

我們將使用 asynq.RedisClientOpt 指定如何連接到本地 Redis 實例。

asynq.RedisClientOpt{
	Addr:     "127.0.0.1:6379",
	Password: "",
	DB:       2,
}

4. Task任務

*asynq.Task

type Task struct {
	// 一個簡單的字符串值,表示要執(zhí)行的任務的類型.
	typename string

	// 有效載荷保存執(zhí)行任務所需的數(shù)據(jù),有效負載值必須是可序列化的.
	payload []byte

	// 保存任務的選項.
	opts []Option

	// 任務的結(jié)果編寫器.
	w *ResultWriter
}

5. 編寫程序

1)test_delivery.go 一個封裝任務創(chuàng)建和任務處理的包

package test_delivery

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/hibiken/asynq"
	"log"
)

const (
	TypeEmailDelivery = "email:deliver"
)

// EmailDeliveryPayload 異步任務需要傳遞的數(shù)據(jù)結(jié)構(gòu)
type EmailDeliveryPayload struct {
	UserID     int
	TemplateID string
	DataStr    string
}

// NewEmailDeliveryTask 異步任務需要傳遞的數(shù)據(jù)
func NewEmailDeliveryTask(userID int, tmplID, dataStr string) (*asynq.Task, error) {
	payload, err := json.Marshal(EmailDeliveryPayload{UserID: userID, TemplateID: tmplID, DataStr: dataStr})
	if err != nil {
		fmt.Println(err)
		return nil, err
	}
	return asynq.NewTask(TypeEmailDelivery, payload), nil
}

// HandleEmailDeliveryTask 發(fā)送email處理邏輯
func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
	//接收任務數(shù)據(jù)
	var p EmailDeliveryPayload
	if err := json.Unmarshal(t.Payload(), &p); err != nil {
		return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
	}
	//邏輯處理start...
	log.Printf("Sending Email to User: user_id=%d, template_id=%s data_str:%s", p.UserID, p.TemplateID, p.DataStr)
	return nil
}

client.go

在應用程序代碼中,導入上述包并用于Client將任務放入隊列中。

package client

import (
	"asynq_task/test_delivery"
	"fmt"
	"github.com/hibiken/asynq"
	"log"
	"time"
)

func EmailDeliveryTaskAdd(i int) {
	client := asynq.NewClient(asynq.RedisClientOpt{
		Addr:     "192.168.0.120:6379",
		Password: "123456",
		DB:       2,
	})
	defer client.Close()

	// 初使貨需要傳遞的數(shù)據(jù)
	task, err := test_delivery.NewEmailDeliveryTask(42, fmt.Sprintf("some:template:id:%d", i), `{"name":"lisi"}`)
	if err != nil {
		log.Fatalf("could not create task: %v", err)
	}
	// 任務入隊
	//info, err := client.Enqueue(task)

	//info, err := client.Enqueue(task, time.Now())
	// 延遲執(zhí)行
	info, err := client.Enqueue(task, asynq.ProcessIn(3*time.Second))
	// MaxRetry 重度次數(shù) Timeout超時時間
	//info, err = client.Enqueue(task, asynq.MaxRetry(10), asynq.Timeout(3*time.Second))
	if err != nil {
		log.Fatalf("could not enqueue task: %v", err)
	}
	log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
}

main.go 異步任務服務入口文件

接下來,啟動一個工作服務器以在后臺處理這些任務。要啟動后臺工作人員,使用Server并提供您Handler來處理任務??梢赃x擇使用ServeMux來創(chuàng)建處理程序,就像使用net/httpHandler 一樣。

package main

import (
	"asynq_task/test_delivery"
	"github.com/hibiken/asynq"
	"log"
)

func main() {
	srv := asynq.NewServer(
		asynq.RedisClientOpt{
			Addr:     "192.168.0.120:6379",
			Password: "123456",
			DB:       2,
		},
		asynq.Config{
			// 每個進程并發(fā)執(zhí)行的worker數(shù)量
			Concurrency: 5,
			// Optionally specify multiple queues with different priority.
			Queues: map[string]int{
				"critical": 6,
				"default":  3,
				"low":      1,
			},
			// See the godoc for other configuration options
		},
	)

	mux := asynq.NewServeMux()
	mux.HandleFunc(test_delivery.TypeEmailDelivery, test_delivery.HandleEmailDeliveryTask)

	if err := srv.Run(mux); err != nil {
		log.Fatalf("could not run server: %v", err)
	}
}

4)test.go 用來分發(fā)異步任務

package main

import (
	"asynq_task/test_delivery/client"
	"time"
)

func main() {
	for i := 0; i < 3; i++ {
		client.EmailDeliveryTaskAdd(i)
		time.Sleep(time.Second * 3)
	}
}

6. 運行查看結(jié)果

首先,我們要先把異步任務啟動起來準備好接收,也就是啟動cmd/main.go

啟動test.go文件向異步任務服務添加任務隊列

結(jié)果如下:

消息者 go run main.go

$ go run main.go 
asynq: pid=12092 2023/02/02 23:18:04.161872 INFO: Starting processing
asynq: pid=12092 2023/02/02 23:18:04.161872 INFO: Send signal TERM or INT to terminate the process
2023/02/03 07:18:14 Sending Email to User: user_id=42, template_id=some:template:id:0 data_str:{"name":"lisi"}
2023/02/03 07:18:19 Sending Email to User: user_id=42, template_id=some:template:id:1 data_str:{"name":"lisi"}
2023/02/03 07:18:19 Sending Email to User: user_id=42, template_id=some:template:id:2 data_str:{"name":"lisi"}

生產(chǎn)者 go run test.go

$ go run test.go 
2023/02/03 07:18:09 enqueued task: id=5d998c6b-3978-4a25-a096-6e564e032359 queue=default
2023/02/03 07:18:12 enqueued task: id=74a5fea4-d4d4-465f-b310-31981e472f6a queue=default
2023/02/03 07:18:15 enqueued task: id=41c46b7b-ea78-4abc-878a-ea65e3859e28 queue=default

三、細節(jié)

1. 關(guān)于asynq的優(yōu)雅退出

如果異步服務突然被暫停,正在執(zhí)行的異步任務會push到隊列中,下次啟動的時候自動執(zhí)行。

我們可以將一個異步任務中途sleep幾秒,發(fā)送一個異步任務,任務沒執(zhí)行完中途停掉任務測試出結(jié)果:

再次啟動異步任務服務,發(fā)現(xiàn)這個任務被重新執(zhí)行。

2. client中 client.Enqueue 的使用

立即處理任務

client.Enqueue(t1, time.Now())

2)延時處理任務, 兩小時后處理

client.Enqueue(t2, asynq.ProcessIn(time.Now().Add(2 * time.Hour)))

任務重試,最大重試次數(shù)為25次。

client.Enqueue(task, asynq.MaxRetry(5))

4)確保任務的唯一性

4-1:使用TaskID選項:自行生成唯一的任務 ID

_, err := client.Enqueue(task, asynq.TaskID("mytaskid"))

// Second task will fail, err is ErrTaskIDConflict (assuming that the first task didn't get processed yet)
_, err = client.Enqueue(task, asynq.TaskID("mytaskid"))

4-2:使用Unique選項:讓 Asynq 為任務創(chuàng)建唯一性鎖

err := c.Enqueue(t1, asynq.Unique(time.Hour))

另外,asynq 異步任務提供了命令行工具和 Asynqmon 用于監(jiān)控和管理 Asynq 異步任務和隊列。WebUI 可以通過傳遞兩個標志來啟用與 Prometheus 的集成。

  #asynqmon asynq延遲隊列、定時隊列的webui
  asynqmon:
    image: hibiken/asynqmon:latest
    container_name: asynqmon
    ports:
      - 8980:8080
    command:
      - '--redis-addr=192.168.0.120:6379'
      - '--redis-password=123456'
      - '--redis-db=2'
    restart: always
#    networks:
#      - looklook_net
#    depends_on:
#      - redis

啟動服務:

docker-compose up

訪問:

http://192.168.0.120:8980/

總結(jié)

到此這篇關(guān)于Go異步任務解決方案之Asynq庫的文章就介紹到這了,更多相關(guān)Go異步任務Asynq庫內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!

相關(guān)文章

最新評論