1.业务场景
假设在没有专业消息中间件的情况下，又要通过消息队列去解耦。redis是个更好的选择。
2.实现方式
简要说明实现方式，这里只做个大概的概括
- 发布与订阅（缺点：典型的一对一，不支持多个消费者公平消费消息，消息无法持久化，如果出现网络断开、Redis 宕机等，消息就会被丢弃等问题）
- list队列（缺点：没有很好 ACK 机制，没有 ConsumerGroup 消费组，不支持一对多消费等问题）
- stream队列（推荐）官方：https://redis.io/docs/data-types/streams/
3.概念
Redis5.0带来了Stream类型。其实就是Redis对消息队列（MQ，Message Queue）的完善实现。
主要有几个概念：
1.消费者组（Consumer Group）：一个消费组有多个消费者(Consumer), 这些消费者之间是竞争关系。也就是说不会出现重复消费的场景。
2.pending_ids ：消费者(Consumer)的状态变量，作用是维护消费者的未确认的 id。 pending_ids 记录了当前已经被客户端读取的消息，但是还没有 ack (Acknowledge character：确认字符）。
3.last_delivered_id ：游标，每个消费组会有个游标 last_delivered_id，任意一个消费者读取了消息都会使游标 last_delivered_id 往前移动。
4.消息ID: 消息ID的形式是timestampInMillis-sequence，例如1527846880572-5
这里简要贴出Redis中Stream操作的相关指令
其实像代码，都是基于命令的高度封装
消息队列相关命令：
- XADD - 添加消息到末尾
- XTRIM - 对流进行修剪，限制长度
- XDEL - 删除消息
- XLEN - 获取流包含的元素数量，即消息长度
- XRANGE - 获取消息列表，会自动过滤已经删除的消息
- XREVRANGE - 反向获取消息列表，ID 从大到小
- XREAD - 以阻塞或非阻塞方式获取消息列表
消费者组相关命令：
- XGROUP CREATE - 创建消费者组
- XREADGROUP GROUP - 读取消费者组中的消息
- XACK - 将消息标记为"已处理"
- XGROUP SETID - 为消费者组设置新的最后递送消息ID
- XGROUP DELCONSUMER - 删除消费者
- XGROUP DESTROY - 删除消费者组
- XPENDING - 显示待处理消息的相关信息
- XCLAIM - 转移消息的归属权
- XINFO - 查看流和消费者组的相关信息；
- XINFO GROUPS - 打印消费者组的信息；
- XINFO STREAM - 打印流信息
4.代码实现
stream相关配置,这里主要配置消费组和消费者相关信息，以及消息的监听机制
@Slf4j @Configuration public class RedisStreamConfig { @Autowired private MyListener myListener; @Autowired private StringRedisTemplate redisTemplate; /** * 实际生产环境中 我们应该把消费者组等信息 写入配置环境中 */ // @Autowired // private StreamProperty streamProperty; /** * 收到消息后不自动确认，需要用户选择合适的时机确认 * 当某个消息被ACK，PEL列表就会减少 * 如果忘记确认（ACK），则PEL列表会不断增长占用内存 * 如果服务器发生意外，重启连接后将再次收到PEL中的消息ID列表 */ @Bean public Subscription subscription(RedisConnectionFactory factory) { initGroup("mystream", "group1"); // 创建Stream消息监听容器配置 StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> options = StreamMessageListenerContainer .StreamMessageListenerContainerOptions .builder() // 读取超时时间 .pollTimeout(Duration.ofSeconds(3)) // 配置消息类型 .targetType(String.class) // 异常处理器 .errorHandler(t -> log.info("redis listener error", t)) .build(); // 创建Stream消息监听容器 StreamMessageListenerContainer<String, ObjectRecord<String, String>> listenerContainer = StreamMessageListenerContainer.create(factory, options); // 设置消费手动提交配置 Subscription subscription = listenerContainer.receive( // 设置消费者分组和名称 Consumer.from("group1","consumer-1"), // 设置订阅Stream的key和获取偏移量，以及消费处理类 StreamOffset.create("mystream", ReadOffset.lastConsumed()), agendaListener); // 监听容器启动 listenerContainer.start(); return subscription; } /** * 初始化分组 */ private void initGroup(String key, String group) { Boolean aBoolean = redisTemplate.hasKey(key); // 创建不存在的分组 if (Boolean.FALSE.equals(aBoolean)) { redisTemplate.opsForStream().createGroup(key, group); } } }
实现消息的监听
@Slf4j @Component public class MyListener implements StreamListener<String, ObjectRecord<String, String>> { @Autowired private StringRedisTemplate redisTemplate; @Override public void onMessage(ObjectRecord<String, String> record) { try { String value = record.getValue(); log.info("stream name :{}, body:{}", record.getStream(), value); if (StrUtil.isBlank(value)) { return; } // todo 业务逻辑 // 手动确认消息 如果不ack 消息就会进入到pending队列中 这个队列都是维护消费者的未确认的消息 redisTemplate.opsForStream().acknowledge("mystream", "group1", record.getId().getValue()); } catch (Exception e) { log.error("error message:{}", e.getMessage()); } } }
这里说一下消息体类型 Record 官方解释：流中的单个条目，由条目 ID 和实际条目值（通常是字段值对的集合）组成
我们就是可以理解为消息体类型。Record接口，常用的就是
- MapRecord（键值对类型）
- ObjectRecord（对象类型）
测试
@PostMapping("/addStream") public ResponseResult<String> addStream(){ // 这里的消息体都是string类型 ObjectRecord<String, String> record = StreamRecords.objectBacked("1234567").withStreamKey("mystream"); // 这里是消息id，消息id在队列里是唯一的 RecordId recordId = stringRedisTemplate.opsForStream().add(record); // 裁剪队列，因为队列即使被消费者消费后任然不会删除，所以我们队列设定最大容量，也就是上面提到的 XTRIM 命令 Long count = stringRedisTemplate.opsForStream().trim("mystream", 100000); System.out.println("trimCount" + count); if (recordId != null) { // 返回打印消息id return ResponseResult.success(recordId.getValue()); } return ResponseResult.success(); }
基于redisson实现
相关消息监听和消费者配置同上
测试
RStream<Object, Object> stream = redissonClient.getStream("mystream", new SerializationCodec()); StreamAddArgs<Object, Object> entry = StreamAddArgs.entry("a","1"); stream.add(entry);
