redis实现消息队列的几种方式

一、了解

众所周知，redis是我们日常开发过程中使用最多的非关系型数据库，也是消息中间件。实际上除了常用的rabbitmq、rocketmq、kafka消息队列（大家自己下去研究吧~模式都是通用的），我们也能使用redis实现消息队列。因为其他中间件可能更适用于大型/企业级项目，在咱们项目前期不需要这么多的数据，redis跟我们也是高度集成的。这里就简化了技术栈。

二、常用的几种使用redis实现的消息队列方式

1、List数据结构

Redis列表是简单的字符串列表，按照插入顺序排序。你可以添加一个元素到列表的头部（左边）或者尾部（右边）一个列表最多可以包含 232 - 1 个元素 (4294967295, 每个列表超过40亿个元素)。
这里的列表大家可以想想为一个横着的通道，假设我现在往右边插入第一条数据，这个元素就会被放在最左边，接着再放入第二条数据，它就会在左边第二条，以此类推…插入了100条数据。 假设这个时候我要取出第一条，我就从最左边取就好。
这就变相实现了有序消息队列。具体实现大家自己研究
优点：操作方便，可以有序的取出自己插入的数据
缺点：不能进行实时消费，没有消费者

2、pub/sub 订阅消费模式

这就是传统的生产者->队列->消费者的模式。生产者的消息所有订阅者都能收到。

优点：实现了发布订阅模式，可以实时进行消费
缺点：没有消息持久化，在系统崩溃、宕机的时候；消息会丢失

3、sorted set有序集合Redis

有序集合和集合一样也是 string 类型元素的集合,且不允许重复的成员。不同的是每一个元素都会关联一个double分数，redis就是通过分数为集合中的成员进行从大到小的排列。
有序集合的成员是唯一的，但是score是可以重复的。
生成消息直接往s-set中插入数据，将score设置为接收到数据的13位时间戳；需要使用的时候再根据score大小有序取出来就行了。

看到这里是不是大家能想到，既然每条消息都带有时间，那我是不是可以顺手实现延迟队列。
这里只需要将score设置为 接受消息的时间戳+延迟时间 。我在使用的时候获取当天时间戳的数据，这样就实现了延迟消息队列。

优点：操作方便，可以实现延迟队列
缺点：不能实时进行消费

4、stream流 （redis5.0版本以上才有 重点讲）

Redis Stream 提供了消息的持久化和主备复制功能，可以让任何客户端访问任何时刻的数据，并且能记住每一个客户端的访问位置，还能保证消息不丢失。
Redis Stream 的结构如下所示，它有一个消息链表，将所有加入的消息都串起来，每个消息都有一个唯一的 ID 和对应的内容：

每个stream流都有自己的名称，它是redis的key，也可以理解为队列名称。
Consumer Group ：消费组，使用 XGROUP CREATE 命令创建，一个消费组有多个消费者(Consumer)。
last_delivered_id ：游标，每个消费组会有个游标 last_delivered_id，任意一个消费者读取了消息都会使游标 last_delivered_id 往前移动。
pending_ids ：消费者(Consumer)的状态变量，作用是维护消费者的未确认的 id。 pending_ids 记录了当前已经被客户端读取的消息，但是还没有 ack (Acknowledge character：确认字符）。

stream常用命令

• XADD 定义stream流，写入消息体

XADD mystream * field1 A field2 B field3 C field4 D
mystream：自定义流名称
*：由redis生成流的id（也可以自定义，但是得保证自增唯一）
field1-A \field2-B\field3-C ：保存的消息体，key-value形式-- 举例
redis> XADD mystream * name Sara surname OConnor
"1601372323627-0"

• XDEL 删除消息

> XADD mystream * a 1
1538561698944-0
> XADD mystream * b 2
1538561700640-0
> XADD mystream * c 3
1538561701744-0
> XDEL mystream 1538561700640-0
(integer) 1
127.0.0.1:6379> XRANGE mystream - +
1) 1) 1538561698944-02) 1) "a"2) "1"
2) 1) 1538561701744-02) 1) "c"2) "3"

• XRANGE 获取消息队列数据

