Redis实现简单延队列, 利用zset有序的数据结构, score设置为延时的时间戳.
实现思路:
1、使用命令 [zrangebyscore keyName socreMin socreMax] 会返回已score排序由小到大的一个list
2、list非空则使用[zrem keyName value] 删除第一个元素, 删除成功即代表消费成功, 可以解决多线程并发消费的问题.
使用jedis实现代码:
package com.nancy.utils; import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import redis.clients.jedis.Jedis; import java.lang.reflect.Type;
import java.util.*; /**
* redis实现延时队列，但是对于要求行极高的环境不建议使用，主要原因：
* 1、没有可靠的消息持久机制，消息容易丢失
* 2、ack应答机制确实，没有传统MQ机制的可靠性
*
* @author zhou.guangfeng on 2019/3/9 下午4:31
*/
public class RedisDelayQueue<T> { static class TaskItem<T>{
public String id ;
public T msg ;
} private Jedis jedis ;
private String queueKey ;
private Type taskType = new TypeReference<TaskItem<T>>(){}.getType(); RedisDelayQueue(Jedis jedis, String queueKey){
this.jedis = jedis ;
this.queueKey = queueKey ;
} public void delay(T msg){
TaskItem<T> item = new TaskItem<>() ; item.id = UUID.randomUUID().toString() ;
item.msg = msg ; String content = JSON.toJSONString(item) ;
try {
Thread.sleep(10L);
} catch (InterruptedException e) {
e.printStackTrace();
}
jedis.zadd(queueKey, System.currentTimeMillis() + 5000, content) ;
} public void loop(){
while (!Thread.interrupted()){
Boolean flag = consumer() ;
try {
if(!flag) {
Thread.sleep(500L);
}
}catch (InterruptedException ex){
break;
} }
} /**
*
* 队列消费，利用zrem操作，删除成功即也消费。 并发环境可能出现zrem删除失败情况，从而导致无效的请求。
* @param
* @return
*/
private Boolean consumer(){
// 按照分数即时间， 有序集成员按 score 值递增(从小到大)次序排列。
Set<String> values = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis()) ;
if (values == null || values.isEmpty()){
return false ;
} String content = values.iterator().next() ;
if(jedis.zrem(queueKey, content) <= 0){
return false ;
} TaskItem<T> item = JSON.parseObject(content, taskType) ;
handleMsg(item.msg) ;
return true ;
} public void handleMsg(T msg){
System.out.println(msg);
} public static void main(String[] args) {
RedisDelayQueue<String> queue = new RedisDelayQueue<>(AbstractDistributedLock.getJedisPool().getResource(), "delay-queue-demo") ; System.out.println("delay queue start, time = " + new Date());
Thread producer = new Thread(){
@Override
public void run() {
for (int i = 0; i < 10; i++) {
queue.delay("codehole:" + i);
}
}
}; Thread consumer = new Thread(){
@Override
public void run() {
queue.loop();
}
}; producer.start();
consumer.start(); try {
producer.join();
Thread.sleep(6000L); consumer.interrupt();
consumer.join();
}catch (InterruptedException ex){ }finally {
System.out.println("delay queue start, end = " + new Date());
} } }
public abstract class AbstractDistributedLock implements RedisLock {
private static JedisPool jedisPool;
protected static final String LOCK_SUCCESS = "OK";
protected static final Long RELEASE_SUCCESS = 1L;
protected static final String SET_IF_NOT_EXIST = "NX";
protected static final String SET_WITH_EXPIRE_TIME = "EX";
protected static final long DEFAULT_EXPIRE_TIME = 1000 * 10 ;
protected static final long DEFAULT_DELAY_TIME = 2 ;
static {
JedisPoolConfig config = new JedisPoolConfig();
// 设置最大连接数
config.setMaxTotal(500);
// 设置最大空闲数
config.setMaxIdle(50);
// 设置最大等待时间
config.setMaxWaitMillis(1000 * 100);
// 在borrow一个jedis实例时，是否需要验证，若为true，则所有jedis实例均是可用的
config.setTestOnBorrow(true);
jedisPool = new JedisPool(config, "127.0.0.1", 6379, 3000);
}
public static JedisPool getJedisPool() {
return jedisPool;
}
@Override
public String getLockKey(String lockKey){
return "lock:" + lockKey;
}
}
AbstractDistributedLock
运行结果:
delay queue start, time = Mon Mar 11 10:21:25 CST 2019
codehole:0
codehole:1
codehole:2
codehole:3
codehole:4
codehole:5
codehole:6
codehole:7
codehole:8
codehole:9
delay queue start, end = Mon Mar 11 10:21:31 CST 2019
以上的代码 jedis.zrangeByScore 和 jedis.zrem 为非原子操作. 如果jedis.zrem一旦失败, 会进入休眠, 造成资源浪费. 因此改造为使用lua脚本执行jedis.zrangeByScore 和 jedis.zrem 保证原子性.
/**
* 队列消费 使用lua脚本， 保证zrangebyscore 和 zrem操作原子性。
*
* @param
* @return
*/
private Boolean consumerWithLua(){
String script = " local resultDelayMsg = {}; " +
" local arr = redis.call('zrangebyscore', KEYS[1], '0', ARGV[1]) ; " +
" if next(arr) == nil then return resultDelayMsg end ;" +
" if redis.call('zrem', KEYS[1], arr[1]) > 0 then table.insert(resultDelayMsg, arr[1]) return resultDelayMsg end ; " +
" return resultDelayMsg ; ";
Object result = jedis.eval(script, Collections.singletonList(queueKey), Collections.singletonList("" + System.currentTimeMillis()));
List<String> msg = null ;
if (result == null || (msg = (List<String>) result).isEmpty()) {
return false ;
} TaskItem<T> item = JSON.parseObject(msg.get(0), taskType) ;
handleMsg(item.msg) ;
return true ;
}
redis实现延时队列，但是对于要求行极高的环境不建议使用，主要原因：
1、没有可靠的消息持久机制，消息容易丢失. 需要自己实现
2、ack应答机制缺失，没有传统MQ机制的可靠性
因此, 如果对数据一致性有严格的要求, 还是建议使用传统MQ.