极客时间 分布式缓存高手课 笔记08 支付 延时队列

延时队列

 

49 支付通知中心设计

Delay Bucket 使用redis的zset

50 支付余额更新

++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

延迟消息整个处理流程如下：

1. Producer发送消息到DelayServer上，携带延迟的相对值，即从现在开始，多久后消费；

2. DelaySever收到消息后，将延迟相对值加上当前时间，即到期时间，作为主key，消息体作为Value，发送到Tair；

3. DelayServer节点通过心跳获取到castle下发的主题，各个节点只处理分发给自己主题，根据时间戳和主题名称组成key，从Cellar中取到期消息；

4. 取出消息后发送给Mafka Broker，后续流程与非延迟消费消息处理逻辑一致，mafka客户端消费消息

Delay-server发送服务的设计

1. 预加载机制

   异步线程预先load数据至内存中，从两个维度去控制加载到DelayQueue中的数据：

   1、时间：预先加载最近半小时即将过期的数据。

   2、控制整个DelayQueue使用的内存数在一定范围之内（如：500M），可通过控制DelayQueue中的消息条数之类的来控制，防止DelayServer出现OOM，假设平均情况下一条消息占用了1k的内存，那么整个DelayQueue如果

   使用500m的话，可以理解为最多存放500000条数据在其中。

2. 维护一张topic到producer实例的table

   每个延迟消费的topic，都会创建一个与之对应的producer实例，可知的情况：这种topic在一定数量范围之内，由于默认的情况下一个producer实例申请的内存池空间是32m，所以内存可控制。

3. DelayServer发送数据到Mafka Broker

   从DelayQueue中取出数据，根据topic找到producer实例，调用sendMessage接口或者sendAsyncMessage 接口即可。

4. 清除tair中的延迟消息

   producer发送后，接收到的response状态为OK，则根据指定的key去删除tair中存储的数据。

     可以看出 DelaySever 主要具有如下功能：接收消息并存储到 Cellar、到期消息拉取、到期消息发送到 Broker。事实上，根据 DelayServer 所提供的功能，相应的 DelayServer 主要分为状态控制模块和消息处理模块，其中消息处理模块又可分为消息接收存储和到期消息拉取发送。

      DelayServer 中主要用到了 prefixPut 和 getRange 两个接口，其中 prefixPut 用于消息存储到 Cellar，getRange 用于到期消息从 Cellar 中获取。存储消息使用 prefixPut 接口而不是 put 操作的原因在于prefixPut 通过主 Key 进行哈希，而 put 是整个 Key 进行哈希，这样可以保证一个主 Key 下的消息存储到同一个桶上，获取时通过 getRange 可以快速的获取到，此接口 QPS 约在 2w~3W 间。通过 getRange 获取主 Key 下的全部或部分内容，如果消息较多，会分多次返回，每次返回的数据约 1M，QPS 受消息大小和个数影响。prefixPut 接口和 getRange 接口的具体接口定义可查看Cellar Java客户端 API文档

  到期消息读取发送模块

    到期消息读取发送模块，主要用于到期生成主 Key，并循环分批从 Cellar 获取该主 Key 下的所有副 Key 对应的消息，对于每一批次的消息调用内部 Producer 逐条发送到 Broker，对于发送失败的消息，将该消息存入主 Key 为“Failed_Message_Key”的 Cellar 中。