【2023】redis-stream配合spring的data-redis详细使用(包括广播和组接收)

背景：
使用该方式实现，主要是因为项目中有个地方刚好需要异步来实现，而项目又没有配置专业的消息中间件，并且使用的也不是太频繁，就觉得没必要专门安装一个MQ服务了，直接通过现有的redis的stream来实现异步消息接收直接具体的业务逻辑。

一、简介

1、介绍

Redis Stream（Redis Streams）是Redis 5.0版本引入的一种数据结构，用于处理时间序列数据、消息队列和日志流。它提供了高吞吐量、持久性、有序、可扩展的消息传递解决方案。Redis Stream 结构是对传统发布/订阅模式的增强，使你能够更灵活地处理数据流，并提供了以下主要特性：

1. 多生产者和多消费者：多个生产者可以同时向 Stream 中写入消息，而多个消费者可以独立订阅并消费消息。每个消费者可以有不同的消费速率。

2. 消费组：Redis Stream引入了消费者组的概念，多个消费者可以加入同一个消费者组并共同消费消息，这确保了消息在消费时不会被多次处理。

3. 消费者阻塞：消费者可以使用 XREADGROUP 命令以阻塞方式获取消息，只有当有新消息到达时才会被推送给消费者。

4. 消费者自动确认：Redis Stream 支持自动确认消息，消费者可以告诉 Redis 何时确认已经成功处理了一条消息。

5. 多 Stream 支持：你可以创建多个 Stream 来存储不同种类的数据，并分别处理它们。

6. 有序性：消息在 Stream 中按照消息的时间戳有序存储，因此你可以按照消息的顺序读取数据。

7. 持久性存储：Redis Stream 使用内存数据结构，但也支持将数据异步保存到磁盘，以确保数据不会丢失。

2、对比

对比redis的其他几种实现方式来说功能更加全面，支持可持久化和通过ack确认的方式基本实现了消息丢失的问题，当然对比专业的消息队列中间件来说还是有些不足的。
需要看详细对比可以看 🔗redis队列对比 这篇文章

二、整合spring的data-redis实现

1、使用依赖

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-pool2</artifactId><version>2.11.1</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency></dependencies>

2、配置类

2.1、配置RedisTemplate bean

重点是下面这一句，不能用json的序列化类，否则会序列化失败
redisTemplate.setHashValueSerializer(RedisSerializer.string());

	@Beanpublic RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();redisTemplate.setConnectionFactory(connectionFactory);redisTemplate.setKeySerializer(new StringRedisSerializer());redisTemplate.setValueSerializer(new StringRedisSerializer());redisTemplate.setHashKeySerializer(new StringRedisSerializer());// 这个地方不可使用 json 序列化，如果使用的是ObjectRecord传输对象时，可能会有问题，会出现一个 java.lang.IllegalArgumentException: Value must not be null! 错误redisTemplate.setHashValueSerializer(RedisSerializer.string());return redisTemplate;}

2.2、异常类

@Slf4j
public class CustomErrorHandler implements ErrorHandler {@Overridepublic void handleError(Throwable throwable) {log.error("发生了异常",throwable);}
}

3、实体类

该地方使用了两个实体类，主要是用于测试，如果不是指定的同一个类型时，指定的是父类的类型，是否可以正常反序列化接收消息

3.1、User

@Data
@NoArgsConstructor
@AllArgsConstructor
@ToString
public class Book extends User{private String title;private String author;
}

3.2、Book

@Data
@NoArgsConstructor
@AllArgsConstructor
@ToString
public class User implements Serializable {private String name;private Integer age;
}

4、发送消息

发送消息主要是通过redisTemplate.opsForStream().add(record);进行发送到redis中（接收时会分两种方式接收，看后续！）

4.1、RedisStreamUtil工具类

用于实现消息发送、初始化组、key绑定组、清除消费了的消息等方法

• 在第一次发送消息时需要先绑定接收的组和可key，否则在接收时会报不存在该组的异常
• 发送消息后，需要把该条消费了的消息清除掉，否则会一直保持在stream中
  

