【Redis】延时队列

布满荆棘的人生 2022-03-16 10:18 260阅读 0赞

Redis 的 list(列表) 数据结构常用来作为异步消息队列使用，使用`rpush/lpush`操作入队列，使用`lpop 和 rpop`来出队列。

## 队列延迟 ##

blpop/brpop。这两个指令的前缀字符`b`代表的是`blocking`，也就是阻塞读。

阻塞读在队列没有数据的时候，会立即进入休眠状态，一旦数据到来，则立刻醒过来。消息的延迟几乎为零。用`blpop/brpop`替代前面的`lpop/rpop。`

## 空闲连接自动断开 ##

如果线程一直阻塞在哪里，Redis 的客户端连接就成了闲置连接，闲置过久，服务器一般会主动断开连接，减少闲置资源占用。这个时候`blpop/brpop`会抛出异常来。

所以编写客户端消费者的时候要小心，注意捕获异常，还要重试。

## 锁冲突处理 ##

上节课我们讲了分布式锁的问题，但是没有提到客户端在处理请求时加锁没加成功怎么办。一般有 3 种策略来处理加锁失败：

1.  直接抛出异常，通知用户稍后重试；
2.  sleep 一会再重试；
3.  将请求转移至延时队列，过一会再试；

## 延时队列的实现  ##

import java.lang.reflect.Type;
    import java.util.Set;
    import java.util.UUID;
    
    import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.TypeReference;
    
    import redis.clients.jedis.Jedis;
    
    public class RedisDelayingQueue<T> {
    
      static class TaskItem<T> {
        public String id;
        public T msg;
      }
    
      // fastjson 序列化对象中存在 generic 类型时，需要使用 TypeReference
      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 delay(T msg) {
        TaskItem<T> task = new TaskItem<T>();
        task.id = UUID.randomUUID().toString(); // 分配唯一的 uuid
        task.msg = msg;
        String s = JSON.toJSONString(task); // fastjson 序列化
        jedis.zadd(queueKey, System.currentTimeMillis() + 5000, s); // 塞入延时队列 ,5s 后再试
      }
    
      public void loop() {
        while (!Thread.interrupted()) {
          // 只取一条
          Set<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); // fastjson 反序列化
            this.handleMsg(task.msg);
          }
        }
      }
    
      public void handleMsg(T msg) {
        System.out.println(msg);
      }
    
      public static void main(String[] args) {
        Jedis jedis = new Jedis();
        RedisDelayingQueue<String> queue = new RedisDelayingQueue<>(jedis, "q-demo");
        Thread producer = new Thread() {
    
          public void run() {
            for (int i = 0; i < 10; i++) {
              queue.delay("codehole" + i);
            }
          }
    
        };
        Thread consumer = new Thread() {
    
          public void run() {
            queue.loop();
          }
    
        };
        producer.start();
        consumer.start();
        try {
          producer.join();
          Thread.sleep(6000);
          consumer.interrupt();
          consumer.join();
        } catch (InterruptedException e) {
        }
      }
    }