中文字幕av专区_日韩电影在线播放_精品国产精品久久一区免费式_av在线免费观看网站

溫馨提示×

redis消息隊列如何延時

小新
191
2020-12-25 17:40:57
欄目: 云計算

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();

}

0
皋兰县| 揭西县| 全南县| 鸡东县| 鹿邑县| 广安市| 灌云县| 富阳市| 佛学| 洞头县| 定结县| 安吉县| 吉木萨尔县| 台湾省| 通州市| 江西省| 常宁市| 乡城县| 甘德县| 清河县| 遂昌县| 通化县| 保定市| 资讯| 大冶市| 崇州市| 三亚市| 麻江县| 施秉县| 咸宁市| 诸暨市| 皮山县| 廉江市| 宝应县| 西藏| 天峻县| 察隅县| 习水县| 定襄县| 恩平市| 加查县|