基于Redis实现的延时队列

基于Redis实现的延时队列

针对于Redis实现延时队列有两种实现方式：

使用zset实现实现的延时队列

借助redis zset来实现延时队列，具体的实现代码很简单，就是从zset中取出score小于当前时间戳的数据

import cn.hutool.json.JSONUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;@Component
public class RedisDelayDemo {@Autowiredprivate RedisTemplate redisTemplate;private static final Long DELETE_SUCCESS = 1L;//线程安全setprivate Set<String> topic = new CopyOnWriteArraySet<>();//放入数据public  void push(String key, Object val, long delayTime) {topic.add(key);//hutool的json转换工具会有些问题，性能相对来说比较原生的jackson也会差一些  我这边是为了图方便String strVal = val instanceof String ? (String) val : JSONUtil.toJsonStr(val);redisTemplate.opsForZSet().add(key, strVal, System.currentTimeMillis() + delayTime);}//获取数据public String get(String key) {Set<String> sets = redisTemplate.opsForZSet().rangeByScore(key, 0, System.currentTimeMillis(), 0, 3);if (CollectionUtils.isEmpty(sets)) {return null;}for (String val : sets) {if (DELETE_SUCCESS.equals(redisTemplate.opsForZSet().remove(key, val))) {// 删除成功，表示抢占到return val;}}return null;}}

为什么会这么设计？

• 为啥将数据返回包在删除中？

  • 如果我们按照正常的实现流程，每次从zset中取一个，在单实例（单机）的情况下这个操作的话是没有问题的，但是你无法保证该情况下就只有一个人拿到了这个数据，在多实例（多机）的场景下，可能存在多个实例同时拿到了它，那我们要如何表示只有一个实例拥有了她呢？此时我们就要借助redis的单线程机制，只可能会有一个实例会删除成功，所以拿到并删除成功的那个实例就是幸运儿

• 为啥一次取3个数据

  • 如果一次只拿一个，那么每次抢占到数据的几率并不太大，特别是当实例比较多时，可能会做多次的无效操作，为了减少这个可能性，所以可以一次多拿几个做备选，具体拿几个这个看你定时任务的时间间隔和具体的业务场景

Demo

 final String key = "myzset";
for (int i = 0; i < 10; i++) {redisDelayDemo.push(key, "xiaozhang"+  i, 9000);
}
while (true) {String value = redisDelayDemo.get(key);if (StringUtils.isBlank(value)) {continue;}System.out.println(value);
}

基于Redis过期监听实现延时队列

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;public class RedisExpiredListenter extends KeyExpirationEventMessageListener {private static final Logger LOGGER = LoggerFactory.getLogger(RedisExpiredListenter.class);public RedisExpiredListenter(RedisMessageListenerContainer listenerContainer) {super(listenerContainer);}@Overridepublic void onMessage(Message message, byte[] pattern) {String channel = new String(message.getChannel(), StandardCharsets.UTF_8);//过期的keyString key = new String(message.getBody(),StandardCharsets.UTF_8);LOGGER.info("redis key 过期：pattern={},channel={},key={}",new String(pattern),channel,key);/***你可以将你的内容一起拼接到key中或者可以在redis中存储两个key （例如mykey和mykey_backup）*针对mykey设置过期时间，对于mykey_backup不设置过期时间，这样就可以通过mykey_backup获取到value了**/}}
}

使用Redis过期监听实现延时队列方法较为便捷，但是该方法也存在一个很大的问题
因为当过期的key数量高于一个阈值的时候， Redis 不能确保 key 在指定时间被删除 , 也就造成了消息可能比你设置的延时时间更长
所以如果你的系统针对于延时队列这个时间要求十分严格，并且在同一时间内会有多个消息需要发生那我就不推荐使用Redis的延时队列，如果你的系统对于该业务并没有如此严格的要求，并且数量不多的情况下是可以使用的。

使用Redis 实现延时队列还有一个比较大的问题，他并不像消息队列一样保证送达。当订阅事件的客户端会丢失所有在断线期间所有分发给它的事件。 这个问题也是开发者需要考虑的，根据自己的业务场景去判断。