Go异步任务解决方案：Asynq-编程学习网

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:
      # 时区上海
      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内至少一次写操作则执行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

目录结构：

.
|-- 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 实例。

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_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将任务放入队列中。

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 一样。

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 用来分发异步任务

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"}

生产者 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

三、细节

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

_, 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 为任务创建唯一性锁

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

另外，asynq 异步任务提供了命令行工具和 Asynqmon 用于监控和管理 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/

图片

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

本文转载自微信公众号「程序员升级打怪之旅」，作者「海马&王中阳Go」，可以通过以下二维码关注。

转载本文请联系「程序员升级打怪之旅」公众号。

文章详情

Go异步任务解决方案：Asynq

二、快速开始

1. 准备工作

redis-server

目录结构

docker-compose.yml

conf/redis.conf

启动 redis 服务

docker-compose up -d

2. 安装asynq软件包

go get -u github.com/hibiken/asynq

3. 创建项目asynq_task

目录结构：

2. Redis连接项

4. Task任务

*asynq.Task

5. 编写程序

1）test_delivery.go 一个封装任务创建和任务处理的包

2）client.go

3）main.go 异步任务服务入口文件

4）test.go 用来分发异步任务

6. 运行查看结果

三、细节

1. 关于asynq的优雅退出

2. client中 client.Enqueue 的使用

1）立即处理任务

2）延时处理任务， 两小时后处理

3）任务重试，最大重试次数为25次。

4）确保任务的唯一性

软考中级精品资料免费领

相关文章

猜你喜欢

Go异步任务解决方案 Asynq

Go异步任务解决方案：Asynq

Go异步任务解决方案之Asynq库详解

JS ES6异步解决方案

JavaScript异步解决方案有哪些

总结5种JavaScript异步解决方案

如何解决Go语言中的并发任务的任务失败重试和异步任务处理问题？

golang异步任务队列问题怎么解决

Shell Path：ASP异步编程的终极解决方案？

Spring @Async无法实现异步的解决方案

JavaScript 异步编程指南 — 解决方案 Async/Await

怎么解决异步任务所导致的问题

JavaScript 异步编程指南 - 如何用异步任务解决递归栈溢出？

windows7任务栏不见了解决方案

PHP日志加载：异步编程的解决方案？

CompletableFuture异步编程中的异常处理陷阱与解决方案

异步编程和日志记录：GO语言有什么高效的解决方案？

ASP异步编程的常见问题及解决方案？

详解Android App中的AsyncTask异步任务执行方式

8 种 Python 定时任务的解决方案

2）延时处理任务，两小时后处理