Redis之使用zset实现异步队列

编程入门行业动态更新时间:2024-10-10 11:25:33

Redis之使用zset实现异步<a href=https://www.elefans.com/category/jswz/34/1771257.html style= 队列"/>

Redis之使用zset实现异步队列

当遇到并发的客户端请求时，为了缓解服务端的处理压力，当请求对响应的处理的实时性要求不高时，可以实现一个异步的请求消息队列。

一种实现策略是使用redis的zset，将消息的到期处理时间作为score，然后用多个线程去轮训获取zset中的任务并进行处理。

需要提前考虑一个问题：

如何避免一个任务被多次处理？

一种解决方案是当多个线程获取到任务时，调用redis的zrem命令，将该任务从指定的zset中移除(利用了redis处理命令时是顺序执行的)。

环境

JDK17
两个jar包
- Jedis
- fastjson2

代码

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import redis.clients.jedis.Jedis;import java.lang.reflect.Type;
import java.util.List;
import java.util.UUID;// 基于Redis实现的延迟队列
public class RedisDelayingQueue<T> {static class TaskItem<T> {public String id;public T msg;}// fastjson序列化对象时如果存在泛型，需要使用TypeReferenceprivate Type TaskType = new TypeReference<TaskItem<T>>(){}.getType();private Jedis jedis;private String queueKey;public RedisDelayingQueue(Jedis jedis, String queueKey) {this.jedis = jedis;this.queueKey = queueKey;}// 将任务添加到 zset 中// 分数是延时的时间public void delay(T msg) {TaskItem<T> task = new TaskItem<T>();task.id = UUID.randomUUID().toString();task.msg = msg;// 序列化任务String s = JSON.toJSONString(task);jedis.zadd(queueKey, System.currentTimeMillis() + 5000, s);}public void loop() {while(!Thread.interrupted()) {// 从zset中取出一个任务List<String> values = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis(), 0, 1);if(values.isEmpty()) {try {Thread.sleep(500);} catch(InterruptedException e) {break;}continue;}String s = values.iterator().next();if(jedis.zrem(queueKey, s) > 0) {TaskItem<T> task = JSON.parseObject(s, TaskType);this.handleMsg(task.msg);}}}public void handleMsg(T msg) {System.out.println(msg);}
}

优化

通过上面loop中代码，多个线程获取到values时，可能会被多个线程同时取到，然后再调用zrem命令去竞争的删除该值，所以会有很多无用的网络请求发送到redis。更容易想到的方案是将取值然后删除的操作变成原子性的，两种实现方案：

通过对代码块进行加锁的方式
利用redis中lua脚本的原子执行的特点

代码块加锁

这种方案不太好，如果两个命令之间发生了网络错误或者延迟，将造成其它线程的阻塞

    public void synchronizedLoop() {while(!Thread.interrupted()) {synchronized(this) {// 从zset中取出一个任务List<String> values = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis(), 0, 1);if(values.isEmpty()) {try {Thread.sleep(500);} catch(InterruptedException e) {break;}continue;}String s = values.iterator().next();if(jedis.zrem(queueKey, s) > 0) {TaskItem<T> task = JSON.parseObject(s, TaskType);this.handleMsg(task.msg);}}}}

Lua脚本

local key = KEYS[1]
local task = redis.call('ZPOPMIN', key)
if task and next(task) != nil thenredis.call('ZREM', key, task[1])return task[1]
elsereturn nil
end

通过查阅文档发现，ZRANGEBYSCORE从6 版本开始已经过时了，所以这里使用ZPOPMIN来获取分数最小的value，可以达到相同的效果。

通过Jedis的eval函数，调用redis执行lua脚本的命令。

    public void luaLoop() {while(!Thread.interrupted()) {Object ans = jedis.eval(script, 1, queueKey);if(ans != null) {String task = (String) ans;TaskItem<T> taskItem = JSON.parseObject(task, TaskType);this.handleMsg(taskItem.msg);}else{try{Thread.sleep(500);}catch(Exception e) {break;}}}}

为什么可以优化：

使用lua脚本的方式，使得一个线程如果zset中有任务都会成功获取任务，而不会多个线程同时拿到同一个任务，再去竞争删除，减少了无效的网络IO

测试程序

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;public class Main {public static void main(String[] args) {JedisPool jedisPool = new JedisPool("url-of-redis", 6379, "username", "pass");Jedis jedis = jedisPool.getResource();RedisDelayingQueue<String> queue = new RedisDelayingQueue<>(jedis, "q-demo");// 创建一个线程充当生产者，并向redis中存10个异步任务Thread producer = new Thread() {public void run() {for (int i = 0; i < 10; i++) {queue.delay("codehole" + i);}}};// 创建一个线程充当消费者，不断从redis中取任务并执行Thread consumer = new Thread() {public void run() {queue.luaLoop();}};producer.start();consumer.start();try {// 等待生产者线程执行结束producer.join();Thread.sleep(6000);consumer.interrupt();consumer.join();}catch(InterruptedException e) {e.printStackTrace();}}
}