基于Redis如何实现阻塞队列

这篇文章主要介绍"基于Redis如何实现阻塞队列"，在日常操作中，相信很多人在基于Redis如何实现阻塞队列问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答"基于Redis如何实现阻塞队列"的疑惑有所帮助！接下来，请跟着小编一起来学习吧！

基于 Redis 的 list 实现队列的方式也有多种，先说第一种不推荐的方式，即使用LPUSH生产消息，然后 while(true) 中通过RPOP消费消息，这种方式的确可以实现，但是不断代码不断的轮询，势必会消耗一些系统的资源。

第二种方式也是不推荐的方式，也是通过 LPUSH生产消息，然后通过 BRPOP 进行阻塞地等待并消费消息，这种方式较第一种方式减少了无用的轮询，降低系统资源的消耗，但是可能会存在队列消息丢失的情况，如果取出了消息然后处理失败，这个被取出的消息就将丢失。

第二种方式就是下文要介绍的方式，首先也是通过 LPUSH 生产消息，然后通过 BRPOPLPUSH阻塞地等待 list 新消息到来，有了新消息才开始消费，同时将消息备份到另外一个 list 当中，这种方式具备了第二种方式的优点，即减少了无用的轮询，同时也对消息进行了备份不会丢失数据，如果处理成功，可以通过 LREM 对备份的 list 中当前的这条消息进行删除处理。

Redis 基础

# 将一个或多个值 value 插入到列表 key 的表头LPUSH key value [value …]# 阻塞式等待，将列表 source 中的最后一个元素 (尾元素) 弹出，并返回给客户端。将 source 弹出的元素插入到列表 destination ，作为 destination 列表的的头元素。超时参数 timeout 接受一个以秒为单位的数字作为值。超时参数设为 0 表示阻塞时间可以无限期延长 (block indefinitely) 。BRPOPLPUSH source destination timeout# 根据参数 count 的值，移除列表中与参数 value 相等的元素。LREM key count value

代码实现队列消息生产者

笔者使用的是 Spring 相关 API 实现对 Redis 指令的调用。首先实现消息的生产代码，封装到一个工具类方法当中。这里很简单，就是调用了 lpush 方法，将序列化的 key 和 value 添加到列表当中去。

@Resourceprivate RedisConnectionFactory connectionFactory;public void lPush(@Nonnull String key, @Nonnull String value) {  RedisConnection connection = RedisConnectionUtils.getConnection(connectionFactory);  try {    byte[] byteKey = RedisSerializer.string().serialize(getKey(key));    byte[] byteValue = RedisSerializer.string().serialize(value);    assert byteKey != null;    connection.lPush(byteKey, byteValue);  } finally {    RedisConnectionUtils.releaseConnection(connection, connectionFactory);  }}

代码实现队列消息消费者

因为实现队列消费消息的代码比较多，不可能每个需要阻塞消费的地方，对需要写这一坨代码，因此使用 Java8 的函数式接口实现方法的传递，同时阻塞式获取消息代码使用新线程去执行。

有人看到以下代码要吐槽了，不是说不用 while(true) 吗，怎么你这里面还是有，这里稍微解释一下，因为 SpringBoot 一般会指定 timeout 的全局超时时间，即使 BRPOPLPUSH 设置了 0，即无限期，当超出了 timeout 设置的值时，就会抛出 QueryTimeoutException 异常导致线程退出，因此添加了 try/catch 对异常进行捕获并忽略，同时使用 while(true) 保证线程可以继续执行。
代码中记录了当前消息处理结果，如果处理结果为成功，需要对备份队列的当前消息进行删除。

public void bRPopLPush(@Nonnull String key, Consumer consumer) {  CompletableFuture.runAsync(() -> {    RedisConnection connection = RedisConnectionUtils.getConnection(connectionFactory);    try {      byte[] srcKey = RedisSerializer.string().serialize(getKey(key));      byte[] dstKey = RedisSerializer.string().serialize(getBackupKey(key));      assert srcKey != null;      assert dstKey != null;      while (true) {        byte[] byteValue = new byte[0];        boolean success = false;        try {          byteValue = connection.bRPopLPush(0, srcKey, dstKey);          if (byteValue != null && byteValue.length != 0) {            consumer.accept(new String(byteValue));            success = true;          }        } catch (Exception ignored) {          // 防止获取 key 达到超时时间抛出 QueryTimeoutException 异常退出        } finally {          if (success) {            // 处理成功才删除备份队列的 key            connection.lRem(dstKey, 1, byteValue);          }        }      }    } finally {      RedisConnectionUtils.releaseConnection(connection, connectionFactory);    }  });}

测试代码

@Testpublic void testLPush() throws InterruptedException {  String queueA = "queueA";  int i = 0;  while (true) {    String msg = "Hello-" + i++;    redisBlockQueue.lPush(queueA, msg);    System.out.println("lPush: " + msg);    Thread.sleep(3000);  }}@Testpublic void testBRPopLPush() {  String queueA = "queueA";  redisBlockQueue.bRPopLPush(queueA, (val) -> {    // 在这里处理具体的业务逻辑    System.out.println("val: " + val);  });  // 防止 Junit 进程退出  LockSupport.park();}

项目使用方式

为了方便使用，我将其抽取为了一个工具类，使用时通过 Spring 注入使用即可，
队列消费可以使用如下方式在项目启动的时候就进行阻塞监听队列，等待消费

@Resourceprivate RedisBlockQueue redisBlockQueue;@PostConstructpublic void init() {   redisBlockQueue.bRPopLPush(xx, (value) -> {     //...   });}

到此，关于"基于Redis如何实现阻塞队列"的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注网站，小编会继续努力为大家带来更多实用的文章！