基于Redis实现消息队列

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 {@Autowiredprivate MyListener myListener;@Autowiredprivate StringRedisTemplate redisTemplate;/*** 实际生产环境中  我们应该把消费者组等信息  写入配置环境中 */
//    @Autowired
//    private StreamProperty streamProperty;/*** 收到消息后不自动确认，需要用户选择合适的时机确认* 当某个消息被ACK，PEL列表就会减少* 如果忘记确认（ACK），则PEL列表会不断增长占用内存* 如果服务器发生意外，重启连接后将再次收到PEL中的消息ID列表*/@Beanpublic 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>> {@Autowiredprivate StringRedisTemplate redisTemplate;@Overridepublic 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) {// 返回打印消息idreturn 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);