91精品欧美综合在线,亚洲成a∨人在线播放欧美

Go異步任務(wù)解決方案：Asynq

一、概述
Asynq 是一個(gè) Go 庫，用于對任務(wù)進(jìn)行排隊(duì)并與工作人員異步處理它們。

Asynq 工作原理的高級概述：
客戶端將任務(wù)放入隊(duì)列
服務(wù)器從隊(duì)列中拉出任務(wù)并為每個(gè)任務(wù)啟動(dòng)一個(gè)工作 goroutine
多個(gè)工作人員同時(shí)處理任務(wù)
git庫地址：
https://github.com/hibiken/asynq

二、快速開始
1. 準(zhǔn)備工作
確保已安裝并運(yùn)行了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:
      # 時(shí)區(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

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

docker-compose up -d

2. 安裝asynq軟件包
go get -u github.com/hibiken/asynq

3. 創(chuàng)建項(xiàng)目asynq_task
目錄結(jié)構(gòu)：

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

2. Redis連接項(xiàng)
Asynq 使用 Redis 作為消息代理。client.go 和 main.go 都需要連接到 Redis 進(jìn)行寫入和讀取。我們將使用 asynq.RedisClientOpt 指定如何連接到本地 Redis 實(shí)例。

asynq.RedisClientOpt{
Addr:     "127.0.0.1:6379",
Password: "",
DB:       2,
}
4. Task任務(wù)
*asynq.Task

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

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

// 保存任務(wù)的選項(xiàng).
opts []Option

// 任務(wù)的結(jié)果編寫器.
w *ResultWriter
}
5. 編寫程序
1）test_delivery.go 一個(gè)封裝任務(wù)創(chuàng)建和任務(wù)處理的包

package test_delivery

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

const (
TypeEmailDelivery = "email:deliver"
)

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

// NewEmailDeliveryTask 異步任務(wù)需要傳遞的數(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 {
//接收任務(wù)數(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
在應(yīng)用程序代碼中，導(dǎo)入上述包并用于Client將任務(wù)放入隊(duì)列中。

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)
}
// 任務(wù)入隊(duì)
//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超時(shí)時(shí)間
//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 異步任務(wù)服務(wù)入口文件
接下來，啟動(dòng)一個(gè)工作服務(wù)器以在后臺處理這些任務(wù)。要啟動(dòng)后臺工作人員，使用Server并提供您Handler來處理任務(wù)。可以選擇使用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{
   // 每個(gè)進(jìn)程并發(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ā)異步任務(wù)

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. 運(yùn)行查看結(jié)果
首先，我們要先把異步任務(wù)啟動(dòng)起來準(zhǔn)備好接收，也就是啟動(dòng)cmd/main.go
啟動(dòng)test.go文件向異步任務(wù)服務(wù)添加任務(wù)隊(duì)列
結(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
三、細(xì)節(jié)
1. 關(guān)于asynq的優(yōu)雅退出
如果異步服務(wù)突然被暫停，正在執(zhí)行的異步任務(wù)會push到隊(duì)列中，下次啟動(dòng)的時(shí)候自動(dòng)執(zhí)行。

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

再次啟動(dòng)異步任務(wù)服務(wù)，發(fā)現(xiàn)這個(gè)任務(wù)被重新執(zhí)行。

2. client中 client.Enqueue 的使用
立即處理任務(wù)
client.Enqueue(t1, time.Now())

2）延時(shí)處理任務(wù)，兩小時(shí)后處理

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

任務(wù)重試，最大重試次數(shù)為25次。
client.Enqueue(task, asynq.MaxRetry(5))

4）確保任務(wù)的唯一性

4-1：使用TaskID選項(xiàng)：自行生成唯一的任務(wù) 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選項(xiàng)：讓 Asynq 為任務(wù)創(chuàng)建唯一性鎖

err := c.Enqueue(t1, asynq.Unique(time.Hour))
另外，asynq 異步任務(wù)提供了命令行工具和 Asynqmon 用于監(jiān)控和管理 Asynq 異步任務(wù)和隊(duì)列。WebUI 可以通過傳遞兩個(gè)標(biāo)志來啟用與 Prometheus 的集成。

#asynqmon asynq延遲隊(duì)列、定時(shí)隊(duì)列的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
啟動(dòng)服務(wù)：

docker-compose up

訪問：

http://192.168.0.120:8980/

原文鏈接：http://haimait.top/docs/golang/asynq

請前往：http://lygongshang.com/TeacherV2.html?id=365

在线午夜精品自拍小视频_无码av无码专区线_亚洲无码精品人妻_人人澡欧美一区

Go語言