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.

最新文章

  1. 关于jsp页面将表单填入数据库出现中文乱码绝对解决方案
  2. jquery weibo 留言
  3. WPF-非矩形窗口的创建
  4. iOS-修改Status Bar
  5. HTTP基础03--HTTP报文
  6. SpringMVC中的controller默认是单例的原因
  7. HDU 4630 No Pain No Game 树状数组+离线查询
  8. linux进程,作业,守护进程,进程间同步
  9. ul ol 列表的样式的控制
  10. 【数位DP】【HDU2089】不要62
  11. zeromq-python使用
  12. 如何测试本地是否能够正常访问云服务器的 Web 端口
  13. 主流页面编辑器比对:TinyMCE VS FCKeditor
  14. Linux-mknod命令(9)
  15. Android线上Bug热修复分析
  16. JAVA篇&lt;一&gt; 继承extends(已转移到JAVA总结篇)
  17. SSM框架集成及配置详解(Maven管理)
  18. 分区默认segment大小变化(64k—&gt;8M)
  19. cisco和华为的设备如何设置命令不分页显示
  20. Codeforces.209C.Trails and Glades(构造 欧拉回路)

热门文章

  1. 安装xcache3.0.3/3.2,为php加速
  2. Android动画之逐帧动画(FrameAnimation)详解
  3. Windows不能在本地计算机启动MongoDB,错误代码 100
  4. 【splunk】一些查询例子
  5. PHP 迭代器和生成器
  6. OpenLdap的加密md5(Java+Python,同时提供明文--&gt;密文,md5(名文)--&gt;密文两种方法)
  7. 详解kubeadm生成的证书(转)
  8. springboot项目连接数据库报错
  9. [转] HTML5利用WebRTC的getUserMedia获取摄像头信息模拟拍照及视频(完整示例)
  10. ThreadPoolExecutor线程池的分析和使用