【Redis】浅尝Redis Stream做消息队列

前言

众所周知redis数据结构中的list的lpush与rpop可以用于常规消息队列，从集合的最左端写入，最右端弹出消费。并且支持多个生产者与多个消费者并发拿数据，数据只能由一个消费者拿到。

但这个方案并不能保证消费者消费消息后是否成功处理的问题（服务挂掉或处理异常等），机制属于点对点模式不能做广播模式（发布/订阅模式）

Pub/sub

于是redis提供了相应的发布订阅功能，为了解除点对点的强绑定模式引入了Channel管道。

当生产者向管道中发布消息，订阅了该管道的消费者能够同时接收到该消息，而且为了简化订阅多个管道需要显式关注多个名称提供了pattern能力。

通过名称匹配如果接收消息的频道wmyskxz.chat，consumer3也会收到消息。

但这个方案也有很大的诟病就是不会持久化，如果服务挂掉重启数据就全丢弃了，也没有提供ack机制，不保证数据可靠性，不管有没有消费成功发后既忘。

Stream

stream的话整体结构很像kafka的设计思想，提供了consumer group和offset机制，结构上感觉跟kafka的topic差不多，只是没有对应partation副本机制，而是一个追加消息的链表结构。客户端调用XADD时候自动创建stream。每个消息都会持久化并存在唯一的id标识

consumer Group

消费者组的概念跟kafka的消费者概念如出一辙，消费者既可以用`XREAD`命令进行独立消费，也可以多个消费者同时加入一个消费者组。一条消息只能由一个消费者组中的一个消费者消费。这样可以在分布式系统中保证消息的唯一性。

其实这个特性我后来仔细琢磨了一下当时自认为无懈可击的流式图表为了保证分布式系统消息唯一做了redis分布式锁。。。。有点鸡肋，明明消费者组已经保证了数据的唯一性。。。。只能说加锁可以压缩资源成本。。。

last_delivered_id

用于标识消费者组消费在stream上消费位置的游标，每个消费者组都有一个stream内唯一的名称，消费者组不会自动创建，需要用`XGROUP CREATE`显式创建。

pending_ids

每个消费者内部都有一个状态变量。用来表示`已经`被客户端消费但没有ack的消费。目的是为了保证客户端至少消费了消息一次（`atleastonce`）。如果消费者收到了消息处理完了但是没有回复ack，就会导致列表不断增长，如果有很多消费组的话，那么这个列表占用的内存就会放大

curd api

• xadd 追加消息
• xdel 删除消息，这里的删除仅仅是设置了标志位，不影响消息总长度
• xrange 获取消息列表，会自动过滤已经删除的消息
• xlen 消息长度
• del 删除Stream

Pending_ids如何避免消息丢失

在客户端消费者读取Stream消息时，Redis服务器将消息回复给客户端的过程中，客户端突然断开了连接，消息就丢失了。

但是pending_ids里已经保存了发出去的消息ID。待客户端重新连上之后，可以再次收到pending_ids中的消息ID列表。

不过此时xreadgroup的起始消息必须是任意有效的消息ID，一般将参数设为0-0，表示读取所有的pending_ids消息以及自last_delivered_id之后的新消息。

嵌入SpringBoot

redis stream虽然还是有一些弊端，但是相比较而言用kafka之类的消息组件太重，redis用作消息队列已经很合适了。这里简单提一下思路，本质上是提供一个管理消息的一个小功能，定义一个注解用于创建stream管道

创建一个注解类，标注该注解的类必须继承StreamListener<String, ObjectRecord<String, Object>>类且重写onMessage方法。方法上也加这个注解

创建一个config类实现`BeanPostProcessor`接口，重写bean声明周期`postProcessAfterInitialization`和`postProcessBeforeInitialization`方法。该方法会在spring启动流程里的refresh方法加载bean的声明周期中扫描到所有加了注解的bean。

通过线程池挨个创建stream的group组与stream的consumer监听连接，config类记得继承DisposableBean类在destroy方法里把连接关掉免得oom。

注册redis stream api提供的consumer容器

@Bean("listenerContainer")
@DependsOn(value="redisConnectionFactory")
publicStreamMessageListenerContainer<String, ObjectRecord<String, Object>>init() {
StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, Object>>options=StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
         .batchSize(10)
         .serializer(newStringRedisSerializer())
         .executor(newForkJoinPool())
         .pollTimeout(Duration.ofSeconds(3))
         .targetType(Object.class)
         .build();
returnStreamMessageListenerContainer.create(redisConnectionFactory, options);
}

创建消费者

privateSubscriptioncreateSubscription(RedisConnectionFactoryfactory, StreamListenerstreamListener, StringstreamKey, Stringgroup, StringconsumerName) {
StreamOperations<String, String, Object>streamOperations=this.stringRedisTemplate.opsForStream();
if (stringRedisTemplate.hasKey(streamKey)) {
StreamInfo.XInfoGroupsgroups=streamOperations.groups(streamKey);
AtomicReference<Boolean>groupHasKey=newAtomicReference<>(false);
groups.forEach(groupInfo-> {
if (Objects.equals(group, groupInfo.getRaw().get("name"))) {
groupHasKey.set(true);
         }
      });
if (groups.isEmpty() ||!groupHasKey.get()) {
creatGroup(streamKey, group);
      } else {
groups.stream().forEach(g-> {
log.info("XInfoGroups:{}", g);
StreamInfo.XInfoConsumersconsumers=streamOperations.consumers(streamKey, g.groupName());
log.info("XInfoConsumers:{}", consumers);
         });
      }
   } else {
creatGroup(streamKey, group);
   }
StreamOffset<String>streamOffset=StreamOffset.create(streamKey, ReadOffset.lastConsumed());
Consumerconsumer=Consumer.from(group, consumerName);
Subscriptionsubscription=listenerContainer.receive(consumer, streamOffset, streamListener);
listenerContainer.start();
this.containerList.add(listenerContainer);
returnsubscription;
}

具体的代码在gitee上，现在慢慢的扩展了一些别的组件，有兴趣可以去看看，或者有一些开发建议也可以私信下，后续想着可以加一些组件做大一点。

[Gitee 消息流转组件服务](https://gitee.com/atuptown/matrix)