环境说明
项目中使用的是springboot2.4.3版本
redis服务安装的是6.x版本
5.x版本也可以，因为redis是在5.x版本开始支持stream数据格式的。老的版本肯定是不行的
各类之间的关系图说明
普通方式的类关系说明
redis发布:订阅消息主要类说明.png
Stream方式的类关系说明
redis的stream发布:订阅消息主要类说明的.png
配置类代码
注意 配置类中有用到线程池，在这里就不展示线程池的配置代码了，可以自行百度配置一个线程池
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cache.annotation.EnableCaching;
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.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.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.concurrent.Executor;
/**
* redis实现消息队列相关配置
*
* @author LengYouNuan
* @create 2021-08-19 上午11:28
*/
@Configuration
@EnableCaching
@Slf4j
public class RedisMessageConfig {
@Value("${redis-message.topic.aliMsg}")
private String aliMsgTopic;
/**
* 注入TaskExecutorConfig 中配置的线程池
*/
@Resource
private Executor taskExecutor;
/**
* redis消息监听器容器
* 可以注册多个消息监听器，每个监听器可以检测多个主题
*
* @param redisConnectionFactory redis连接工厂
* @param listenerAdapter 消息监听器
* @return
*/
@Bean
public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory,
MessageListenerAdapter listenerAdapter) {
RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
// 订阅多个频道
redisMessageListenerContainer.addMessageListener(listenerAdapter, new PatternTopic(aliMsgTopic));
//redisMessageListenerContainer.addMessageListener(listenerAdapter, new PatternTopic("xxxxxxx"));
//序列化对象 发布的时候需要设置序列化；订阅方也需要设置同样的序列化
Jackson2JsonRedisSerializer seria = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
seria.setObjectMapper(objectMapper);
redisMessageListenerContainer.setTopicSerializer(seria);
return redisMessageListenerContainer;
}
//表示监听一个频道
@Bean
public MessageListenerAdapter listenerAdapter(MessageConsumer messageConsumer) {
//这个地方 是给messageListenerAdapter 传入一个消息接受的处理器，利用反射的方法调用“MessageConsumer ”
return new MessageListenerAdapter(messageConsumer, "getMessage");
}
/**
* stream监听容器配置
*
* @param redisConnectionFactory redis连接工厂
* @param streamListener 消息监听器，消息的实际消费者
* @return
*/
@Bean
public StreamMessageListenerContainer streamMessageListenerContainer(RedisConnectionFactory redisConnectionFactory, RedisStreamMessageListener streamListener) {
//构建StreamMessageListenerContainerOptions
StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
.builder()
//超时时间
.pollTimeout(Duration.ofSeconds(2))
.batchSize(10)
.targetType(String.class)
//执行时用的线程池
.executor(taskExecutor)
//还可以设置序列化方式，这里不做设置，使用默认的方式
.build();
//创建StreamMessageListenerContainer
StreamMessageListenerContainer<String, ObjectRecord<String, String>> container = StreamMessageListenerContainer
.create(redisConnectionFactory, options);
//指定消费最新的消息
StreamOffset<String> offset = StreamOffset.create(aliMsgTopic, ReadOffset.lastConsumed());
//创建消费者 指定消费者组和消费者名字（注意，这里使用到用户组时，发送消息时必须有用户组，不然会报错，消息消费不成功）
Consumer consumer = Consumer.from("group-1", "consumer-1");
StreamMessageListenerContainer.StreamReadRequest<String> streamReadRequest = StreamMessageListenerContainer.StreamReadRequest.builder(offset)
.errorHandler((error) -> log.error(error.getMessage()))
.cancelOnError(e -> false)
.consumer(consumer)
//关闭自动ack确认
.autoAcknowledge(false)
.build();
//指定消费者对象
container.register(streamReadRequest, streamListener);
container.start();
return container;
}
}
普通形式的监听类（也就是消息的消费者）
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.stereotype.Component;
/**
* redis消息订阅者
*
* @author LengYouNuan
* @create 2021-08-19 上午10:41
*/
@Component
public class MessageConsumer {
public void getMessage(String object) {
//使用和发布时一样的序列化方式，此处设置的消息格式是String类型，实际中可以根据业务需要设置消息格式
Jackson2JsonRedisSerializer seria = new Jackson2JsonRedisSerializer(String.class);
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
seria.setObjectMapper(objectMapper);
String mes = (String) seria.deserialize(object.getBytes());
//TODO：拿到消息之后执行业务操作
//System.out.println("消费：客户端消费了消息==》：" + mes);
}
}
stream方式的监听类（消息的实际消费者）
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
/**
* redis的stream消息监听
*
* @author LengYouNuan
* @create 2021-08-19 下午3:10
*/
@Component
@Slf4j
public class RedisStreamMessageListener implements StreamListener<String, ObjectRecord<String, String>> {
@Autowired
StringRedisTemplate stringRedisTemplate;
@Override
public void onMessage(ObjectRecord<String, String> message) {
// 消息ID
RecordId messageId = message.getId();
// 消息的key和value
String stream = message.getStream();
String string = message.getValue();
log.info("========stream监听器监听到消息========它们分别是。messageId={}, stream={}, body={}", messageId,
stream, string);
// 通过RedisTemplate手动确认消息，确认之后消息会从队列中消失，如果不确认，可能存在重复消费
Long acknowledge = this.stringRedisTemplate.opsForStream().acknowledge("group-1", message);
if (acknowledge > 0) {
System.out.println("acknowledge的值是：" + acknowledge);
}
}
}
最后测试类
@Resource
private RedisTemplate redisTemplate;
@Value("${redis-message.topic.aliMsg}")
private String aliMsgTopic;
@Test
public void test1() {
for (int i = 0; i < 50; i++) {
redisTemplate.convertAndSend(aliMsgTopic, "主题" + aliMsgTopic + "====》发送消息" + i);
}
}
/**
* 无用户组和ack确认
*/
@Test
public void testStream() {
Map<String, String> msg = new HashMap<>();
msg.put("aaa", "消息aaaaaaa===》");
MapRecord mapRecord = MapRecord.create(aliMsgTopic, msg);
StringRecord stringRecord = StringRecord.of(mapRecord).withStreamKey(aliMsgTopic);
redisTemplate.opsForStream().add(stringRecord);
}
/**
* 有用户组，无ack确认
*/
@Test
public void testConsumerStream() {
Map<String, String> msg = new HashMap<>();
msg.put("aaa", "消息aaaaaaa===》");
MapRecord mapRecord = MapRecord.create(aliMsgTopic, msg);
StringRecord stringRecord = StringRecord.of(mapRecord).withStreamKey(aliMsgTopic);
StreamOperations streamOperations = redisTemplate.opsForStream();
Map<String, String> msg1 = new HashMap<>();
msg1.put("aaa1", "消息aaaaaaa11111===》");
MapRecord mapRecord1 = MapRecord.create(aliMsgTopic, msg1);
StringRecord stringRecord1 = StringRecord.of(mapRecord1).withStreamKey(aliMsgTopic);
Map<String, String> msg2 = new HashMap<>();
msg2.put("aaa2", "消息aaaaaaa222222222===》");
MapRecord mapRecord2 = MapRecord.create(aliMsgTopic, msg2);
StringRecord stringRecord2 = StringRecord.of(mapRecord2).withStreamKey(aliMsgTopic);
//注意，当用户组已经存在时，执行创建同名用户组会报错，测试时调用了销毁用户组方法，但是没有效果，业务中用到用户组的可能性不大，如果真的用到了，建议在程序初始化时创建用户组
// String group = streamOperations.createGroup(aliMsgTopic, "group-1");
// org.springframework.data.redis.connection.stream.Consumer consumer=
// org.springframework.data.redis.connection.stream.Consumer.from(group, "consumer-1");
streamOperations.add(stringRecord1);
streamOperations.add(stringRecord);
streamOperations.add(stringRecord2);
}
/**
* 有用户组，有ack确认，ack确认实际在消费者端做，ack确认必须有用户组存在
*/
@Test
public void testACKStream() {
Map<String, String> msg = new HashMap<>();
msg.put("aaa", "消息aaaaa333333a===》");
MapRecord mapRecord = MapRecord.create(aliMsgTopic, msg);
StringRecord stringRecord = StringRecord.of(mapRecord).withStreamKey(aliMsgTopic);
StreamOperations streamOperations = redisTemplate.opsForStream();
//注意，当用户组已经存在时，执行创建同名用户组会报错，测试时调用了销毁用户组方法，但是没有效果，业务中用到用户组的可能性不大，如果真的用到了，建议在程序初始化时创建用户组
// String group = streamOperations.createGroup(aliMsgTopic, "group-1");
// org.springframework.data.redis.connection.stream.Consumer consumer=
// org.springframework.data.redis.connection.stream.Consumer.from(group, "consumer-1");
streamOperations.add(stringRecord);
}