XRANGE key start end [COUNT count]key:strem流名称
start：开始值，- 表示最小值
end：结束值，+ 表示最大值-- 举例：
redis> XRANGE mystream - + 2
从mystrem全部数据中取出两条数据redis> XRANGE mystream + - 1
从mystream倒叙取一条数据

• XREVRANGE 自动过滤已删除的消息

redis> XADD writers * name Virginia surname Woolf
"1601372731458-0"
redis> XADD writers * name Jane surname Austen
"1601372731459-0"
redis> XADD writers * name Toni surname Morrison
"1601372731459-1"
redis> XADD writers * name Agatha surname Christie
"1601372731459-2"
redis> XADD writers * name Ngozi surname Adichie
"1601372731459-3"
redis> XLEN writers
(integer) 5
redis> XREVRANGE writers + - COUNT 1
1) 1) "1601372731459-3"2) 1) "name"2) "Ngozi"3) "surname"4) "Adichie"
redis>

• XREAD 阻塞或者非阻塞获取消息

# 从 Stream 头部读取两条消息
> XREAD COUNT 2 STREAMS mystream writers 0-0 0-0
1) 1) "mystream"2) 1) 1) 1526984818136-02) 1) "duration"2) "1532"3) "event-id"4) "5"5) "user-id"6) "7782813"2) 1) 1526999352406-02) 1) "duration"2) "812"3) "event-id"4) "9"5) "user-id"6) "388234"
2) 1) "writers"2) 1) 1) 1526985676425-02) 1) "name"2) "Virginia"3) "surname"4) "Woolf"2) 1) 1526985685298-02) 1) "name"2) "Jane"3) "surname"4) "Austen"
count ：数量
milliseconds ：可选，阻塞毫秒数，没有设置就是非阻塞模式
key ：队列名
id ：消息 ID

• XGROUP CREATE 创建消费者组

XGROUP [CREATE key groupname id-or-$] [SETID key groupname id-or-$] [DESTROY key groupname] [DELCONSUMER key groupname consumername]key ：队列名称，如果不存在就创建
groupname ：组名。
$ ： 表示从尾部开始消费，只接受新消息，当前 Stream 消息会全部忽略从头开始消费:
XGROUP CREATE mystream consumer-group-name 0-0  从尾部开始消费:
XGROUP CREATE mystream consumer-group-name $

以上就是常用的steam流的命令，大家下来自己测试，练习。

三、springboot整合redis stream流

java中提供了连接redis的客户端，jedis和lettuce、redistemplate；RedisTemplate 是 Spring Data Redis 提供的一个高级抽象层，封装了 Jedis 或 Lettuce 等底层客户端。
它提供了丰富的功能，如序列化、事务支持、键过期等。这里主要讲主流的redistemplate整合，大家以后能直接使用。

实时消费

实时消费顾名思义，生产者发送消息，消费者立马进行消费逻辑处理。

• RedisStreamUtils工具类，方便后续进行stream操作没根据自己项目需求来定义

