phptaskqueue技巧_切切级延时责任队列若何实现看美图开源的LMSTFY

lmstfy(Let Me Schedule Task For You) 是美图架构根本做事团队在 2018 年初基于 Redis 实现的大略任务行列步队(Task Queue)做事，目前在美图多个线上产品利用靠近两年的韶光。
紧张供应以下特性:

任务具备延时、自动重试、优先级以及过期等功能

通过 HTTP restful API 供应做事

具备横向扩展能力

丰富的业务和性能指标

Github 项目地址: https://github.com/meitu/lmstfy

利用场景

任务行列步队跟行列步队在利用场景上最大的差异是： 任务之间是没有顺序约束而哀求顺序(FIFO)，且可能会对任务的状态更新而一样平常只会消费不会更新。
类似 Kafka 利用 FIFO 和不须要更新(不须要对做索引)的特性来设计存储，将读写变成磁盘的顺序读写来实现比较好的性能。
而任务行列步队须要能够任务状态进行更新则须要对每个进行索引，如果把两者放到一起实现则很难实现在功能和性能上兼得。
在美图内部选型上，如果是异步模型一样平常会选择行列步队，比如类似日志上报，抢购等。
而对付须要延时/定时下发或者修正状态任务则是利用任务行列步队。

比如在以下几种场景会利用任务行列步队:

定时任务，如每天早上 8 点开始推送，定期删除过期数据等

任务流，如自动创建 Redis 流程由资源创建，资源配置，DNS 修正等部分组成，利用任务行列步队可以简化整体的设计和重试流程

重试任务，范例场景如离线图片处理

目标与调研

在自研任务行列步队之前，我们基于以下几个哀求作为约束调研了现有一些开源方案:

任务支持延时/优先级任务和自动重试

高可用，做事不能有单点以及担保数据不丢失

可扩展，紧张是容量和性能须要可扩展

