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

更新時間：2023年02月15日 12:38:19 作者：王中陽Go

需要在Go應用程序中異步處理任務? Asynq,簡單高效的任務隊列實現(xiàn),下面這篇文章主要給大家介紹了關(guān)于Go異步任務解決方案之Asynq庫的相關(guā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.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

	// 任務的結(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)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

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

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

目錄

一、概述

二、快速開始

1. 準備工作

2. 安裝asynq軟件包

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

4. Task任務

5. 編寫程序

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

三、細節(jié)

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

2. client中 client.Enqueue 的使用

總結(jié)

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

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

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

目錄

一、概述

二、快速開始

1. 準備工作

2. 安裝asynq軟件包

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

4. Task任務

5. 編寫程序

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

三、細節(jié)

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

2. client中 client.Enqueue 的使用

總結(jié)

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

三、細節(jié)