在工作中想实现一个延迟功能，一般会借助rocketmq或者kafka的延迟队列功能来实现，但是这俩个消息中间件都有一个弊端，就是很难支持任意时间段的延迟，所以我想借助redis实现一个任意时间段的延迟功能
总体架构图
上述图主要分为5个模块
1 路由模块，为了支持分布式部署，每接收一个延迟消息，都为这个消息生成一个全局唯一的消息ID，根据消息ID和路由算法，决定把延迟消息的消息ID加入到对应的redis.sortSet队列中
2 消息存储，所有的消息元数据存储采用redis.hashmap结构，key为生成的全局消息ID(可以加一个前缀)，value为消息的JSON格式，例如{"ttl":600,"topic":"XXX".......}
3 延迟队列，所有的消息的延迟存储在redis.sortSet中，sortSet中的每一个对象为全局生成的消息ID，score为到期时间时间戳
4 定时扫描timer，轮训redis.sortSet队列，使用ZRANGEBYSCORE命令，获取score小于等于当前时间的所有消息ID，然后根据消息ID查询redis.hashmap中的消息元数据，然后根据业务发送到对应的topic下
5 消息中间件，借助消息中间件，订阅者直接消费消息，达到延迟的功能
总结
上述只是整体的罗列了一下借助redis怎么实现任意时间段延迟的功能，一些细节没有详细说明，如果想实现一个比较完美的延迟功能，需要考虑以下几点
a 消息的发送失败如果处理
b redis操作没有事物保证
c 怎么保证hashmap和sortSet中的数据一致性
d 如果消息量大了怎么进行动态的扩展
网友评论