@Component
@Slf4j
public class RedisStreamUtil {public static final String STREAM_KEY_001 = "stream-001";@Resourceprivate RedisTemplate<String,Object> redisTemplate;/*** 添加记录到流中* @param streamKey* @param t* @param <T>*/public <T> RecordId add(String streamKey,T t){ObjectRecord<String, T> record = StreamRecords.newRecord().in(streamKey)  //key.ofObject(t) //消息数据.withId(RecordId.autoGenerate());
//        发送消息RecordId recordId = redisTemplate.opsForStream().add(record);log.info("添加成功，返回的record-id[{}]",recordId);return recordId;}/*** 用来创建绑定流和组*/public void addGroup(String key, String groupName){redisTemplate.opsForStream().createGroup(key,groupName);}/*** 用来判断key是否存在*/public boolean hasKey(String key){if(key==null){return false;}else{return redisTemplate.hasKey(key);}}/*** 用来删除掉消费了的消息*/public void delField(String key,RecordId recordIds){redisTemplate.opsForStream().delete(key,recordIds);}/*** 用来初始化 实现绑定key和消费组*/public void initStream(String key, String group){//判断key是否存在，如果不存在则创建boolean hasKey = hasKey(key);if(!hasKey){Map<String,Object> map = new HashMap<>();map.put("key","value");RecordId recordId = add(key, map);addGroup(key,group);   //第一次初始化时需要把Stream和group绑定，delField(key,recordId);  //清除掉该条无用数据log.info("stream:{}-group:{} initialize success",key,group);}}public String getStreamKey001(){return STREAM_KEY_001;}
}

4.2、通过延时队列线程池模拟发送消息

• 该方法里通过模拟延时5秒后，每隔3秒发送一条数据，发送10条后关闭线程池

/*** 在spring初始化时执行，定时发送消息到stream中，用于模拟发送消息*/
//@Component
public class StreamMessageRunner implements ApplicationRunner {@Resourceprivate RedisStreamUtil redisStreamUtil;@Overridepublic void run(ApplicationArguments args) throws Exception {ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);AtomicInteger integer = new AtomicInteger(0);//使用延时队列线程池模拟发送数据消息pool.scheduleAtFixedRate(()->{User zhangsan = new User("zhangsan"+integer.get(), 1 + integer.get());RecordId recordId = redisStreamUtil.add(redisStreamUtil.getStreamKey001(), zhangsan);integer.getAndIncrement();
//需要把消费了的消息清除掉，否则会一直保持在stream中，会被重复消费redisStreamUtil.delField(redisStreamUtil.getStreamKey001(),recordId);if (integer.get()>10){System.out.println("---------退出发送消息--------");pool.shutdown();}},5,3, TimeUnit.SECONDS);}
}

4.3、通过http主动发送消息

• 通过分别发送父类和子类比对查看不同效果

@RestController
@RequestMapping("/index")
public class index {@Resourceprivate RedisStreamUtil redisStreamUtil;/*** 父类*/@GetMapping("/login")public String login(User user){RecordId recordId = redisStreamUtil.add(redisStreamUtil.getStreamKey001(), user);redisStreamUtil.delField(redisStreamUtil.getStreamKey001(),recordId);return "成功！";}/*** 子类*/@GetMapping("/login2")public String login(Book book){RecordId recordId = redisStreamUtil.add(redisStreamUtil.getStreamKey001(), book);redisStreamUtil.delField(redisStreamUtil.getStreamKey001(),recordId);return "成功！";}
}

5、🌟消息接收

5.1、不绑定消费组—可以实现广播📢效果

节点消费者不绑定消费组，直接和stream进行绑定，即可实现广播的效果，每次有消息发送到该指定节点的stream，都可以接收到。

如下图：有消息发送到redis Stream 里面绑定的A0到B2全部可以接收到这条消息

方式1：主动读取

通过redisTemplate.opsForStream().read()方法主动去stream中读取消息

/*** 独立消费者---可以读取到该key的全部消息*/
@Component
@Slf4j
public class XreadNonBlockConsumer01 implements InitializingBean, DisposableBean {private ThreadPoolExecutor threadPoolExecutor;@Resourceprivate RedisTemplate<String,Object> redisTemplate;private volatile boolean stop = false;/*** 初始化bean时执行，以轮询的方式主动去stream的指定key里读取消息* @throws Exception*/@Overridepublic void afterPropertiesSet() throws Exception {// 初始化线程池threadPoolExecutor = new ThreadPoolExecutor(3, 5, 0, TimeUnit.SECONDS,new LinkedBlockingDeque<>(), r -> {Thread thread = new Thread(r);thread.setDaemon(true);thread.setName("xread-nonblock-01");return thread;});StreamReadOptions options = StreamReadOptions.empty()
//                如果没有数据，则阻塞1s，阻塞时间需要小于·spring.redis.timeout·.block(Duration.ofMillis(1000))
//                一直阻塞直到获取数据，可能会报超时异常
//                .block(Duration.ofMillis(0))
//                一次获取10条数据.count(10);StringBuilder readBuilder = new StringBuilder("0-0");threadPoolExecutor.execute(()->{while (!stop){//主动到redis的stream中去读取，options设置了每读取一次阻塞一秒List<ObjectRecord<String, User>> objectRecords = redisTemplate.opsForStream().read(User.class, options,StreamOffset.create(RedisStreamUtil.STREAM_KEY_001, ReadOffset.from(readBuilder.toString())));if (CollectionUtils.isEmpty(objectRecords)){log.warn("没有读取到数据");continue;}objectRecords.stream().forEach(objectRecord->{log.info("获取到的数据信息 id:[{}] book:[{}]", objectRecord.getId(), objectRecord.getValue());readBuilder.setLength(0);readBuilder.append(objectRecord.getId());});}});}/*** 在销毁bean时把线程池关闭* @throws Exception*/@Overridepublic void destroy() throws Exception {stop = true;threadPoolExecutor.shutdown();threadPoolExecutor.awaitTermination(3,TimeUnit.SECONDS);}}

测试日志

🍉方式2：通过监听器监听是否有新消息

具体代码和分组的是一样的，只不过不指定组而已，就合并在下面写了
主要通过StreamMessageListenerContainer这个监听器类实现。

主要通过下面这一句：

container.receive(StreamOffset.fromStart(RedisStreamUtil.STREAM_KEY_001), new MonitorStreamListener("独立消费", null, null));

5.2、指定消费组----实现一个组内只有一个成员可以接收到

进行分组之后，一个组内，只会有一个成员可以读到消息，具体如下图，当然在使用时也可以绑定多个组，每个组接收不听的消息。下面方式就相当于mq了，交换机，队列和路由键的关系

5.2.1、配置类

下面代码具体流程时先创建一个线程池；然后在配置消息监听容器，最后在把用于接收消息的监听器放入到监听容器中去，最后把这个侦听容器注入到bean去

@Configuration
public class RedisStreamConfiguration {@Resourceprivate RedisStreamUtil redisStreamUtil;@Resourceprivate RedisConnectionFactory redisConnectionFactory;@Bean(initMethod = "start",destroyMethod = "stop")public StreamMessageListenerContainer<String, ObjectRecord<String,User>> streamMessageListenerContainer(){AtomicInteger index = new AtomicInteger(1);
//        获取本机线程数int processors = Runtime.getRuntime().availableProcessors();ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0,TimeUnit.SECONDS,new ArrayBlockingQueue<>(50), (r) -> {Thread thread = new Thread(r);thread.setName("async-stream-consumer-" + index.getAndIncrement());thread.setDaemon(true);return thread;}, new ThreadPoolExecutor.CallerRunsPolicy());//        消息监听容器，不能在外部实现。创建后，StreamMessageListenerContainer可以订阅Redis流并使用传入的消息StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String,ObjectRecord<String,User>> options =StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()// 一次最多获取多少条消息.batchSize(10)// 运行 Stream 的 poll task.executor(executor)// Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小.pollTimeout(Duration.ofSeconds(1))// ObjectRecord 时，将 对象的 filed 和 value 转换成一个 Map 比如：将Book对象转换成map
//                        .objectMapper(new ObjectHashMapper())// 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理.errorHandler(new CustomErrorHandler())// 将发送到Stream中的Record转换成ObjectRecord，转换成具体的类型是这个地方指定的类型.targetType(User.class).build();StreamMessageListenerContainer<String, ObjectRecord<String, User>> container = StreamMessageListenerContainer.create(redisConnectionFactory, options);//        初始化-绑定key和消费组redisStreamUtil.initStream(RedisStreamUtil.STREAM_KEY_001,"group-a");//        不绑定消费组，独立消费container.receive(StreamOffset.fromStart(RedisStreamUtil.STREAM_KEY_001), new MonitorStreamListener("独立消费", null, null));// 消费组A,不自动ack// 从消费组中没有分配给消费者的消息开始消费
//        container.receive(Consumer.from("group-a","consumer-a"),
//                StreamOffset.create(RedisStreamUtil.STREAM_KEY_001, ReadOffset.lastConsumed()),new MonitorStreamListener("消费者组A","group-a", "consumer-a"));//        自动ackcontainer.receiveAutoAck(Consumer.from("group-a","consumer-b"),StreamOffset.create(RedisStreamUtil.STREAM_KEY_001, ReadOffset.lastConsumed()),new MonitorStreamListener("消费者组B","group-a", "consumer-b"));return container;}
}

重要代码解析：

1. .targetType(User.class) ：在配置监听容器时，用于指定类型，不指定时默认是string类型，如果你传入的不是string机需要指定；如果配置的是父类的，也可以接收子类的消息，进行转换。但如果是配置的Object类型，接收时就会为路径，不能正常得到传入的对象（不知道为什么，有研究懂的可以解答一下）
2. redisStreamUtil.initStream(RedisStreamUtil.STREAM_KEY_001,"group-a")：在第一次生成时，需要把消费组绑定该stream的key，否则会报错，具体内部执行逻辑可以看initStream()方法（或者自己手动通过命令到redis去绑定：xgroup create stream-001 group-a $）stream-001(key) group-a(消费组)
3. container.receive(StreamOffset.fromStart(RedisStreamUtil.STREAM_KEY_001), new MonitorStreamListener("独立消费", null, null)) ：该句是不绑定消费组，也就是广播的方式监听该key中的所有消息（和上面的区别是，该方式是被动的监听消息）
4. 🌟container.receiveAutoAck(Consumer.from("group-a","consumer-b") ,StreamOffset.create(RedisStreamUtil.STREAM_KEY_001, ReadOffset.lastConsumed()),new MonitorStreamListener("消费者组B","group-a", "consumer-b"))：就是通过该句代码实现分组监听消息的，绑定了消费组和消费者的名字，以及监听器类。然后使用的自动ack的方式回复stream确认接收到了消息(或者通过手动ack的方式返回stream接收到了消息，否则会重复发送)，

5.2.2、监听器

用于接收消息然后实现具体业务代码

@Slf4j
public class MonitorStreamListener <T> implements StreamListener<String, ObjectRecord<String,T>> {/*** 消费者类型：独立消费、消费组消费*/private String consumerType;/*** 消费组*/private String group;/*** 消费组中的某个消费者*/private String consumerName;public MonitorStreamListener(String consumerType, String group, String consumerName) {this.consumerType = consumerType;this.group = group;this.consumerName = consumerName;}@Overridepublic void onMessage(ObjectRecord<String, T> message) {log.info("接受到来自redis的消息");String stream = message.getStream();RecordId id = message.getId();User value = (User) message.getValue();value.getName();//        执行具体的接收到消息的业务逻辑if (StringUtils.isEmpty(group)) {log.info("[{}]: 接收到一个消息 stream:[{}],id:[{}],value:[{}]", consumerType, stream, id, value);} else {log.info("[{}] group:[{}] consumerName:[{}] 接收到一个消息 stream:[{}],id:[{}],value:[{}]", consumerType,group, consumerName, stream, id, value);}// 当是消费组消费时，如果不是自动ack，则需要在这个地方手动ack
//        redisTemplate.opsForStream()
//                 .acknowledge("key","group","recordId");}
}

通过延时队列发送到消息—测试结果：

通过http发送User到子类Book对象数据测试结果

结果：也是可以正常接受到的

三、完整代码

🪟完整代码仓库地址

四、引用

https://juejin.cn/post/7029302992364896270#heading-0
https://juejin.cn/post/6844904125822435341？searchId=202310141054532F9807A1000F6680C0DF#heading-1