Redis 实现延迟队列

关于消息队列，我们习惯于使用Rabbitmq 和 Kafka 作为消息队列中间件。但对于消费队列只有一组消费者时，也不需要非常高的可靠性，使用中间件显得十分繁琐，此时我们可以利用redis的特性来实现简单的消息队列。

异步消息队列

Redis 的list 数据结构常用来作为异步消息队列使用，用rpush和lpush 操作如队列，用lpop 和 rpop 的操作出队列

它支持多个生产者和多个消费者并发进出消息，每个消费者拿到的消息都是不同的列表元素，如图所示：

> rpush notify-queue apple banba pear
> lpop notify-queue
“apple”
> lpop notify-queue
"banana" 
> lpop notify-queue
"pear"
> lpop notify-queue
(nil)
1
2
3
4
5
6
7
8
9

上面是rpush 和 lpop 结合使用，也可以吧lpush 和 rpop 结合使用，效果一样。

队列为空时的处理

客户端通过队列的pop操作来获取消息，然后进行处理。处理完接着获取消息，再进行处理。如此循环，便是作为队列消费者的客户端的生命周期。
但如果队列空了，客户端就会陷入pop死循环，不停地pop，这种空轮训不仅拉高了客户端的CPU消耗，Redis的QPS 也会被拉高，如果这样的空轮询的客户端有几十个，Redis的慢查询可能会显著增多。

通常我们使用sleep来解决这个问题，如果队列为空就让线程睡一会，这样既降低了客户端的CPU消耗，Redis的QPS也降下来了。

阻塞读

用上面睡眠的办法可以解决问题。但是又有个小问题，那就是睡眠会导致消息的延迟增大。如果只有1个消费者，那么这个延迟就是1s。如果有多个消费者，这个延迟就会有所下降，因为消费者的睡眠时间时岔开的。
有没有什么什么办法既能解决消耗问题，又能解决延迟问题呢？当然也有，那就是blpop/brpop.
这两个指令的前缀字符b代表的是blocking，也就是阻塞读。
阻塞读在对列没有数据的时候，会立即进入休眠状态，一旦数据到来，则立即醒过来。消息的延迟几乎为零。

空闲连接自动断开

上面的方案虽然解决了对列为空的问题，但并不是很完美，它仍然会出现空闲连接的问题。
如果线程一直阻塞在哪，Redis的客户端连接就成了闲置连接，闲置过久，服务器一般会主动断开连接，减少闲置资源占用。这个时候blpop/brpop会抛出异常。所以编写客户端消费者的时候要小心，如果捕获到异常，还要重试。

延时队列的实现

延时队列可以通过Redis 的zset 来实现。我们将消息序列化成一个字符串作为zset的value，这个消息的到期处理时间作为score，然后用多个线程轮询zset获取到期的任务进行处理。多个线程是为了保障可用性，万一挂了一个线程还有其他线程可以继续处理。因为有多个线程，所以需要考虑并发争抢任务，确保任务不会被多次执行。

Redis 的 zrem 方法时多线程多进程争抢任务的关键，它的返回值决定了当前实例有没有抢到任务，因为loop方法可能会被多个线程、多个进程调用，同一个任务可能会被多个进程、多个线程抢到。
同时我们要对handle_msg 进行一场捕获，防止个别人物处理问题导致循环失败异常退出。

public class RedisDelayingQueue<T> {
    static class  TaskItem<T> {
        public String id;
        public T msg;
    }

    private 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;
    }

    public void deplay(T msg) {
        TaskItem<T> task = new TaskItem<>();
        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()) {
            Set<String>  values = jedis.zrangeByScore(queueKey,0,System.currentTimeMillis(),0 , 1);
            if (values.isEmpty()) {
                try {
                    Thread.sleep(50);
                }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);
    }

    public static void main(String[] args) {
        Jedis jedis = new Jedis("124.221.52.57",6379);
        jedis.auth("li12345...");
        jedis.select(2);
        final RedisDelayingQueue<String> queue = new RedisDelayingQueue<>(jedis,"q-demo");
        Thread producer = new Thread() {
            public void run() {
                for (int i = 0; i < 10; i++) {
                    queue.deplay("codehole"+i);
                }
            }
        };
        Thread consumer = new Thread() {
            public void run() {
                queue.loop();
            }
        };
        producer.start();
        consumer.start();
        try {
//            producer.join();
            Thread.sleep(6000);
            consumer.interrupt();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77