目录
1. 背景
2. 方案选型
3. 系统设计
3.1 整体架构
3.2 系统优化
4. 总结和展望
在营销自动化业务场景中，经常存在一类事件发生后，延迟一段时间，再执行另一动作的需求。这需要我们在系统中引入延迟服务来管理这样的任务。随着我们的业务发展，系统对延迟服务的数据量和吞吐性能也不断提高，亟需一种高并发、高吞吐、大容量的延迟服务解决方案。
本文将介绍 Shopee Off-Platform Ads 延迟服务的技术架构和演进历程。首先，根据业务带来的技术挑战，我们分析了备选的技术选型，说明它们的优缺点和适用场景。然后，我们会介绍当前方案的整体架构，说明主要场景下的处理流程，以及分享实现过程中的调优经验。最后，我们将概述系统当前运行状况，并探讨系统未来的优化方向。
1. 背景
1.1 需求背景
Shopee Off-Platform Ads 团队为营销业务提供了快速搭建和管理营销流程的可视化平台，支持业务团队以拖放节点的方式配置营销策略和活动。
其中，延迟操作是营销策略的常见组成部分。例如，为了验证发放优惠券对用户购买率的提升，业务团队希望在发放优惠券的某段时间后，再去检查用户的购买行为。又比如，业务可能希望对加购物车、关注、浏览某个商品一段时间后的用户执行营销逻辑。
这样的延迟操作需要平台提供可靠、高性能的延迟服务来支持。
Fig1. 延迟服务应用场景举例
1.2 技术挑战
区别于其他的延迟服务系统，我们面临的延迟服务需求有以下几点特征：
- 延迟任务提交和到期处理的性能要求高，大促期间需要具备 600 万/分钟量级的处理能力；
- 延迟时间范围广，需要支持 1 分钟 ~ 30 天之间分钟粒度的任意延迟时长配置；
- 延迟时间配置灵活，需要达到分钟级别的精度。
上述的延迟服务需求细化到技术层面，主要带来了以下几点挑战：
- 高性能的延迟任务提交和到期处理能力，峰值 OPS 需要达到 600 万/分钟量级；
- 海量延迟任务的高效存储和管理，需要达到百亿级别的任务管理能力；
- 需要考虑系统的水平扩展、故障转移和恢复的能力；
- 需要考虑支持多个业务场景快速接入的能力。
2. 方案选型
延迟服务的应用场景比较广泛，针对不同量级、不同需求的场景，业界已经有较为丰富的实践。下面我们来分析不同方案选型在 CRM 业务场景下的优缺点和可行性。
2.1 基于 MySQL 的实现方案
基于 MySQL 实现延迟服务的思路十分简单直接。在提交延迟任务时，把延迟任务的到期时间作为索引写入；在检查到期任务时，用到期时间作为过滤条件读出符合条件的任务，并对已读出的任务做标记和删除。
MySQL 单库单表的性能显然不能满足我们的业务需求，作为扩展，我们可以引入分库分表机制来提高并发读写能力。但是，从原理上来说，B+ 树并不适合高并发的随机写入；同时，MySQL 也不支持数据的过期，处理到期任务会增加显著的性能开销。
综合上述考虑，基于 MySQL 的存储延迟任务的实现方案不适合我们的业务场景。
2.2 基于 Redis Zset 的实现方案
Redis Zset 同样支持按延迟任务的到期时间进行排序。在元素个数超过一定数量时，Zset 底层采用跳表实现，内存中的跳表写入和读取性能都比较高。进一步地，如果我们使用多个 Zset 来分担延迟任务，控制每个跳表的元素个数小于 8k，就能避免跳表性能随着元素个数增多而下滑。
在 CRM 的延迟服务场景下，需要存储百亿级别的延迟任务，每个延迟任务除了到期时间，还需要携带任务自身的参数，最终所需的存储空间可达百 TB。使用内存来存储这么大规模的数据无疑是一笔很大的开销。
因此，单纯使用内存中的 Zset 存储延迟任务的方案实现成本过高，不符合优化资源使用效率的要求。
2.3 基于延迟消息队列的实现方案
不少消息中间件提供了延迟消息的功能，例如 RocketMQ、Pulsar 和 Rabbit MQ。
其中，Pulsar 在延迟时间长和任务量较大时存在性能瓶颈，而 Rabbit MQ 本身性能较其他消息队列存在一定差距，因此暂不考虑。
Rocket MQ 只能支持配置好的 18 个延迟时间档位。虽然我们可以采用重复入队出队的方式，利用固定的 18 个档位拼凑出任意时长的延迟时间，但是这种方式导致了写放大，也就是一次延迟任务的提交和到期，对应了底层消息队列的多次读写，会导致磁盘 I/O 高和存储空间利用率低的问题，带来额外的开销。
综上，支持延迟任务的消息队列也不能很好地满足当前的需求。
2.4 基于层级时间轮算法的实现方案
时间轮算法是经典的延迟任务实现方案，层级时间轮的方案可以支持任意时间精度的延迟任务，并且时间复杂度可以控制在 O(n)。我们对实现层级时间轮算法也做了前期调研和实现。
考虑到需要管理百亿级别的延迟任务，我们决定采用消息中间件 Kafka 而不是内存来模拟时间轮中的队列。在完成初步开发后，我们在测试中发现了以下问题：
- 层级时间轮算法在降层时会出现显著的 I/O 性能尖峰，会对集群的性能造成严重的影响；
- 层级时间轮存在着写放大现象，会增加存储的开销。
另外，用 Kafka topic 作为队列实现的层级时间轮算法，需要对 Kafka 的批量消费、offset 提交做一系列的特殊处理，导致具体实现较为复杂，理解起来比较困难。
因此，基于 Kafka 实现的层级时间轮方案也不能满足我们的需求。
Fig2. 层级时间轮算法示意
2.5 其他方案
在选型的过程中，我们也调研了基于其他存储、大数据组件或者外部开源软件的方案，但是这些方案或实现复杂、或存在可靠性和持续维护的风险。限于篇幅的原因，在这里不再深入讨论。
3. 系统设计
3.1 整体架构
让我们再次考察 2.2 的 Redis Zset 方案，它的缺点是需要在内存中存储海量的延迟任务，由此带来硬件开销过大的问题。这个缺点是否可以规避呢？答案是肯定的。一个延迟任务可以分为三个部分，分别是：
- 到期时间；
- 唯一键；
- 任务的 payload（透传给业务的其他信息）。
其中，到期时间和唯一键的大小是可控的。如果用雪花算法生成分布式唯一键，那么两者均可以控制为 8 Byte。在 Redis 的 Zset 中，以执行时间为 score，以唯一键为 key，对于百亿级别的延迟任务，内存使用可以控制在百 GB 级别，这个开销是可以接受的。
至于任务的 payload，我们可以将其存储到磁盘型的存储中，在 Redis Zset 取出到期任务后，再根据唯一键查询出任务 payload，拼接成完整的延迟任务并返回业务端。
我们选择 HBase 作为任务 payload 的存储，主要理由如下：
- 读写性能较好：在合理设计 rowkey 和缓存配置的前提下，单机读写性能可达 5w+；
- 存储容量和读写性能可以水平扩展；
- 支持 scan 操作，在发生异常时方便回补数据；
- 支持表数据按 TTL 过期。
基于Redis Zset + HBase + 消息队列的选型，我们设计的延迟服务整体架构如下图所示：
Fig3. 延迟服务整体架构
图中的流程分为两个部分，分别是任务提交流程和任务到期流程。
3.1.1 延迟任务提交流程
延迟服务对外提供 RPC 服务。业务方通过调用 RPC 接口，向服务提交延迟任务。延迟服务收到任务后，首先将任务写入一个 Kafka topic 中，写入成功后向业务方返回。我们称这个 topic 为 delay input topic。
写入 delay input topic 的延迟任务消息分为三个部分，分别是 unique_id，exec_time 和 payload，对应到任务唯一 ID，延迟任务期望执行时间和延迟任务透传信息。其中，unique_id 在提交时由业务方指定或者通过雪花算法生成，用于唯一标识一个延迟任务。
消费 delay input topic 的客户端负责将延迟任务写入 HBase 和 Redis。在写入 HBase 时，会利用 HBase 客户端的批量写入机制来减少网络往返的开销。每个延迟任务在成功写入 HBase 后，才会执行后续的 Redis 操作。
Redis 中包含多个 Zset。每个延迟任务通过 hash 自己的唯一 ID 来找到对应 Zset。Zset 的数量跟系统同时维护的任务数量正相关。通过增加或减少 Zset 的总数，我们可以控制每个 Zset 中元素的个数，以保持单次 Zset 操作的高效。
一个延迟任务写入 HBase 和 Redis 成功后，提交流程就完成了。如果提交流程失败，我们会将失败的任务写入另一个 topic，并记录错误日志。
3.1.2 延迟任务到期流程
延迟任务到期流程可以进一步分为两个部分，分别是扫描任务分发和到期任务写回业务方。
延迟服务对外承诺分钟级别的时间精度。每分钟，系统会对已经存储的延迟任务进行检查，并且将到期的任务写入业务方。每分钟的检查操作可以分解为对单个 Zset 的扫描操作。检查操作由 dispatcher 协程产生。
多个实例上的 dispatcher 协程通过每分钟定时抢占分布式锁，来决定哪个实例分发这分钟的扫描任务。这种多实例机制是为了保证扫描任务分发不会出现单点故障。
对于系统中的每个 Zset，dispatcher 协程会生成这个 Zset 的扫描任务，将任务信息描述为一条消息，并写入 zset scan dispatch topic。扫描任务的主要信息包括需要扫描的 Zset key 以及要扫描的时间范围。
zset scan dispatch topic 的消费者取到一个扫描任务后，会从对应的 Zset 读取到期时间小于或等于扫描时间的任务。
需要注意的是，为了避免多个消费者同时对一个 Zset 进行操作，我们在写入 zset scan dispatch topic 时，采用 Zset key 作为 Kafka topic 的分区键。这样同一个 Zset 的扫描任务只会被单一消费者串行处理，从而避免了并发带来的问题。
从 Zset 扫描出到期任务的信息，包含任务的唯一 ID 和到期时间。这些信息，我们会到 HBase 中查询对应的任务 payload，最终拼接出完整的延迟任务，写回对应的业务 Kafka topic。
3.2 系统优化
3.1 中描述了系统实现的整体流程，在具体的实现和调优过程中，我们进一步针对具体问题细化了实现方案。
3.2.1 HBase 性能调优
我们设计的 rowkey 如下，主要分为三段，分别是 salt、exec_time 和 unique_id，其中 salt = hash(concat(exec_time, unique_id)) % region number。
Fig4 rowkey 设计
预分区策略和分区负载均衡
HBase 中的数据按照 rowkey 的字典序进行排序，在读写时容易产生热点问题，具体表现为：
- 读写的压力集中在某个 region server，不能很好地利用集群的分布式能力；
- 持续对同一个 region 写入导致频繁分裂，造成额外开销。
预分区和加盐是解决热点问题的常见方案。首先，我们预估系统积累的最大延迟任务个数，并据此估计所需的分区数量。在创建 HBase 表时，指定预分区规则。在写入延迟任务时，在 rowkey 前缀进行哈希加盐，将写入的数据均匀打散到各个分区。这样读写的压力就可以均摊到集群的每台机器，达到充分利用集群性能和可水平扩容的效果。
提高缓存命中
延迟任务的读写具有按到期时间聚集的特征，也就是说，exec_time 一样的记录，在短时间内会被先后读写。为了更好地利用 HBase 的读写缓存，我们将 exec_time 拼接在 salt 字段后。这样相同到期时间的记录会落到同一个缓存块中，进而大幅提升缓存命中率。
分表设置 TTL
HBase 存储的数据量较为庞大，如果不对数据设置 TTL，最终的存储开销将会变得无法接受。延迟任务写入 HBase 后，只需要保存到它到期时间之后即可。不同的延迟任务由于到期时间不同，无法统一设置一个合理的 TTL。
因此，我们在 HBase 中预先创建了不同 TTL 的表。在写入 HBase 时，根据需要保存的时长，将延迟任务写入对应的表中。这样可以使得这些延迟任务对应的记录尽早过期，以节省存储的开销。这一改动只需要我们在 Redis 中额外记录延迟任务对应的表序号。
3.2.2 扫描性能优化
Zset 分段机制
在 3.1 的整体设计中，我们描述了简单使用哈希来决定延迟任务分配到哪个 Zset 的方案。
这个方案存在一个问题，就是每分钟我们需要扫描 Redis 中所有的 Zset，才能获取当前系统中所有的到期任务。我们需要控制 Zset 包含的元素个数以确保 Zset 读写性能，所以，当系统中维护的延迟任务数量增加时，我们每分钟需要扫描的 Zset 个数也随之增加。
为了控制每分钟检查的 Zset 的个数，我们设计了 Zset 分段的方案。
Fig5. Zset key 设计
对于到期时间在一段时间内的延迟任务，我们将其放在一组 Zset 下面。例如 2022-06-27 10:00:00 ~ 2022-06-27 10:30:00 这 30 分钟到期的延迟任务，我们放在对应的 N 个 Zset 下面。
这些 Zset 的 key 为 [key_prefix]:[202206271000]:[0] ~ [key_prefix]:[202206271000]:[N-1]，它们的过期时间设置为 2022-06-27 10:30:00 之后的一段时间。
如此，在 2022-06-27 10:00:00 ~ 2022-06-27 10:30:00 这段时间内，我们下发的扫描任务只需要检查这 N 个 Zset 即可，大大减少了每分钟需要处理的 Zset 数量。
多协程消费 Kafka 机制
扫描 Zset 任务的执行是由 zset scan dispatch topic 的消费者来执行的。如果只依赖 Kafka 的 partition 提供的并发能力，那么扫描 Zset 的并发数会受到较大的限制。因为 Kafka topic 的 partiton 不能随意增加。
过多的 partition 不仅会导致 rebalance 时间过长，而且也会劣化 Kafka 集群的性能，将高效的顺序读写降级成多文件竞争的随机读写。
因此，我们在每个 partition 的 consumer 中进一步开启协程池来执行 Zset 的扫描任务。
对 partiton 开启协程池消费需要考虑到 offset 的提交策略。我们参考滑动窗口机制设计了对应的提交策略，确保每次提交 offset 时，这个 offset 之前的消息对应的扫描任务已经被协程处理完成。
我们向协程分发扫描任务时，同样遵循按 Zset key 进行 hash 的策略，保证同一个 Zset 不被并发处理。
Fig6. 多协程消费示意图
3.2.3 延迟任务的快慢提交策略
在我们的业务场景下，很多延迟任务的延迟时长是天级别的，这些任务并不需要跟其他到期时间较短的任务一样及时地写入系统内。
因此，我们可以在任务提交时区分快慢任务，让到期时间长的任务慢速写入，将系统的资源更多地留给较为急迫的延迟任务。
我们按照延迟时长的长短，将延迟时长大的任务写入慢速提交 topic，通过 Kafka 的配额限速机制，限制慢速提交 topic 的处理速度。
Fig7. 快慢提交 topic
3.2.4 可靠性提升
我们期望延迟服务系统在正常发布更新时实现精确一次语义，在出现系统上下游服务或中间件宕机时，保证至少一次语义。
避免并发问题
避免并发问题的核心是保证同一个 Zset 在一个时刻只会被一个协程处理，前文已进行过讲解，这里不再赘述。
另一个可能出现并发问题的场景是任务扫描流程扫描完某个 Zset 后，任务提交流程再写入同一个 Zset 被扫描过的范围。这个问题的解决方案也比较直观：
- 每次扫描只设到期时间的上限，不设下限。后提交的延迟任务会被下次扫描扫出；
- 扫描后清理已扫描数据，不是用到期时间范围去清理，而是用扫描出来的 key 做批量删除。
优雅退出
因为采用了多协程处理方案，为了避免发布前后的实例重复处理扫描任务，我们采用了优雅退出的设计。在接受到 K8s 的 sigterm 信号后，程序先停止对 zset scan dispatch topic 的消费，并且保证消费完已取出任务后再退出主函数。
异常回补策略
假如系统依赖的服务出现了长时间的宕机，我们可以利用 HBase 存储的延迟任务信息来进行异常恢复。具体做法是，扫描宕机时间段未处理的延迟任务，直接写入到下游的业务 topic 中。
4. 总结和展望
4.1 运行效果
经过开发测试后，我们对上文所述的延迟服务进行了全链路的压测。在 5 节点的 HBase 集群下，系统达到了预期的性能指标，并发读写 QPS 均达到了 20w+/s，且使用的 K8s 资源比前期方案减少了 75%。
4.2 未来优化方向
在设计方案时，我们已经考虑了多个业务方接入的场景。未来，对于不同 SLA 要求的业务，我们将会提供资源的隔离以保障敏感业务不受影响，进一步提升系统的通用性和稳定性。
本文作者
Du，后端技术专家，来自 Shopee Off-Platform Ads 团队。
