基于Redis实现阻塞队列

日常需求开发过程中,不免会遇到需要通过代码进行异步处理的情况,比如批量发送邮件,批量发送短信,数据导入,为了减少用户的等待,不希望一直菊花转啊转,因此需要进行异步处理,做法就是讲要处理的数据添加到队列当中,然后按照排队的先后顺序进行异步处理。

这个队列,可以是专业的消息队列,如 RocketMQ/RabbitMQ 等,一般项目中,如果只是为了进行异步,未免有点杀鸡用牛刀的意味。
也可以使用基于 JVM 内存实现队列,但是如果项目进行了重启,就会造成队列数据丢失。
大部分的项目都会用到 Redis 中间件作为缓存使用,此时使用 Redis 的 list 结构来实现队列则是非常合适的选择。

因此,本文主要讲解基于 Redis 的方式实现异步队列。

本文首发个人技术博客: https://nullpointer.pw/redis-block-queue.html

基于 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 添加到列表当中去。

@Resource
private 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<String> 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);
    }
  });
}

测试代码

@Test
public 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);
  }
}

@Test
public void testBRPopLPush() {
  String queueA = "queueA";
  redisBlockQueue.bRPopLPush(queueA, (val) -> {
    // 在这里处理具体的业务逻辑
    System.out.println("val: " + val);
  });

  // 防止 Junit 进程退出
  LockSupport.park();
}

项目使用方式

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

@Resource
private RedisBlockQueue redisBlockQueue;

@PostConstruct
public void init() {
   redisBlockQueue.bRPopLPush(xx, (value) -> {
     //...
   });
}

本文完整代码下载github 地址

到此这篇关于基于Redis实现阻塞队列的文章就介绍到这了,更多相关Redis阻塞队列内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

(0)

