highlight: a11y-dark theme: Chinese-red

今天為大家介紹一個Go處理異步任務的解決方案：Asynq，是一個 Go 庫，用於排隊任務並與 worker 異步處理它們。它由Redis提供支持，旨在實現可擴展且易於上手。

一、概述

Asynq 是一個 Go 庫，用於對任務進行排隊並與工作人員異步處理它們。 1. Asynq 工作原理的高級概述： - 客户端將任務放入隊列 - 服務器從隊列中拉出任務併為每個任務啟動一個工作 goroutine - 多個工作人員同時處理任務 2. git庫地址：

http://github.com/hibiken/asynq

二、快速開始

1. 準備工作

確保已安裝並運行了redis redis 版本大於5.0

redis-server

目錄結構

shell . ├── conf │ └── redis.conf └── docker-compose.yml docker-compose.yml ```yaml version: '3.8' services: myredis: container_name: myredis image: redis:6.2.5 #6.0.6 restart: always ports: - 6379:6379 privileged: true environment: # 時區上海 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 ```yaml

開啟保護

protected-mode yes

開啟遠程連接

bind 127.0.0.1

自定義密碼

requirepass 123456 port 6379 timeout 0

900s內至少一次寫操作則執行bgsave進行RDB持久化

save 900 1 save 300 10 save 60 10000

rdbcompression ；默認值是yes。對於存儲到磁盤中的快照，可以設置是否進行壓縮存儲。如果是的話，redis會採用LZF算法進行壓縮。如果你不想消耗CPU來進行壓縮的話，可以設置為關閉此功能，但是存儲在磁盤上的快照會比較大。

rdbcompression yes

dbfilename ：設置快照的文件名，默認是 dump.rdb

dbfilename dump.rdb

dir：設置快照文件的存放路徑，這個配置項一定是個目錄，而不能是文件名。使用上面的 dbfilename 作為保存的文件名。

dir /data

默認redis使用的是rdb方式持久化，這種方式在許多應用中已經足夠用了。但是redis如果中途宕機，會導致可能有幾分鐘的數據丟失，根據save來策略進行持久化，Append Only File是另一種持久化方式，可以提供更好的持久化特性。Redis會把每次寫入的數據在接收後都寫入appendonly.aof文件，每次啟動時Redis都會先把這個文件的數據讀入內存裏，先忽略RDB文件。默認值為no。

appendonly yes

appendfilename ：aof文件名，默認是"appendonly.aof"

appendfsync：aof持久化策略的配置；no表示不執行fsync，由操作系統保證數據同步到磁盤，速度最快；always表示每次寫入都執行fsync，以保證數據同步到磁盤；everysec表示每秒執行一次fsync，可能會導致丟失這1s數據

appendfsync everysec `` 啟動redis` 服務

docker-compose up -d

2. 安裝asynq軟件包

go get -u github.com/hibiken/asynq

3. 創建項目asynq_task

```

2. Redis連接項

Asynq 使用 Redis 作為消息代理。 client.go 和 main.go 都需要連接到 Redis 進行寫入和讀取。我們將使用 asynq.RedisClientOpt 指定如何連接到本地 Redis 實例。

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

4. Task任務

*asynq.Task ```go type Task struct { // 一個簡單的字符串值，表示要執行的任務的類型. typename string

// 有效載荷保存執行任務所需的數據，有效負載值必須是可序列化的.
payload []byte

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

// 任務的結果編寫器.
w *ResultWriter

} ```

5. 編寫程序

1）test_delivery.go 一個封裝任務創建和任務處理的包 ```go package test_delivery

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

const ( TypeEmailDelivery = "email:deliver" )

// EmailDeliveryPayload 異步任務需要傳遞的數據結構 type EmailDeliveryPayload struct { UserID int TemplateID string DataStr string }

// NewEmailDeliveryTask 異步任務需要傳遞的數據 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 發送email處理邏輯 func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error { //接收任務數據 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 }

```

2) client.go

在應用程序代碼中，導入上述包並用於Client將任務放入隊列中。 ```go 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()

// 初使貨需要傳遞的數據
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())
// 延遲執行
info, err := client.Enqueue(task, asynq.ProcessIn(3*time.Second))
// MaxRetry 重度次數 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)

}

```

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

接下來，啟動一個工作服務器以在後台處理這些任務。要啟動後台工作人員，使用Server並提供您Handler來處理任務。可以選擇使用ServeMux來創建處理程序，就像使用net/httpHandler 一樣。 ```go 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{ // 每個進程併發執行的worker數量 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 用來分發異步任務 ```go 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. 運行查看結果

首先，我們要先把異步任務啟動起來準備好接收，也就是啟動cmd/main.go
啟動test.go文件向異步任務服務添加任務隊列

結果如下：

消息者 go run main.go shell $ 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"}

生產者 go run test.go shell $ 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

三、細節

1. 關於asynq的優雅退出

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

我們可以將一個異步任務中途sleep幾秒，發送一個異步任務，任務沒執行完中途停掉任務測試出結果：

再次啟動異步任務服務，發現這個任務被重新執行。

2. client中 client.Enqueue 的使用

1) 立即處理任務

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

2）延時處理任務，兩小時後處理

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

3) 任務重試，最大重試次數為25次。

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

4）確保任務的唯一性

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

```go _, 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 為任務創建唯一性鎖

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

另外，asynq 異步任務提供了命令行工具和 Asynqmon 用於監控和管理 Asynq 異步任務和隊列。WebUI 可以通過傳遞兩個標誌來啟用與 Prometheus 的集成。

```yaml #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/

原文鏈接

http://haimait.top/docs/golang/asynq

一起進步

歡迎和我一起學習進步：可以在掘金私信我

也可以加我微信號：wangzhongyang1993 公眾號：程序員升職加薪之旅 B站視頻：王中陽Go

Go異步任務處理解決方案：Asynq

一、概述

二、快速開始

1. 準備工作

networks:

開啟保護

開啟遠程連接

bind 127.0.0.1

自定義密碼

900s內至少一次寫操作則執行bgsave進行RDB持久化

rdbcompression ；默認值是yes。對於存儲到磁盤中的快照，可以設置是否進行壓縮存儲。如果是的話，redis會採用LZF算法進行壓縮。如果你不想消耗CPU來進行壓縮的話，可以設置為關閉此功能，但是存儲在磁盤上的快照會比較大。

dbfilename ：設置快照的文件名，默認是 dump.rdb

dir：設置快照文件的存放路徑，這個配置項一定是個目錄，而不能是文件名。使用上面的 dbfilename 作為保存的文件名。

appendfilename ：aof文件名，默認是"appendonly.aof"

appendfsync：aof持久化策略的配置；no表示不執行fsync，由操作系統保證數據同步到磁盤，速度最快；always表示每次寫入都執行fsync，以保證數據同步到磁盤；everysec表示每秒執行一次fsync，可能會導致丟失這1s數據

2. 安裝asynq軟件包

3. 創建項目asynq_task

2. Redis連接項

4. Task任務

5. 編寫程序

6. 運行查看結果

三、細節

1. 關於asynq的優雅退出

2. client中 client.Enqueue 的使用

networks:

- looklook_net

depends_on:

- redis

一起進步