日韩成人免费在线_国产成人一二_精品国产免费人成电影在线观..._日本一区二区三区久久久久久久久不

當前位置:首頁 > 科技  > 軟件

Go異步任務解決方案:Asynq

來源: 責編: 時間:2023-11-03 17:07:03 281觀看
導讀一、概述Asynq 是一個 Go 庫,用于對任務進行排隊并與工作人員異步處理它們。Asynq 工作原理的高級概述:客戶端將任務放入隊列服務器從隊列中拉出任務并為每個任務啟動一個工作 goroutine多個工作人員同時處理任務git庫

一、概述

Asynq 是一個 Go 庫,用于對任務進行排隊并與工作人員異步處理它們。2qM28資訊網——每日最新資訊28at.com

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

二、快速開始

1. 準備工作

  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:      # 時區上海      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 6379timeout 0# 900s內至少一次寫操作則執行bgsave進行RDB持久化save 900 1 save 300 10save 60 10000# rdbcompression ;默認值是yes。對于存儲到磁盤中的快照,可以設置是否進行壓縮存儲。如果是的話,redis會采用LZF算法進行壓縮。如果你不想消耗CPU來進行壓縮的話,可以設置為關閉此功能,但是存儲在磁盤上的快照會比較大。rdbcompression yes# dbfilename :設置快照的文件名,默認是 dump.rdbdbfilename 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

目錄結構:

.|-- README.md|-- cmd|   `-- main.go  # 啟動消息者監聽|-- go.mod|-- go.sum|-- test.go # 生成者 發送測試數據`-- test_delivery    |-- client # 具體發送測試數據的邏輯    |   `-- client.go    `-- test_delivery.go  # 消息者,執行任務具體處理邏輯

2. Redis連接項

Asynq 使用 Redis 作為消息代理。client.go 和 main.go 都需要連接到 Redis 進行寫入和讀取。我們將使用 asynq.RedisClientOpt 指定如何連接到本地 Redis 實例。2qM28資訊網——每日最新資訊28at.com

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

4. Task任務

*asynq.Task

type Task struct { // 一個簡單的字符串值,表示要執行的任務的類型. typename string // 有效載荷保存執行任務所需的數據,有效負載值必須是可序列化的. payload []byte // 保存任務的選項. opts []Option // 任務的結果編寫器. w *ResultWriter}

5. 編寫程序

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

package test_deliveryimport ( "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將任務放入隊列中。2qM28資訊網——每日最新資訊28at.com

package clientimport ( "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 一樣。2qM28資訊網——每日最新資訊28at.com

package mainimport ( "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 用來分發異步任務

package mainimport ( "asynq_task/test_delivery/client" "time")func main() { for i := 0; i < 3; i++ {  client.EmailDeliveryTaskAdd(i)  time.Sleep(time.Second * 3) }}

6. 運行查看結果

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

結果如下:2qM28資訊網——每日最新資訊28at.com

消息者 go run main.go2qM28資訊網——每日最新資訊28at.com

$ go run main.go asynq: pid=12092 2023/02/02 23:18:04.161872 INFO: Starting processingasynq: pid=12092 2023/02/02 23:18:04.161872 INFO: Send signal TERM or INT to terminate the process2023/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.go2qM28資訊網——每日最新資訊28at.com

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

三、細節

1. 關于asynq的優雅退出

如果異步服務突然被暫停,正在執行的異步任務會push到隊列中,下次啟動的時候自動執行。2qM28資訊網——每日最新資訊28at.com

我們可以將一個異步任務中途sleep幾秒,發送一個異步任務,任務沒執行完中途停掉任務測試出結果:2qM28資訊網——每日最新資訊28at.com

圖片圖片2qM28資訊網——每日最新資訊28at.com

再次啟動異步任務服務,發現這個任務被重新執行。2qM28資訊網——每日最新資訊28at.com

2. client中 client.Enqueue 的使用

1)立即處理任務

client.Enqueue(t1, time.Now())2qM28資訊網——每日最新資訊28at.com

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

client.Enqueue(t2, asynq.ProcessIn(time.Now().Add(2 * time.Hour)))2qM28資訊網——每日最新資訊28at.com

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

client.Enqueue(task, asynq.MaxRetry(5))2qM28資訊網——每日最新資訊28at.com

4)確保任務的唯一性

4-1:使用TaskID選項:自行生成唯一的任務 ID2qM28資訊網——每日最新資訊28at.com

_, 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 為任務創建唯一性鎖2qM28資訊網——每日最新資訊28at.com

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

另外,asynq 異步任務提供了命令行工具和 Asynqmon 用于監控和管理 Asynq 異步任務和隊列。WebUI 可以通過傳遞兩個標志來啟用與 Prometheus 的集成。2qM28資訊網——每日最新資訊28at.com

#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 up2qM28資訊網——每日最新資訊28at.com

訪問:http://192.168.0.120:8980/2qM28資訊網——每日最新資訊28at.com

圖片圖片2qM28資訊網——每日最新資訊28at.com

原文鏈接:http://haimait.top/docs/golang/asynq2qM28資訊網——每日最新資訊28at.com

本文轉載自微信公眾號「 程序員升級打怪之旅」,作者「 海馬&王中陽Go」,可以通過以下二維碼關注。2qM28資訊網——每日最新資訊28at.com

2qM28資訊網——每日最新資訊28at.com

轉載本文請聯系「 程序員升級打怪之旅」公眾號。2qM28資訊網——每日最新資訊28at.com

本文鏈接:http://www.www897cc.com/showinfo-26-16838-0.htmlGo異步任務解決方案:Asynq

聲明:本網頁內容旨在傳播知識,若有侵權等問題請及時與本網聯系,我們將在第一時間刪除處理。郵件:2376512515@qq.com

上一篇: 聊聊分布式服務下的八種異步實現方式

下一篇: 玩轉 Go Slices 切片泛型庫

標簽:
  • 熱門焦點
Top 主站蜘蛛池模板: 桓台县| 大英县| 横峰县| 慈利县| 松潘县| 乌拉特前旗| 留坝县| 岚皋县| 富宁县| 柳江县| 安阳市| 诸暨市| 连山| 西畴县| 肥乡县| 乐都县| 高密市| 九台市| 镇远县| 颍上县| 福海县| 仁化县| 琼中| 新和县| 东港市| 乐至县| 广宗县| 崇明县| 潜江市| 保靖县| 玉门市| 张掖市| 阜新| 乌兰县| 安阳市| 庄河市| 敦煌市| 呼玛县| 准格尔旗| 玛沁县| 宁晋县|