@Configuration
@SuppressWarnings("all")
public class RedisStreamUtils {@Resourceprivate RedisTemplate<String, Object> redisTemplate;/*** 创建消费组** @param streamKey   键名称* @param group 组名称* @return {@link String}*/public String createGroup(String streamKey, String group) {return redisTemplate.opsForStream().createGroup(streamKey, group);}/*** 获取消费者信息** @param streamKey   键名称* @param group 组名称* @return {@link StreamInfo.XInfoConsumers}*/public StreamInfo.XInfoConsumers queryConsumers(String streamKey, String group) {return redisTemplate.opsForStream().consumers(streamKey, group);}/*** 查询组信息** @param streamKey 键名称* @return*/public StreamInfo.XInfoGroups queryGroups(String streamKey) {return redisTemplate.opsForStream().groups(streamKey);}// 添加Map消息public String addMap(String streamKey, Map<String, Object> value) {return Objects.requireNonNull(redisTemplate.opsForStream().add(streamKey, value)).getValue();}// 读取消息public List<MapRecord<String, Object, Object>> read(String streamKey) {return redisTemplate.opsForStream().read(StreamOffset.fromStart(streamKey));}// 确认消费public Long ack(String streamKey, String group, String... recordIds) {return redisTemplate.opsForStream().acknowledge(streamKey, group, recordIds);}// 删除消息。当一个节点的所有消息都被删除，那么该节点会自动销毁public Long del(String key, String... recordIds) {return redisTemplate.opsForStream().delete(key, recordIds);}// 判断是否存在keypublic boolean hasKey(String key) {Boolean aBoolean = redisTemplate.hasKey(key);return aBoolean != null && aBoolean;}}

• RedisConfig配置文件

@Configuration
@Slf4j
@RequiredArgsConstructor
public class RedisConfig {private final RedisStreamUtils redisStreamUtil;private final Environment environment;//消费者处理消息配置@Beanpublic Subscription subscription(RedisConnectionFactory factory) {AtomicInteger index = new AtomicInteger(1);//获取系统处理器数量 创建线程池，开启守护线程int processors = Runtime.getRuntime().availableProcessors();ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,new LinkedBlockingDeque<>(), r -> {Thread thread = new Thread(r);thread.setName("async-stream-consumer-" + index.getAndIncrement());thread.setDaemon(true);return thread;});//流消息监听容器参数设置 StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()// 一次最多获取多少条消息.batchSize(5)//执行线程池.executor(executor)//阻塞消息读取（延迟消息）.pollTimeout(Duration.ofSeconds(1))//异常处理.errorHandler(throwable -> {log.error("[MQ handler exception]", throwable);throwable.printStackTrace();}).build();//通过redis连接工厂，创建流消息监听容器var listenerContainer = StreamMessageListenerContainer.create(factory, options);//初始化流和消费者处理配置//初始化流和消费者处理配置Subscription subscription = initStreamAndConsumer(listenerContainer);//开启监听容器listenerContainer.start();return subscription;}private Subscription initStreamAndConsumer(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer){//↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓//这一部分可以不用配置，可以根据自己的实际情况配置//该key和group可根据需求自定义配置String streamName = "mystream";String groupname = "mygroup";initStream(streamName, groupname);// 手动ask消息//消费者处理完消息之后，会进行确认；这里有一个pending状态会变成已处理Subscription subscription = listenerContainer.receive(Consumer.from(groupname, "zhuyazhou"),StreamOffset.create(streamName, ReadOffset.lastConsumed()), new RedisConsumer(redisStreamUtil));// 自动ask消息/* Subscription subscription = listenerContainer.receiveAutoAck(Consumer.from(redisMqGroup.getName(), redisMqGroup.getConsumers()[0]),StreamOffset.create(streamName, ReadOffset.lastConsumed()), new ReportReadMqListener());*///这一部分可以不用配置，可以根据自己的实际情况配置//↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑return subscription;}private void initStream(String key, String group) {boolean hasKey = redisStreamUtil.hasKey(key);if (!hasKey) {Map<String, Object> map = new HashMap<>(1);map.put("field", "value");//创建主题String result = redisStreamUtil.addMap(key, map);//创建消费组redisStreamUtil.createGroup(key, group);//将初始化的值删除掉redisStreamUtil.del(key, result);log.info("stream:{}-group:{} initialize success", key, group);}}
}

大家这里可以想一想，这种写法是不是符合生产过程中的创建队列/消费者的逻辑，是不是不方便。能不能在我需要的时候直接调用方法去创建？？？假设现在我新增了一个业务需求，需要用不同的业务逻辑去处理，而且我希望定制不同的消费者应答模式，这个时候就需要一个通用方法去实现，这里我是这样做的。还是在工具类中

创建redis流消息监听容器
主要参数 ：定义线程池、一次最大获取消息数、超时重新获取、异常处理

