Go異步任務解決方案之Asynq庫詳解
今天為大家介紹一個Go處理異步任務的解決方案：Asynq，是一個 Go 庫，用于排隊任務并與 worker 異步處理它們。它由Redis提供支持，旨在實現(xiàn)可擴展且易于上手。
一、概述
Asynq 是一個 Go 庫，用于對任務進行排隊并與工作人員異步處理它們。
- Asynq 工作原理的高級概述：
- 客戶端將任務放入隊列
- 服務器從隊列中拉出任務并為每個任務啟動一個工作 goroutine
- 多個工作人員同時處理任務
- git庫地址：
https://github.com/hibiken/asynq
二、快速開始
1. 準備工作
確保已安裝并運行了redis
redis 版本大于5.0
redis-server
目錄結構
.
├── 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: bridgeconf/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來進行壓縮的話，可以設置為關閉此功能，但是存儲在磁盤上的快照會比較大。 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
目錄結構：
.
|-- 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.go 和 main.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
// 任務的結果編寫器.
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ù)結構
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. 運行查看結果
首先，我們要先把異步任務啟動起來準備好接收，也就是啟動cmd/main.go
啟動test.go文件向異步任務服務添加任務隊列
結果如下：
消息者 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. 關于asynq的優(yōu)雅退出
如果異步服務突然被暫停，正在執(zhí)行的異步任務會push到隊列中，下次啟動的時候自動執(zhí)行。
我們可以將一個異步任務中途sleep幾秒，發(fā)送一個異步任務，任務沒執(zhí)行完中途停掉任務測試出結果：
再次啟動異步任務服務，發(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/
總結
到此這篇關于Go異步任務解決方案之Asynq庫的文章就介紹到這了,更多相關Go異步任務Asynq庫內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
這篇文章主要介紹了go語言實現(xiàn)簡單http服務的方法,涉及Go語言http操作技巧,具有一定參考借鑒價值,需要的朋友可以參考下2015-02-02
GO中?分組聲明與array,?slice,?map函數(shù)
這篇文章主要介紹了GO中?分組聲明與array,slice,map函數(shù)，Go語言中，同時聲明多個常量、變量，或者導入多個包時，可采用分組的方式進行聲明，下面詳細介紹需要的小伙伴可以參考一下2022-03-03
這篇文章主要為大家介紹了自動生成代碼controller?tool的簡單使用示例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-05-05
這篇文章主要介紹了go語言按顯示長度截取字符串的方法,涉及Go語言操作字符串的技巧,具有一定參考借鑒價值,需要的朋友可以參考下2015-02-02
Go 簡單實現(xiàn)多租戶數(shù)據(jù)庫隔離
本文主要介紹了Go 簡單實現(xiàn)多租戶數(shù)據(jù)庫隔離，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-05-05
golang通過context控制并發(fā)的應用場景實現(xiàn)
這篇文章主要介紹了golang通過context控制并發(fā)的應用場景實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-01-01
這篇文章主要介紹了Go語言展現(xiàn)快速排序算法全過程的思路及代碼示例,文章最后作者還提到了對Quick Sort算法優(yōu)化的一些想法,需要的朋友可以參考下2016-04-04
在Go語言中，我們可以使用for、append()和copy()進行數(shù)組拷貝。這篇文章主要為大家詳細介紹一下這三種方式的具體實現(xiàn)與性能分析，需要的可以參考一下2022-08-08
這篇文章主要和大家分享9個Golang中map可能是常用的使用案例，從1到9，越來越厲害。文中的示例代碼講解詳細，希望對大家學習Golang有一定的幫助2023-02-02