Go asynq 异步任务队列

asynq 是 Go 实现的一种简单、可靠、高效的分布式任务队列

介绍

• Asynq 是一个 Go 库，用于对任务进行排队并与工作线程异步处理它们
  • 它使用 Redis 作为队列
• Asynq 工作原理
  • 客户端将任务放入队列
  • 服务器从队列中取出任务并为每个任务启动一个工作协程
  • 任务由多个 worker 同时处理
  • 任务队列是一种在多台机器上分配工作的机制。一个系统可以由多个工作服务器和代理组成，从而实现高可用性和横向扩展
• 特性包括
  • 保证至少执行一次任务
  • 任务调度
  • 重试失败的任务
  • 在工作线程崩溃时自动恢复任务
  • 加权优先级队列
  • 严格优先级队列
  • 由于 Redis 中的写操作快速，添加任务的延迟低
  • 使用唯一选项对任务进行去重
  • 允许为每个任务设置超时和截止时间
  • 允许聚合一组任务以批量执行多个连续操作
  • 灵活的处理程序接口，支持中间件
  • 允许暂停队列以停止从队列中处理任务
  • 周期性任务
  • 支持 Redis Cluster 以自动分片和实现高可用性
  • 支持 Redis Sentinel 实现高可用性
  • 与 Prometheus 集成，以收集和可视化队列指标
  • Web 界面，用于检查和远程控制队列和任务
  • 命令行界面，用于检查和远程控制队列和任务

使用

• 支持中间件

server

source repo / download raw

client

source repo / download raw

task

source repo / download raw

• Task Lifecycle：
  • Scheduled : task is waiting to be processed in the future (Only applies to tasks with ProcessAt or ProcessIn option).
  • Pending : task is ready to be processed and will be picked up by a free worker.
  • Active : task is being processed by a worker (i.e. handler is invoked with the task).
  • Retry : worker failed to process the task and the task is waiting to be retried in the future.
  • Archived : task reached its max retry and stored in an archive for manual inspection.
  • Completed: task was successfully processed and retained until retention TTL expires (Only applies to tasks with Retention option).
• Task Timeout and Cancelation
• Task Retention and Result

其他

• Inspector 是一个客户端接口，用于检查和改变队列和任务的状态
• scheduler 定时任务、周期性任务，类似于 crontab 的作用
• ErrorHandler 用来捕获错误事件，解决 task lease expired 错误，参考
• RetryDelayFunc 根据不同的 typename 设定 retry 时间
• SkipRetry

监控

Web UI

• https://github.com/hibiken/asynqmon

docker run --rm \
    --name asynqmon \
    -p 8080:8080 \
    hibiken/asynqmon

使用 lib 库集成到项目中

• Import as a Library，gin 参考

Client

go install github.com/hibiken/asynq/tools/asynq@latest

asynq --help

asynq ...