  @Beanpublic StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(RedisConnectionFactory factory) {log.info("redis ip:{},port:{}",environment.getProperty("spring.data.redis.host"),environment.getProperty("spring.data.redis.port"));AtomicInteger index = new AtomicInteger(1);int processors = Runtime.getRuntime().availableProcessors();ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,new LinkedBlockingDeque<>(), r -> {Thread thread = new Thread(r);thread.setName("async-stream-consumer-" + index.getAndIncrement());thread.setDaemon(true);return thread;});StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()// 一次最多获取多少条消息.batchSize(5).executor(executor).pollTimeout(Duration.ofSeconds(3)).errorHandler(throwable -> {log.error("[MQ handler exception]", throwable);throwable.printStackTrace();}).build();return StreamMessageListenerContainer.create(factory, options);}//业务需求调用此方法即可
public void addNewStreamAndSubscribe(String streamName, String groupName, String consumerId, StreamListener listener) {initStream(streamName, groupName);subscribeToStream(streamName, groupName, Consumer.from(groupName, consumerId), listener);}public void addNewStreamAndSubscribe(String streamName, String groupName, String consumerId, RedisConsumer listener,Map<String,Object> recodMap) {initStream(streamName, groupName);subscribeToStream(streamName, groupName, Consumer.from(groupName, consumerId), listener);addMap(streamName, recodMap);}private void subscribeToStream(String streamName, String groupName, Consumer consumer, StreamListener listener) {StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = streamMessageListenerContainer(redisConnectionFactory);Subscription subscription = container.receive(Consumer.from(groupName, consumer.getName()),StreamOffset.create(streamName, ReadOffset.lastConsumed()), listener);//开始消息容器监听container.start();log.info("Subscribed to stream: {} with group: {} and consumer: {}", streamName, groupName, consumer);}

streamMessageListenerContainer中的 .batchSize(1) 设置需要着重说一下。意思是在消费者在监听到数据的时候，一次从redis中取出的多少条数据，假设我设置1，就意味着我的监听器会redis中取出1条未消费的数据，随后进入消费者逻辑，处理完毕之后返回；继续由监听器读取1条数据，在进入消费者逻辑；这个值设置得越小消息处理数据越快，但是也会增加redis链接的资源。
较大的 batchSize 可以减少与 Redis 服务器的交互次数，降低网络通信开销，提高处理效率。
较小的 batchSize 适用于需要低延迟处理的场景，但会增加网络通信开销和 CPU 使用率。

• RedisConsumer消费者

@Component("RedisConsumer")
@RequiredArgsConstructor
@Slf4j
public class RedisConsumer implements StreamListener<String, MapRecord<String,String,String>> {private final RedisStreamUtils redisStreamUtils;@Overridepublic void onMessage(MapRecord<String, String, String> message) {try {log.info("RedisConsumer1获取到了消息：{}",message);String streamKey = message.getStream();RecordId recordId = message.getId();Map<String, String> value = message.getValue();//获取这个流下 所有的消费者组StreamInfo.XInfoGroups xInfoGroups = redisStreamUtils.queryGroups(streamKey);//处理逻辑//↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓log.info("【streamKey】= {}，【recordId】= {},【msg】= {}",streamKey,recordId, value);//手动确认ack消息，并删除已处理的消息//我这里使用手动xInfoGroups.forEach(xInfoGroup -> redisStreamUtils.ack(streamKey, xInfoGroup.groupName(), recordId.getValue()));//自动确认消息 ---------自己下来研究//↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑//根据业务场景来看是否需要删除消息
//        redisStreamUtils.del(streamKey, recordId.getValue());} catch (Exception e) {throw new ServiceException("消费异常");}}
}

• RedisConsumer2消费者

@Component("RedisConsumer2")
@RequiredArgsConstructor
@Slf4j
public class RedisConsumer2 implements StreamListener<String, MapRecord<String,String,String>> {private final RedisStreamUtils redisStreamUtils;@Overridepublic void onMessage(MapRecord<String, String, String> message) {try {log.info("RedisConsumer2获取到了消息：{}",message);String streamKey = message.getStream();RecordId recordId = message.getId();Map<String, String> value = message.getValue();//获取这个流下 所有的消费者组StreamInfo.XInfoGroups xInfoGroups = redisStreamUtils.queryGroups(streamKey);//处理逻辑//↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓log.info("【streamKey】= {}，【recordId】= {},【msg】= {}",streamKey,recordId, value);//↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑//手动确认ack消息，并删除已处理的消息xInfoGroups.forEach(xInfoGroup -> redisStreamUtils.ack(streamKey, xInfoGroup.groupName(), recordId.getValue()));
//        redisStreamUtils.del(streamKey, recordId.getValue());} catch (Exception e) {throw new ServiceException("消费异常");}}
}

• RedisStreamcontroller模拟测试

@RequestMapping(value = "/redisStream")
@RestController
@RequiredArgsConstructor
@Slf4j
@SuppressWarnings("all")
public class RedisStreamController {private final RedisStreamUtils redisStreamUtils;private final RedisConsumer redisConsumer;private final RedisTemplate redisTemplate;private final ApplicationContext applicationContext;@GetMapping(value = "/addNewStreamAndSubscribe")public ResultVO addNewStreamAndSubscribe(@RequestParam("streamKey") String streamKey,@RequestParam("groupName") String groupName,@RequestParam("consumer")String consumer,@RequestParam("consumerClass") String consumerClass){try {// 获取实现类的实例StreamListener consumerInstance = (StreamListener) applicationContext.getBean(consumerClass);redisStreamUtils.addNewStreamAndSubscribe(streamKey, groupName, consumer,consumerInstance );} catch (Exception e) {throw new RuntimeException(e);}return ResultVO.success();}@GetMapping(value = "/addMap")public ResultVO addMap(@RequestParam("streamKey") String streamKey,@RequestParam("key")String key,@RequestParam("value")String value) {HashMap<String, Object> objectObjectHashMap = new HashMap<>();objectObjectHashMap.put(key,value);redisStreamUtils.addMap(streamKey,objectObjectHashMap);return ResultVO.success();}@GetMapping(value = "/getGroup")public ResultVO getGroup(@RequestParam("streamKey") String streamKey,@RequestParam("groupName") String groupName) {boolean b = redisStreamUtils.hasKey(streamKey);if(b){StreamInfo.XInfoGroups xInfoGroups = redisStreamUtils.queryGroups(streamKey);List<Object> list = new ArrayList<>();for (StreamInfo.XInfoGroup xInfoGroup : xInfoGroups) {StreamInfo.XInfoConsumers xInfoConsumers = null;if(StrUtil.isNotEmpty(groupName)){xInfoConsumers = redisStreamUtils.queryConsumers(streamKey, groupName);for (StreamInfo.XInfoConsumer xInfoConsumer : xInfoConsumers) {log.info("group:{},pending:{},consumerCount:{},consumerName:{},lastDeliveryId:{}",xInfoGroup.groupName(),xInfoGroup.pendingCount(),xInfoGroup.consumerCount(),xInfoConsumer.consumerName(),xInfoGroup.lastDeliveredId());}}}}else{log.info("streamKey不存在：{}",streamKey);return ResultVO.error("streamKey不存在");}return ResultVO.success();}@GetMapping(value = "/delStream")public ResultVO delStream(@RequestParam("streamKey") String streamKey){redisTemplate.delete(streamKey);return ResultVO.success();}@GetMapping(value = "/readMsg")public ResultVO readMsg(@RequestParam("streamKey") String streamKey,@RequestParam("groupName") String groupName,@RequestParam("consumer") String consumer){// 读取消息，每次读取最多 5 条List read = redisTemplate.opsForStream().read(Consumer.from(groupName, consumer),StreamReadOptions.empty().count(10).block(Duration.ofSeconds(1)),StreamOffset.create(streamKey, ReadOffset.lastConsumed()));return ResultVO.success(JSON.toJSONString(read));}

项目启动

调用/addNewStreamAndSubscribe接口

• 创建流、监听容器
• 消费者绑定流+消费者逻辑处理类
• 接收生产者消息方式（最新、偏移量）
• 开启消息容器监听

调用/addMap接口，发送消息

如果只有一个消费者，那么当消费者出现异常的时候，直到服务恢复，会从上一次消费的数据开始进行消费。

假设现在消费者组有两个消费者，都绑定了同一个消息流，这个时候发送消息就是轮询访问。
RedisConsumer1获取到了消息
RedisConsumer2获取到了消息
RedisConsumer1获取到了消息
RedisConsumer2获取到了消息
…
如果consumer1出现了异常，这个时候consumer2会正常消费所有的数据。
stream本身就支持持久化数据，也是dbs和aof两种。不用担心数据丢失。