第一种方案是 Redis 作者开源的分布式内存行列步队 disque(https://github.com/antirez/disque)。
disque 采取和 Redis Cluster 类似无中央设计，所有节点都可以写入并复制到其他节点。
不管是从功能上、设计还是可靠性都是比较好的选择。
我们在 2017 年也引入 disque 在部分业务利用过一段韶光，后面碰着 bug 在内部修复后想反馈到社区，创造 Redis 作者决定不再掩护这个项目(要把 disque 功能作为 redis module 来掩护，该当是会伴随 Redis 6 发布)。
终极我们也放弃了 disque 方案，将数据迁移到我们自研任务行列步队做事。

第二种方案是 2007 年就开源的 beanstalkd(https://github.com/beanstalkd/beanstalkd)，现在仍旧还是在掩护状态。
beanstalkd 是类 memcached 协议全内存任务行列步队，断电或者重启时通过 WAL 文件来规复数据。
但 benstalkd 不支持复制功能，做事存在单点问题且数据可靠性也无法知足。
当时也有考虑基于 beanstalkd 去做二次开拓，但看完代码之后以为须要改造的点不但是复制，还有类似内存掌握等等，以是没有选择 beanstalkd 二次开拓的方案。

也考虑过类似基于 kafka/rocketmq 等行列步队作为存储的方案，末了从存储设计模型和团队技能栈等缘故原由决定选择基于 redis 作为存储来实现任务行列步队的功能。
举个例子，假设以 Kafka 这种行列步队存储来实现延时功能，每个行列步队的韶光都须要创建一个单独的 topic(如: Q1-1s, Q1-2s..)。
这种设计在延时时间比较固定的场景下问题不太大，但如果是延时时间变革比较大会导致 topic 数目过多，会把磁盘从顺序读写会变成随机读写从导致性能衰减，同时也会带来其他类似重启或者规复韶光过长的问题。

设计和实现

lmstfy 是 HTTP 协议的无状态做事，可以通过 4/L7 的 LB 来接入。
内部紧张由四个模块组成:

Pump Thread: 每秒轮询 Redis 将到期的任务迁移到就绪行列步队(ready queue)

Metric Collector, 定时网络行列步队干系统计数据到实例再通过 prometheus exporter 暴露给监控系统

Token Manager，用来管理 namespace 和 token 的模块，namespace 是用来做业务隔离的单位

Producer/Consumer，用来处理用户的任务和消费要求

Default Pool 除了用来存储业务数据，namespace/token 这类元数据也会默认存储到 Default 这个 Redis 池子里面

namespace - 用来隔离业务，每个业务是独立的 namespace

queue - 行列步队名称，用区分同一业务不同类型

job - 业务定义的业务，紧张包含以下几个属性:

id: 任务 ID，全局唯一

delay: 任务延时下发韶光， 单位是秒

tries: 任务最大重试次数，tries = N 表示任务会最多下发 N 次

ttl(time to live): 任务最长有效期，超过之后任务自动消逝

ttr(time to run): 任务预期实行韶光，超过 ttr 则认为任务消费失落败，触发任务自动重试

数据存储

lmstfy 的 redis 存储由四部分组成:

timer(sorted set) - 用来实现延迟任务的排序，再由后台线程定期将到期的任务写入到 Ready Queue 里面

ready queue (list) - 无延时或者已到期任务的行列步队

deadletter (list) - 消费失落败(重试次数到达上限)的任务，可以手动重新放回行列步队

job pool(string) - 存储内容的池子

支持延迟的任务行列步队实质上是两个数据构造的结合: FIFO 和 sorted set。
sorted set 用来实现延时的部分，将任务按照到期韶光戳升序存储，然后定期将到期的任务迁移至 FIFO(ready queue)。
任务的详细内容只会存储一份在 job pool 里面，其他的像 ready queue，timer，deadletter 只是存储 job id，这样可以节省一些内存空间。

以下是整体设计:

任务写入

任务在写入时会先产生一个 job id，目前 job id (16bytes) 包含写入韶光戳、 随机数和延迟秒数， 然后写入 key 为 j:{namespace}/{queue}/{ID} 的任务到任务池 (pool) 里面。
之后根据延时时间来决定这个 job id 该当到 ready queue 还是 timer 里面:

delay = 0，表示不须要延时则直接写到 ready queue(list)

delay = n(n > 0)，表示须要延时，将延时加上当前系统韶光作为绝对韶光戳写到 timer(sorted set)

timer 的实现是利用 zset 根据绝对韶光戳进行排序，再由旁路线程定期轮询将到期的任务通过 redis lua script 来将数据原子地转移到 ready queue 里面。

任务消费

之条件到任务在消费失落败之后预期能够重试，以是必须知道什么时候可认为任务消费失落败？业务在消费时须要携带 ttr(time to run) 参数，用来表示业务预期任务最长实行韶光，如果在 ttr 韶光内没有收到业务主动回答 ACK 则会认为任务失落败(类似 tcp 的重传 timer)。

消费时从 ready queue 中 (B)RPOP 出任务的 job id，然后根据 job id 从 pool 中将任务内容发送给消费者。
同时对 tries 减1，根据消费的 ttr(time to run) 参数, 将任务放入 timer 中。
如果 tries 为零, 在 ttr 韶光到期后该 job id 会被放入 dead letter 行列步队中(表示任务实行失落败)。

同步任务模型

lmstfy 除了可以用来实现异步和延时任务模型之外，由于 namespace 下面的行列步队是动态创建且 job id 全局唯一，还可以用来实现同步任务模型 (producer 等到任务实行成功之后返回)。
大概如下:

producer 写入任务之后拿到 job id, 然后监听(consume)以 job id 为名的行列步队

consumer 消费任务成功后，写回答到同样以 job id 为名的行列步队中

producer 如果规定韶光内能读到回答则认为消费成功，等待超时则认为任务失落败

如何实现横向扩展

lmstfy 本身是无状态的做事可以很大略的实现横向扩展，这里的横向扩展紧张是存储(目前只支持 Redis)的横向扩展。
设计也比较大略，紧张通过通过 namespace 对应的 token 路由来实现， 比如我们当前配置两组 Redis 资源: default 和 meipai:

[Pool][Pool.default]Addr = \公众1.1.1.1:6379\公众[Pool.meipai]Addr = \"大众2.2.2.2:6389\"大众

在创建 namespace 时可以指定资源池，token 里面会携带资源池名字作为前缀。
比指定美拍资源池，那么 token 类似: meipai:01DT8EZ1N6XT ，后续在处理要求时就可以根据 token 里面携带的资源池名称来进行路由数据。
不过这种设计实现行列步队级别的扩展，如果单行列步队存储量超过 Redis 内存上限则须要其他手段来办理(后面会支持磁盘类型存储)。

如何利用

# 创建 namespace 和 token, 把稳这里利用管理端口$ ./scripts/token-cli -c -n test_ns -p default -D \"大众test ns apply by @hulk\"大众 127.0.0.1:7778{ \"大众token\公众: \"大众01DT9323JACNBQ9JESV80G0000\公众}# 写入内容为 value 的任务$ curl -XPUT -d \公众value\"大众 -i \"大众http://127.0.0.1:7777/api/test_ns/q1?tries=3&delay=1&token=01DT931XGSPKNB7E2XFKPY3ZPB\公众{\"大众job_id\"大众:\"大众01DT9323JACNBQ9JESV80G0000\"大众,\"大众msg\公众:\"大众published\"大众}# 消费任务$ curl -i \"大众http://127.0.0.1:7777/api/test_ns/q1?ttr=30&timeout=3&&token=01DT931XGSPKNB7E2XFKPY3ZPB\"大众{\公众data\"大众:\"大众value\"大众,\"大众elapsed_ms\"大众:272612,\"大众job_id\"大众:\"大众01DT9323JACNBQ9JESV80G0000\"大众,\"大众msg\公众:\"大众new job\公众,\"大众namespace\"大众:\公众test_ns\"大众,\公众queue\"大众:\"大众q1\公众,\"大众ttl\"大众:86127}# ACK 任务 id，表示消费成功不再重新下发改任务curl -i -XDELETE \公众http://127.0.0.1:7777/api/test_ns/q1/job/01DT9323JACNBQ9JESV80G0000?token=01DT931XGSPKNB7E2XFKPY3ZPB\公众

更详细 API 解释见项目 README，目前我们供应了 PHP/Golang 两种措辞 SDK，其他措辞可以直接基于 HTTP 库封装即可。

监控指标

lmstfy 任务行列步队的其余一个设计目标是供应足够多的监控指标，除了作为监控报警之外，也可以为类似 k8s 的 scheduler 供应反馈指标，以当前行列步队堆积情形辅导系统进行动态缩扩容。

业务指标:

生产速率

消费速率

延迟数量

堆积数量 (queue size)

失落败数量 (deadletter size)

任务从生产到被消费的韶光分布 (P50, P95 etc.)

性能干系指标:

生产接口延迟 (P95)

消费接口延迟 (P95)

并发连接数

未来操持

在我们当前的利用场景下, 一个 2G 的 redis 实例就能够支撑千万级旁边的延迟任务量。
但类似工具存储的生命周期管理(工具存储的 TTL)这种量大且延韶光长的场景，利用 Redis 存储本钱比较高。
后续会考虑基于本地文件或者以 kvrocks (自研的 SSD Redis KV) 作为存储，将数据落到磁盘。
kvrocks 目前也是开源状态，美图内部线上已经支配靠近 100 个实例，外部也有一些类似白山云等公司在利用，后面也会输出干系设计和实现文章。
欢迎大家去关注和利用，更加欢迎 issue 和 PR。

kvrocks Github 项目地址: https://github.com/meitu/kvrocks

lmsty 的 Github 项目地址: https://github.com/meitu/lmstfy

如有更多技能问题想要互换可以发邮件给我: hulk.website@gmail.com

本文由高可用架构约稿。
技能原创及架构实践文章，欢迎通过公众号菜单「联系我们」进行投稿。

高可用架构

改变互联网的构建办法