redis消息队列如何延时

redis消息队列如何延时

redis消息队列延时的示例:

延时队列可通过zset来实现,消息的处理时间作为score,最后通过多线程轮询获取到期的score任务即可,代码:

public class DelayQueue {

static class TaskItem {

public String id;

public T msg;

}

private Type taskType = new TypeReference>() {

}.getType();

private Jedis jedis;

private String queueName;

public DelayQueue(Jedis jedis, String queueName) {

this.jedis = jedis;

this.queueName = queueName;

}

public void delay(T msg, long delayTime) {

TaskItem task = new TaskItem<>();

task.id = UUID.randomUUID().toString();

task.msg = msg;

jedis.zadd(queueName, System.currentTimeMillis() + delayTime, JSON.toJSONString(task));

}

public void loop() {

while (!Thread.interrupted()) {

Set set = jedis.zrangeByScore(queueName, 0, System.currentTimeMillis(), 0, 1);

if (set.isEmpty()) {

try {

Thread.sleep(500);

} catch (InterruptedException e) {

break;

}

continue;

}

String s = set.iterator().next();

if (jedis.zrem(queueName, s) > 0) {

TaskItem task = JSON.parseObject(s, taskType);

System.out.println(task.msg);

}

}

}

}

2.测试代码:

public static void main(String[] args) throws InterruptedException {

Jedis jedis = new Jedis("127.0.0.1", 6379);

DelayQueue delayQueue = new DelayQueue<>(jedis, "delay_queue");

Thread producer = new Thread(() -> {

for (int i = 0; i < 10; i++)

delayQueue.delay("Mr.Wang's Hub" + i, 5000);

});

Thread consumer = new Thread(() -> {

delayQueue.loop();

});

consumer.start();

producer.start();

while (Thread.activeCount() > 1)

Thread.yield();

}

我们的服务
专业提供中小企业网站建设.公司网站建设.外贸网站建设.营销型网站建设,响应式网站建设,商城网站定制,手机网站,微信小程序定制.php应用开发服务,公司成立于2014年,8年专注网站建设,秉承“以客户为根本,以科技为核心,以市场为导向”的经营理念;服务电话13714666846(微信同号)

声明:本站所有作品(图文、音视频)均由用户自行上传分享,或互联网相关知识整合,仅供网友学习交流,若您的权利被侵害,请联系 管理员 删除。

本文链接:http://news.ew35.com/