SpringBoot使用Redis Stream实现轻量消息队列的示例代码
更新时间：2024年08月23日 09:19:32 作者：少年酱105974
Redis Stream 是 Redis 5.0 引入的一种数据结构,用于处理日志类型的数据,它提供了高效、可靠的方式来处理和存储时间序列数据,如事件、消息等,本文介绍了SpringBoot使用Redis Stream实现轻量消息队列,需要的朋友可以参考下
引言
Redis Stream 是 Redis 5.0 引入的一种数据结构，用于处理日志类型的数据。它提供了高效、可靠的方式来处理和存储时间序列数据，如事件、消息等。其设计灵感源于 Kafka 和类似的消息队列系统，且完全集成在 Redis 中，利用了 Redis 的高性能和持久化特性。
依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
说明：此部分定义了 Redis 相关的依赖，确保项目能够引入并使用 Spring Boot 提供的 Redis 启动器。
RedisTemplate 配置
package com.mjg.config;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
@Configuration
public class RedisConfig {
@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
RedisTemplate<String, Object> template = new RedisTemplate<>();
template.setConnectionFactory(connectionFactory);
Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
// om.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);
// 注册 Java 8 日期时间模块
om.registerModule(new JavaTimeModule());
om.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
om.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
jackson2JsonRedisSerializer.serialize(om);
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
// key 采用 String 的序列化方式
template.setKeySerializer(stringRedisSerializer);
// hash 的 key 也采用 String 的序列化方式
template.setHashKeySerializer(stringRedisSerializer);
// value 序列化方式采用 jackson
template.setValueSerializer(jackson2JsonRedisSerializer);
// hash 的 value 序列化方式采用 jackson
template.setHashValueSerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
}
}
说明：此配置类用于设置 RedisTemplate 的序列化方式，以满足不同数据类型的存储和读取需求。
RedisStreamConfig
package com.mjg.config;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.util.StrUtil;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisServerCommands;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;
import java.net.InetAddress;
import java.time.Duration;
import java.util.Properties;
@Slf4j
@RequiredArgsConstructor
@Configuration
public class RedisStreamConfig implements InitializingBean, DisposableBean {
private final RedisTemplate<String, Object> redisTemplate;
public static String streamName = "user-event-stream";
public static String userEventGroup = "user-event-group";
private final ThreadPoolTaskExecutor threadPoolTaskExecutor;
/**
* 消息侦听器容器，用于监听 Redis Stream 中的消息
*
* @param connectionFactory Redis 连接工厂，用于创建 Redis 连接
* @param messageConsumer 消息消费者，用于处理接收到的消息
* @return 返回 {@link StreamMessageListenerContainer}<{@link String}, {@link ObjectRecord}<{@link String}, {@link String}>> 类型的消息侦听器容器
*/
@Bean
public StreamMessageListenerContainer<String, ObjectRecord<String, String>> messageListenerContainer(RedisConnectionFactory connectionFactory, MessageConsumer messageConsumer) {
StreamMessageListenerContainer<String, ObjectRecord<String, String>> listenerContainer = streamContainer(streamName, connectionFactory, messageConsumer);
listenerContainer.start();
return listenerContainer;
}
/**
* 创建一个流容器，用于监听 Redis Stream 中的数据
*
* @param streamName Redis Stream 的名称
* @param connectionFactory Redis 连接工厂
* @param streamListener 绑定的监听类
* @return 返回 StreamMessageListenerContainer 对象
*/
@SneakyThrows
private StreamMessageListenerContainer<String, ObjectRecord<String, String>> streamContainer(String streamName, RedisConnectionFactory connectionFactory, StreamListener<String, ObjectRecord<String, String>> streamListener) {
StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> options =
StreamMessageListenerContainer.StreamMessageListenerContainerOptions
.builder()
.pollTimeout(Duration.ofSeconds(5)) // 拉取消息超时时间
.batchSize(10) // 批量抓取消息
.targetType(String.class) // 传递的数据类型
.executor(threadPoolTaskExecutor)
.build();
StreamMessageListenerContainer<String, ObjectRecord<String, String>> container = StreamMessageListenerContainer
.create(connectionFactory, options);
// 指定消费最新的消息
StreamOffset<String> offset = StreamOffset.create(streamName, ReadOffset.lastConsumed());
// 创建消费者
StreamMessageListenerContainer.StreamReadRequest<String> streamReadRequest = buildStreamReadRequest(offset, streamListener);
// 指定消费者对象
container.register(streamReadRequest, streamListener);
return container;
}
/**
* 生成流读取请求
*
* @param offset 偏移量，用于指定从 Redis Stream 中的哪个位置开始读取消息
* @param streamListener 流侦听器，用于处理接收到的消息
* @return 返回一个 StreamReadRequest 对象，表示一个流读取请求
* @throws Exception 当 streamListener 无法识别为 MessageConsumer 类型时，抛出异常
*/
private StreamMessageListenerContainer.StreamReadRequest<String> buildStreamReadRequest(StreamOffset<String> offset, StreamListener<String, ObjectRecord<String, String>> streamListener) throws Exception {
Consumer consumer;
if (streamListener instanceof MessageConsumer) {
consumer = Consumer.from(userEventGroup, InetAddress.getLocalHost().getHostName());
} else {
throw new Exception("无法识别的 stream key");
}
// 关闭自动 ack 确认
return StreamMessageListenerContainer.StreamReadRequest.builder(offset)
.errorHandler((error) -> {
log.error(error.getMessage());
})
.cancelOnError(e -> false)
.consumer(consumer)
// 关闭自动 ack 确认
.autoAcknowledge(false)
.build();
}
/**
* 检查 Redis 版本是否符合要求
*
* @throws IllegalStateException 如果 Redis 版本小于 5.0.0 版本，抛出该异常
*/
private void checkRedisVersion() {
// 获得 Redis 版本
Properties info = redisTemplate.execute((RedisCallback<Properties>) RedisServerCommands::info);
Assert.notNull(info, "Redis info is null");
Object redisVersion = info.get("redis_version");
Integer anInt = Convert.toInt(redisVersion);
if (anInt < 5) {
throw new IllegalStateException(StrUtil.format("您当前的 Redis 版本为 {}，小于最低要求的 5.0.0 版本！", redisVersion));
}
}
@Override
public void destroy() throws Exception {
}
@Override
public void afterPropertiesSet() throws Exception {
checkRedisVersion();
StreamOperations<String, Object, Object> streamOperations = redisTemplate.opsForStream();
if (Boolean.FALSE.equals(redisTemplate.hasKey(streamName))) {
streamOperations.createGroup(streamName, ReadOffset.from("0"), userEventGroup);
}
}
}
说明：该配置类实现了对 Redis Stream 的相关配置，包括消息监听容器的创建、流读取请求的生成、Redis 版本的检查以及组的创建等功能。
生产者
package com.mjg.config;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import java.util.Collections;
@Component
@RequiredArgsConstructor
@Slf4j
public class MessageProducer {
private final RedisTemplate<String, Object> redisTemplate;
public void sendMessage(String streamKey, Object message) {
RecordId recordId = redisTemplate
.opsForStream().add(StreamRecords.newRecord()
.ofMap(Collections.singletonMap("data", message))
.withStreamKey(streamKey));
if (recordId!= null) {
log.info("Message sent to Stream '{}' with RecordId: {}", streamKey, recordId);
}
}
}
说明：MessageProducer 类负责向 Redis Stream 发送消息。
消费者
package com.mjg.config;
import lombok.RequiredArgsConstructor;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
@RequiredArgsConstructor
@Component
public class MessageConsumer implements StreamListener<String, ObjectRecord<String, String>> {
private final RedisTemplate<String, Object> redisTemplate;
@Override
public void onMessage(ObjectRecord<String, String> message) {
String stream = message.getStream();
String messageId = message.getId().toString();
String messageBody = message.getValue();
System.out.println("Received message from Stream '" + stream + "' with messageId: " + messageId);
System.out.println("Message body: " + messageBody);
// 消息应答
redisTemplate.opsForStream().acknowledge(RedisStreamConfig.streamName, RedisStreamConfig.userEventGroup, message.getId());
}
}
说明：MessageConsumer 类实现了 StreamListener 接口，用于处理从 Redis Stream 接收到的消息，并进行相应的应答操作。
测试
@RequiredArgsConstructor
@Slf4j
@RestController
public class MessageController {
public static String streamName = "user-event-stream";
private final MessageProducer messageProducer;
@GetMapping("/send")
public void send() {
messageProducer.sendMessage(streamName, "hello 啦啦啦啦" + LocalDateTime.now());
}
}
说明：MessageController 类中的 send 方法通过调用 MessageProducer 来发送消息到指定的 Redis Stream 中。
以上就是SpringBoot使用Redis Stream实现轻量消息队列的示例代码的详细内容，更多关于SpringBoot Redis Stream轻量消息队列的资料请关注脚本之家其它相关文章！
相关文章
这篇文章主要介绍了Jenkins的安装配置详解,小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧2018-06-06
汉诺塔是源于印度一个古老传说的益智玩具。大梵天创造世界时做了三根石柱，在一根柱子上从下往上按大小顺序摞着64片黄金圆盘。大梵天命令婆罗门把圆盘从下面开始按大小顺序重新摆放在另一根柱子上。并且规定，在小圆盘上不能放大圆盘，三根柱子之间一次只能移动一个圆盘2022-02-02
Java中的RoundingMode枚举类用于控制高精度数值处理时的舍入行为,提供多种舍入模式,本文给大家介绍Java的RoundingMode舍入模式,感兴趣的朋友跟随小编一起看看吧2026-01-01
Intellij IDEA使用restclient测试的教程图解
这篇文章主要介绍了Intellij IDEA使用restclient测试的教程图解,本文通过图文并茂的形式给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下2021-01-01
这篇文章主要给大家介绍了关于Java终止线程的几种方式,线程停止即Terminated状态是伴随run方法的结束而生,也就是run完成后由Thread类来决定线程停止了,销毁资源释放空间,下面需要的朋友可以参考下2023-06-06
Spring不能注入Static变量的原因及Spring注入静态变量
这篇文章主要介绍了Spring不能注入Static变量的原因及Spring注入静态变量,需要的朋友可以参考下2016-01-01
这篇文章主要为大家详细介绍了Java使用OpenCV3.2实现视频读取与播放，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2019-07-07
这篇文章主要介绍了Spring Boot 排除自动配置的4个方法，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2021-08-08
这篇文章主要给大家介绍了关于feign服务端发现异常客户端处理的方法，文中通过示例代码介绍的非常详细，对大家学习或者使用feign具有一定的参考学习价值，需要的朋友们下面来一起学习学习吧2019-07-07
本篇文章是小编总结的关于Java IO的flush的相关知识点内容，有需要的朋友可以跟着学习下。2018-06-06
最新评论