基于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 {
@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);