相关推荐

  • 基于Redis延迟队列的实现代码

    使用场景 工作中大家往往会遇到类似的场景: 1.对于红包场景,账户 A 对账户 B 发出红包通常在 1 天后会自动归还到原账户. 2.对于实时支付场景,如果账户 A 对商户 S 付款 100 元,5秒后没有收到支付方回调将自动取消订单. 解决方案分析 方案一: 采用通过定时任务采用数据库/非关系型数据库轮询方案. 优点: 1. 实现简单,对于项目前期这样是最容易的解决方案. 缺点: 1. DB 有效使用率低,需要将一部分的数据库的QPS分配给 JOB 的无效轮询. 2. 服务资源浪费,因为轮询需

  • Java redisTemplate阻塞式处理消息队列

    目录 Redis 消息队列 redis五种数据结构 队列生产者 队列消费者 测试类 并发情况下使用increment递增 补充 Redis 消息队列 redis五种数据结构 队列生产者 package cn.stylefeng.guns.knowledge.modular.knowledge.schedule; import lombok.extern.slf4j.Slf4j; import org.springframework.data.redis.core.RedisTemplate; i

  • 基于Redis实现阻塞队列

    日常需求开发过程中,不免会遇到需要通过代码进行异步处理的情况,比如批量发送邮件,批量发送短信,数据导入,为了减少用户的等待,不希望一直菊花转啊转,因此需要进行异步处理,做法就是讲要处理的数据添加到队列当中,然后按照排队的先后顺序进行异步处理. 这个队列,可以是专业的消息队列,如 RocketMQ/RabbitMQ 等,一般项目中,如果只是为了进行异步,未免有点杀鸡用牛刀的意味. 也可以使用基于 JVM 内存实现队列,但是如果项目进行了重启,就会造成队列数据丢失. 大部分的项目都会用到 Redis

  • 基于Redis实现阻塞队列的方式

    日常需求开发过程中,不免会遇到需要通过代码进行异步处理的情况,比如批量发送邮件,批量发送短信,数据导入,为了减少用户的等待,不希望一直菊花转啊转,因此需要进行异步处理,做法就是讲要处理的数据添加到队列当中,然后按照排队的先后顺序进行异步处理. 这个队列,可以是专业的消息队列,如 RocketMQ/RabbitMQ 等,一般项目中,如果只是为了进行异步,未免有点杀鸡用牛刀的意味. 也可以使用基于 JVM 内存实现队列,但是如果项目进行了重启,就会造成队列数据丢失. 大部分的项目都会用到 Redis

  • PHP实现基于Redis的MessageQueue队列封装操作示例

    本文实例讲述了PHP实现基于Redis的MessageQueue队列封装操作.分享给大家供大家参考,具体如下: Redis的链表List可以用来做链表,高并发的特性非常适合做分布式的并行消息传递. 项目地址:https://github.com/huyanping/Zebra-PHP-Framework 左进右出 $redis->lPush($key, $value); $redis->rPop($key); 以下程序已在生产环境中正式使用. 基于Redis的PHP消息队列封装 <?ph

  • 基于Redis实现延时队列的优化方案小结

    目录 一.延时队列的应用 二.延时队列的实现 三.总结 一.延时队列的应用 近期在开发部门的新项目,其中有个关键功能就是智能推送,即根据用户行为在特定的时间点向用户推送相应的提醒消息,比如以下业务场景: 在用户点击充值项后,半小时内未充值,向用户推送充值未完成提醒. 在用户最近一次阅读行为2小时后,向用户推送继续阅读提醒. 在用户新注册或退出应用N分钟后,向用户推送合适的推荐消息. … 上述场景的共同特征就是在某事件触发后延迟一定时间后再执行特定任务,若事件触发时间点可知,则上述逻辑也可等价于在

  • Java 阻塞队列详解及简单使用

     Java 阻塞队列详解 概要: 在新增的Concurrent包中,BlockingQueue很好的解决了多线程中,如何高效安全"传输"数据的问题.通过这些高效并且线程安全的队列类,为我们快速搭建高质量的多线程程序带来极大的便利.本文详细介绍了BlockingQueue家庭中的所有成员,包括他们各自的功能以及常见使用场景. 认识BlockingQueue阻塞队列,顾名思义,首先它是一个队列,而一个队列在数据结构中所起的作用大致如下图所示: 从上图我们可以很清楚看到,通过一个共享的队列,

  • Java中常用阻塞队列的问题小结

    Java常用阻塞队列 ArrayBlockingQueue 内部由一个固定长度的数组来实现阻塞队列 /** The queued items */ final Object[] items; /** items index for next take, poll, peek or remove */ int takeIndex; /** items index for next put, offer, or add */ int putIndex; public ArrayBlockingQue

  • Java 阻塞队列BlockingQueue详解

    目录 一. 前言 二. 认识BlockingQueue 三.BlockingQueue的核心方法: 四.常见BlockingQueue 五. 小结 一. 前言 在新增的Concurrent包中,BlockingQueue很好的解决了多线程中,如何高效安全“传输”数据的问题.通过这些高效并且线程安全的队列类,为我们快速搭建高质量的多线程程序带来极大的便利.本文详细介绍了BlockingQueue家庭中的所有成员,包括他们各自的功能以及常见使用场景. 二. 认识BlockingQueue 阻塞队列,

  • 微服务Spring Boot 整合Redis 阻塞队列实现异步秒杀下单思路详解

    目录 引言 一.秒杀优化 - 异步秒杀思路 二.秒杀优化 - 基于Redis完成秒杀资格判断 三.基于阻塞队列完成异步秒杀下单 四.测试程序 五.源码地址 引言 本章节,介绍使用阻塞队列实现秒杀的优化,采用异步秒杀完成下单的优化! 一.秒杀优化 - 异步秒杀思路 当用户发起请求,此时会请求nginx,nginx会访问到tomcat,而tomcat中的程序,会进行串行操作,分成如下几个步骤 查询优惠卷 判断秒杀库存是否足够 查询订单 校验是否是一人一单 扣减库存 创建订单,完成 在以上6个步骤中,

  • PHP基于Redis消息队列实现发布微博的方法

    本文实例讲述了PHP基于Redis消息队列实现发布微博的方法.分享给大家供大家参考,具体如下: phpRedisAdmin :github地址  图形化管理界面 git clone [url]https://github.com/ErikDubbelboer/phpRedisAdmin.git[/url] cd phpRedisAdmin git clone [url]https://github.com/nrk/predis.git[/url] vendor 首先安装上述的Redis图形化管理

随机推荐