Redis消息队列

Redis是一个支持消息队列功能的内存数据库，可以作为消息队列系统使用。

基于List结构模拟消息队列

消息队列（Message Queue），字面意思就是存放消息的队列。而Redis的list数据结构是一个双向链表，很容易模拟出队列效果。队列是入口和出口不在一边，因此我们可以利用：LPUSH 结合 RPOP、或者 RPUSH 结合 LPOP来实现。不过要注意的是，当队列中没有消息时RPOP或LPOP操作会返回null，并不像JVM的阻塞队列那样会阻塞并等待消息。因此这里应该使用BRPOP或者BLPOP来实现阻塞效果。

架构

架构如下：

引入依赖

        <dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.8.10</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><!-- redis 缓存操作 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><!--切面--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-aop</artifactId></dependency>

properties

#redis
spring.redis.host=127.0.0.1
spring.redis.port=6379
spring.redis.timeout=10s
spring.redis.password=123

自定义注解

消费者注解

import org.springframework.core.annotation.AliasFor;
import org.springframework.stereotype.Component;import java.lang.annotation.*;/*** 消费者注解*/
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Component
public @interface MessageConsumer {@AliasFor(annotation = Component.class)String value() default "";
}

处理器注解

import org.springframework.stereotype.Component;import java.lang.annotation.*;/*** 处理器注解，不同的类型使用不同的注解标准*/
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Component
public @interface MessageHandler {MessageListener.Mode value() default MessageListener.Mode.TOPIC;}

监听注解


import java.lang.annotation.*;/*** 监听注解*/
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface MessageListener {String value() default "";String topic() default "";String channel() default "";String streamKey() default "";/*** group-name是关联到流的消费者组的名称。*/String consumerGroup() default "";/*** consumer-name是客户端用于在消费者组内标识自己的字符串*/String consumerName() default "";/*** 读取未ack**/boolean pending() default false;Mode mode() default Mode.TOPIC;enum Mode {/*** topic 模式，主题订阅*/TOPIC(),/*** pub/sub 模式 订阅发布 广播*/PUBSUB(),/*** stream 模式*/STREAM()}
}

消费者


import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.core.RedisTemplate;import javax.annotation.Resource;/*** 消费者*/
@MessageConsumer
@Slf4j
public class RedisMqConsumer {@ResourceRedisTemplate<Object, Object> redisTemplate;@MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)public void testTopic1(Message<?> message) {log.info("工作者1===> " + message);}@MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)public void testTopic11(Message<?> message) {log.info("工作者2===> " + message);}@MessageListener(topic = "topic2", mode = MessageListener.Mode.TOPIC)public void testTopic2(Message<?> message) {log.info("topic2===> " + message);}@MessageListener(topic = "topic3", mode = MessageListener.Mode.TOPIC)public void testTopic3(Message<?> message) {log.info("topic3===> " + message);}}

处理者

处理接口


import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import cn.hutool.core.util.CharsetUtil;
import cn.hutool.json.JSONUtil;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisTemplate;import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Set;/*** 处理接口*/
@Slf4j
public abstract class AbstractMessageHandler implements ApplicationContextAware {// 用于获取应用程序上下文的成员变量。protected ApplicationContext applicationContext;protected RedisTemplate<Object,Object> redisTemplate;/*** 通过反射 执行 Method 方法* @param method 方法* @param message 消息* @param bean bean 对象 本案例中是 RedisMqConsumer*/protected void invokeMethod(Method method, Message<?> message, Object bean) {try {method.invoke(bean, message);} catch (IllegalAccessException | InvocationTargetException e) {e.printStackTrace();}}// 将字节数组转换为字符串，并使用 JSONUtil 将其转换为消息对象。protected Message<?> getMessage(byte[] bytes) {String s = new String(bytes, CharsetUtil.CHARSET_UTF_8);return JSONUtil.toBean(s, Message.class);}// 构造函数，接收 RedisTemplate 对象作为参数，并将其赋值给 redisTemplate 成员变量。public AbstractMessageHandler(RedisTemplate<Object,Object> redisTemplate) {this.redisTemplate = redisTemplate;}/*** 获取 Redis 连接的方法，返回 RedisConnection 对象。* @return RedisConnection*/protected RedisConnection getConnection() {return redisTemplate.getRequiredConnectionFactory().getConnection();}/*** 执行消息监听逻辑* @param method 方法*/public abstract void invokeMessage(Method method);// 实现 ApplicationContextAware 接口的方法，用于设置应用程序上下文。@Overridepublic void setApplicationContext(@NonNull ApplicationContext applicationContext) {this.applicationContext = applicationContext;}// 消息消费方法，从字节数组中获取消息对象，并根据消息的唯一标识进行消费判断，然后调用 invokeMethod 方法执行指定方法。protected void consumer(Method method, Set<String> consumers, Object bean, byte[] message) {Message<?> msg = getMessage(message);// 已经消费过的消息的 id 集合，用于判断消息是否已经被处理过。if (consumers.add(msg.getId())) {invokeMethod(method, msg, bean);} else {log.error("消息已经被消费 {}",msg);}}
}

工作队列


import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisTemplate;import java.lang.reflect.Method;
import java.util.HashSet;
import java.util.List;
import java.util.Set;/*** topic处理方式 实现 工作队列* 基于 list 的 lPush/brPop*/
@MessageHandler(value = MessageListener.Mode.TOPIC)
public class TopicMessageHandler extends AbstractMessageHandler {public TopicMessageHandler(RedisTemplate<Object,Object> redisTemplate) {super(redisTemplate);}@Overridepublic void invokeMessage(Method method) {Set<String> consumers = new HashSet<>();// 获取了MessageListener注解的相关信息MessageListener annotation = method.getAnnotation(MessageListener.class);// 主题（topic）String topic = getTopic(annotation);// 获取到一个RedisConnection对象RedisConnection connection = getConnection();// 通过反射获取到方法所在的类和方法所在的Bean对象。Class<?> declaringClass = method.getDeclaringClass();Object bean = applicationContext.getBean(declaringClass);/*** 通过调用connection.bRPop(1, topic.getBytes())方法阻塞地等待从Redis的列表中获取消息。* 方法参数中的1表示阻塞时间为1秒，topic.getBytes()将主题转换为字节数组形式。*/while (true) {List<byte[]> bytes = connection.bRPop(1, topic.getBytes());if (CollectionUtil.isNotEmpty(bytes)) {if (null != bytes.get(1)) {consumer(method, consumers, bean, bytes.get(1));}}}}private String getTopic(MessageListener annotation) {// 获取注解中的 value 属性值String value = annotation.value();// 获取注解中的 topic 属性值String topic = annotation.topic();// 如果 topic 为空，则使用 value，否则使用 topicreturn StrUtil.isBlank(topic) ? value : topic;}}

消费者注册中心


import cn.hutool.core.util.ArrayUtil;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Map;
import java.util.stream.Collectors;/*** 启动配置，在项目启动时自动运行，主要用于消费者的注册*/
@Component
@Slf4j
public class MessageConsumerStater implements ApplicationRunner, ApplicationContextAware {private ApplicationContext applicationContext;/*** 项目启动时，获取所有的 标注 MessageConsumer 注解的方法 ，开启消息监听逻辑** @param args ApplicationArguments*/@Overridepublic void run(ApplicationArguments args) {// 通过 getInvokers() 方法获取所有的 AbstractMessageHandler 实例Map<MessageListener.Mode, AbstractMessageHandler> invokers = getInvokers();/*** 使用 applicationContext.getBeansWithAnnotation(MessageConsumer.class) 获取所有标注了 MessageConsumer 注解的 Bean，* 并使用 values() 方法获取这些 Bean 的集合。*/applicationContext.getBeansWithAnnotation(MessageConsumer.class).values().parallelStream().forEach(consumer -> {/*** 通过 consumer.getClass().getMethods() 获取该 Bean 类的所有方法，并存储在 methods 数组中。* 如果 methods 数组非空，就通过并行流遍历这些方法，并对每个方法执行以下操作：* 调用 startMessageListener(method, invokers) 方法，传入该方法和之前获取的消息处理器映射对象 invokers。这个方法用于启动消息监听逻辑。*/Method[] methods = consumer.getClass().getMethods();if (ArrayUtil.isNotEmpty(methods)) {Arrays.stream(methods).parallel().forEach(method -> startMessageListener(method, invokers));}});}/*** 找到对应的处理方式来处理消息的消费逻辑** @param method     消费者方法 -> 本案例对应的是 MqConsumer 中的方法* @param handlerMap 所有的处理方式集合*/private void startMessageListener(Method method, Map<MessageListener.Mode, AbstractMessageHandler> handlerMap) {MessageListener listener = method.getAnnotation(MessageListener.class);if (null == listener) {return;}MessageListener.Mode mode = listener.mode();AbstractMessageHandler handler = handlerMap.get(mode);if (handler == null) {log.error("invoker is null");return;}handler.invokeMessage(method);}private Map<MessageListener.Mode, AbstractMessageHandler> getInvokers() {// 通过调用 getBeansWithAnnotation() 方法从应用程序上下文中获取所有带有 MessageHandler 注解的 BeanMap<String, Object> beansWithAnnotation = applicationContext.getBeansWithAnnotation(MessageHandler.class);/*** 使用 .collect(Collectors.toMap(k -> k.getClass().getAnnotation(MessageHandler.class).value(), k -> (AbstractMessageHandler) k)) 进行收集操作。具体来说：* k -> k.getClass().getAnnotation(MessageHandler.class).value() 是用于获取每个 Bean 的 MessageHandler 注解上的 value 属性值作为 Map 的键。* k -> (AbstractMessageHandler) k 是将 Bean 强制转换为 AbstractMessageHandler 类型后作为 Map 的值。*/Map<MessageListener.Mode, AbstractMessageHandler> collect =beansWithAnnotation.values().stream().collect(Collectors.toMap(k -> k.getClass().getAnnotation(MessageHandler.class).value(), k -> (AbstractMessageHandler) k));return collect;}public MessageConsumerStater() {}@Overridepublic void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}
}

RedisConfig


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.ObjectRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.time.Duration;@Configuration
public class RedisConfig{@Bean@SuppressWarnings("all")public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();redisTemplate.setConnectionFactory(redisConnectionFactory);RedisSerializer<String> redisSerializer = RedisSerializer.string();redisTemplate.setValueSerializer(redisSerializer);redisTemplate.setKeySerializer(redisSerializer);redisTemplate.setHashKeySerializer(redisSerializer);redisTemplate.setDefaultSerializer(redisSerializer);redisTemplate.setStringSerializer(redisSerializer);return redisTemplate;}@Beanpublic StreamMessageListenerContainer<String, ObjectRecord<String, String>> redisListenerContainer(ThreadPoolTaskExecutor executor, RedisTemplate<Object, Object> redisTemplate) {StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> options =StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()// 拉取消息超时时间.pollTimeout(Duration.ofSeconds(5))// 批量抓取消息.batchSize(1)// 传递的数据类型.targetType(String.class).executor(executor).build();RedisConnectionFactory connectionFactory = redisTemplate.getConnectionFactory();assert connectionFactory != null;StreamMessageListenerContainer<String, ObjectRecord<String, String>> container = StreamMessageListenerContainer.create(connectionFactory, options);container.start();return container;}
}

消息实体


import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;import java.io.Serializable;/*** 消息实体**/
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class Message<T> implements Serializable {private String id;private T content;
}

生产者


import cn.hutool.json.JSONUtil;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;
import java.util.UUID;@RestController
@RequestMapping("/test")
public class ProductController {@ResourceRedisTemplate<Object, Object> redisTemplate;@GetMapping("/listMessage")public void listMessage() {Message<String> message = new Message<>();// 随机生成idmessage.setId(UUID.randomUUID().toString());// 消息内容message.setContent("0001");redisTemplate.opsForList().leftPush("topic1", JSONUtil.toJsonStr(message));}
}

测试

获取消息成功，工作者1获取消息。

再次发送消息，工作者2获取消息，这个结果表明只能有其中一个工作者去消费消息。

总结

优点：

利用Redis存储，不受限于JVM内存上限
基于Redis的持久化机制，数据安全性有保证
可以满足消息有序性

缺点：

无法避免消息丢失
只支持单消费者

基于PubSub的消息队列

PubSub（发布订阅）是Redis2.0版本引入的消息传递模型。顾名思义，消费者可以订阅一个或多个channel，生产者向对应channel发送消息后，所有订阅者都能收到相关消息。

SUBSCRIBE channel [channel] ：订阅一个或多个频道
PUBLISH channel msg ：向一个频道发送消息
PSUBSCRIBE pattern[pattern] ：订阅与pattern格式匹配的所有频道

在原有的基础上实现。

架构

架构如下：

消费者


import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.core.RedisTemplate;import javax.annotation.Resource;/*** 消费者*/
@MessageConsumer
@Slf4j
public class RedisMqConsumer {@ResourceRedisTemplate<Object, Object> redisTemplate;@MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)public void testTopic1(Message<?> message) {log.info("工作者1===> " + message);}@MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)public void testTopic11(Message<?> message) {log.info("工作者2===> " + message);}@MessageListener(topic = "topic2", mode = MessageListener.Mode.TOPIC)public void testTopic2(Message<?> message) {log.info("topic2===> " + message);}@MessageListener(topic = "topic3", mode = MessageListener.Mode.TOPIC)public void testTopic3(Message<?> message) {log.info("topic3===> " + message);}@MessageListener(channel = "pubsub", mode = MessageListener.Mode.PUBSUB)public void testPubsub1(Message<?> message) {log.info("pubsub1===> " + message);}@MessageListener(channel = "pubsub", mode = MessageListener.Mode.PUBSUB)public void testPubsub2(Message<?> message) {log.info("pubsub2===> " + message);}}

处理者


import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisTemplate;
import cn.hutool.core.util.StrUtil;import java.lang.reflect.Method;
import java.util.HashSet;
import java.util.Set;/*** Pub/sub 处理方式 实现 广播消息*/
@MessageHandler(value = MessageListener.Mode.PUBSUB)
public class PubsubMessageHandler extends AbstractMessageHandler {public PubsubMessageHandler(RedisTemplate<Object,Object> redisTemplate) {super(redisTemplate);}@Overridepublic void invokeMessage(Method method) {Set<String> consumers = new HashSet<>();MessageListener listener = method.getAnnotation(MessageListener.class);String channel = getChannel(listener);RedisConnection connection = getConnection();/*** 模式匹配订阅*/if (listener.channel().contains("*")) {connection.pSubscribe((message, pattern) -> {Class<?> declaringClass = method.getDeclaringClass();Object bean = applicationContext.getBean(declaringClass);byte[] body = message.getBody();consumer(method, consumers, bean, body);}, channel.getBytes());}/*** 普通订阅*/else {connection.subscribe((message, pattern) -> {Class<?> declaringClass = method.getDeclaringClass();Object bean = applicationContext.getBean(declaringClass);byte[] body = message.getBody();consumer(method, consumers, bean, body);}, channel.getBytes());}}private String getChannel(MessageListener annotation) {String value = annotation.value();String channel = annotation.channel();return StrUtil.isBlank(channel) ? value : channel;}
}

生产者


import cn.hutool.json.JSONUtil;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;
import java.util.UUID;@RestController
@RequestMapping("/test")
public class ProductController {@ResourceRedisTemplate<Object, Object> redisTemplate;@GetMapping("/listMessage")public void listMessage() {Message<String> message = new Message<>();// 随机生成idmessage.setId(UUID.randomUUID().toString());// 消息内容message.setContent("0001");redisTemplate.convertAndSend("pubsub", JSONUtil.toJsonStr(message));
//        redisTemplate.opsForList().leftPush("topic1", JSONUtil.toJsonStr(message));}
}

测试

发送普通消息，消费者1和消费者2都接收到信息了

发送模式匹配消息，只有消费者2接收到信息。

总结

优点：

采用发布订阅模型，支持多生产、多消费

缺点：

不支持数据持久化
无法避免消息丢失
消息堆积有上限，超出时数据丢失

基于Stream的消息队列

Stream 是 Redis 5.0 引入的一种新数据类型，可以实现一个功能非常完善的消息队列。

消息队列相关命令

XADD - 添加消息到末尾
XTRIM - 对流进行修剪，限制长度
XDEL - 删除消息
XLEN - 获取流包含的元素数量，即消息长度
XRANGE - 获取消息列表，会自动过滤已经删除的消息
XREVRANGE - 反向获取消息列表，ID 从大到小
XREAD - 以阻塞或非阻塞方式获取消息列表

架构

架构如下：

消费者


import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.core.RedisTemplate;import javax.annotation.Resource;/*** 消费者*/
@MessageConsumer
@Slf4j
public class RedisMqConsumer {@ResourceRedisTemplate<Object, Object> redisTemplate;@MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)public void testTopic1(Message<?> message) {log.info("工作者1===> " + message);}@MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)public void testTopic11(Message<?> message) {log.info("工作者2===> " + message);}@MessageListener(topic = "topic2", mode = MessageListener.Mode.TOPIC)public void testTopic2(Message<?> message) {log.info("topic2===> " + message);}@MessageListener(topic = "topic3", mode = MessageListener.Mode.TOPIC)public void testTopic3(Message<?> message) {log.info("topic3===> " + message);}@MessageListener(channel = "pubsub.test", mode = MessageListener.Mode.PUBSUB)public void testPubsub1(Message<?> message) {log.info("pubsub1===> " + message);}@MessageListener(channel = "pubsub.*", mode = MessageListener.Mode.PUBSUB)public void testPubsub2(Message<?> message) {log.info("pubsub2===> " + message);}@MessageListener(streamKey = "streamKey",mode = MessageListener.Mode.STREAM)public void testStream1(ObjectRecord<Object, Object> message) {log.info("testStream消费消息 ===> " + message);// 只获取最新一条redisTemplate.opsForStream().trim("streamKey",1);}}

处理者

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;import javax.annotation.Resource;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;/*** Stream方式 实现消息队列*/
@MessageHandler(value = MessageListener.Mode.STREAM)
@Slf4j
public class StreamMessageHandler extends AbstractMessageHandler {@ResourceStreamMessageListenerContainer<String, ObjectRecord<String, String>> redisListenerContainer;public StreamMessageHandler(RedisTemplate<Object, Object> redisTemplate) {super(redisTemplate);}@Overridepublic synchronized void invokeMessage(Method method) {// 获取注解中的相关信息MessageListener annotation = method.getAnnotation(MessageListener.class);String streamKey = annotation.streamKey();StreamMessageListenerContainer.StreamReadRequest<String> streamReadRequest =StreamMessageListenerContainer.StreamReadRequest.builder(StreamOffset.create(streamKey, ReadOffset.lastConsumed())).errorHandler((error) -> log.error("redis stream 异常 ：{}", error.getMessage())).cancelOnError(e -> false).build();// 指定消费者对象注册到容器中去redisListenerContainer.register(streamReadRequest, (message) -> {Class<?> declaringClass = method.getDeclaringClass();Object bean = applicationContext.getBean(declaringClass);try {method.invoke(bean, message);} catch (IllegalAccessException | InvocationTargetException e) {e.printStackTrace();}});}}

生产者

import cn.hutool.json.JSONUtil;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;
import java.util.UUID;@RestController
@RequestMapping("/test")
public class ProductController {@ResourceRedisTemplate<Object, Object> redisTemplate;@GetMapping("/listMessage")public void listMessage() {Message<String> message = new Message<>();// 随机生成idmessage.setId(UUID.randomUUID().toString());// 消息内容message.setContent("0001");redisTemplate.convertAndSend("pubsub.122", JSONUtil.toJsonStr(message));
//        redisTemplate.opsForList().leftPush("topic1", JSONUtil.toJsonStr(message));}@GetMapping("/test-stream")public String stream() {Message<String> message = new Message<>();message.setId(UUID.randomUUID().toString());message.setContent("0001");ObjectRecord<Object, Object> record = ObjectRecord.create("streamKey", JSONUtil.toJsonStr(message));// 将消息添加至消息队列中RecordId recordId = redisTemplate.opsForStream().add(record);return recordId == null ? "null" : recordId.getValue();}
}

测试

发送消息，消费者获取消息成功

总结

STREAM类型消息队列的XREAD命令特点：

消息可回溯
一个消息可以被多个消费者读取
可以阻塞读取
有消息漏读的风险

基于Stream的消息队列-消费者组

消费者组（Consumer Group）：将多个消费者划分到一个组中，监听同一个队列。具备下列特点：

消费者组相关命令

XGROUP CREATE - 创建消费者组
XREADGROUP GROUP - 读取消费者组中的消息
XACK - 将消息标记为"已处理"
XGROUP SETID - 为消费者组设置新的最后递送消息ID
XGROUP DELCONSUMER - 删除消费者
XGROUP DESTROY - 删除消费者组
XPENDING - 显示待处理消息的相关信息
XCLAIM - 转移消息的归属权
XINFO - 查看流和消费者组的相关信息；
XINFO GROUPS - 打印消费者组的信息；
XINFO STREAM - 打印流信息

架构

架构如下：

消费者


import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.core.RedisTemplate;import javax.annotation.Resource;/*** 消费者*/
@MessageConsumer
@Slf4j
public class RedisMqConsumer {@ResourceRedisTemplate<Object, Object> redisTemplate;@MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)public void testTopic1(Message<?> message) {log.info("工作者1===> " + message);}@MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)public void testTopic11(Message<?> message) {log.info("工作者2===> " + message);}@MessageListener(topic = "topic2", mode = MessageListener.Mode.TOPIC)public void testTopic2(Message<?> message) {log.info("topic2===> " + message);}@MessageListener(topic = "topic3", mode = MessageListener.Mode.TOPIC)public void testTopic3(Message<?> message) {log.info("topic3===> " + message);}@MessageListener(channel = "pubsub.test", mode = MessageListener.Mode.PUBSUB)public void testPubsub1(Message<?> message) {log.info("pubsub1===> " + message);}@MessageListener(channel = "pubsub.*", mode = MessageListener.Mode.PUBSUB)public void testPubsub2(Message<?> message) {log.info("pubsub2===> " + message);}@MessageListener(streamKey = "streamKey", consumerGroup = "consumerGroup", consumerName = "consumerName", mode = MessageListener.Mode.STREAM)public void testStream1(ObjectRecord<Object, Object> message) {log.info("testStream1 组 consumerGroup 名 consumerName 消费消息 ===> " + message);// 手动 ackredisTemplate.opsForStream().acknowledge("consumerGroup", message);}/*** 同一个组，不同的 consumerName 都可以进行消息的消费，与上面 testStream() 为竞争关系，消息仅被其中一个消费** @param message msg*/@MessageListener(streamKey = "streamKey", consumerGroup = "consumerGroup", consumerName = "consumerName2", mode = MessageListener.Mode.STREAM)public void testStream2(ObjectRecord<Object, Object> message) {log.info("testStream2  组 consumerGroup 名 consumerName2 消费消息 ===> " + message);
//        redisTemplate.opsForStream().acknowledge("consumerGroup", message);}/*** 不同组的消息，可以正常消费** @param message message*/@MessageListener(streamKey = "streamKey", consumerGroup = "consumerGroup3", consumerName = "consumerName3", mode = MessageListener.Mode.STREAM)public void testStream3(ObjectRecord<Object, Object> message) {log.info("testStream3 组 consumerGroup3 名 consumerName3 消费消息 ===> " + message);}/*** 消费 streamKey 中，没有进行 ack 的消息** @param message message*/@MessageListener(streamKey = "streamKey",consumerGroup = "consumerGroup",consumerName = "consumerName",pending = true,mode = MessageListener.Mode.STREAM)public void testStreamPending(ObjectRecord<Object, Object> message) {log.info("testStreamPending 组 consumerGroup 名 consumerName 消费未 ack 消息 ===> " + message);redisTemplate.opsForStream().acknowledge("consumerGroup", message);}
}

处理者

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;import javax.annotation.Resource;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;/*** Stream方式 实现消息队列*/
@MessageHandler(value = MessageListener.Mode.STREAM)
@Slf4j
public class StreamMessageHandler extends AbstractMessageHandler {@ResourceStreamMessageListenerContainer<String, ObjectRecord<String, String>> redisListenerContainer;public StreamMessageHandler(RedisTemplate<Object, Object> redisTemplate) {super(redisTemplate);}@Overridepublic synchronized void invokeMessage(Method method) {// 获取注解中的相关信息MessageListener annotation = method.getAnnotation(MessageListener.class);String streamKey = annotation.streamKey();String consumerGroup = annotation.consumerGroup();String consumerName = annotation.consumerName();boolean pending = annotation.pending();// 检测组是否存在this.checkAndCreatGroup(streamKey, consumerGroup);// 获取 StreamOffsetStreamOffset<String> offset = this.getOffset(streamKey, pending);// 创建消费者Consumer consumer = Consumer.from(consumerGroup, consumerName);StreamMessageListenerContainer.StreamReadRequest<String> streamReadRequest =StreamMessageListenerContainer.StreamReadRequest.builder(offset).errorHandler((error) -> log.error("redis stream 异常 ：{}", error.getMessage())).cancelOnError(e -> false).consumer(consumer)//关闭自动ack确认.autoAcknowledge(false).build();// 指定消费者对象注册到容器中去redisListenerContainer.register(streamReadRequest, (message) -> {Class<?> declaringClass = method.getDeclaringClass();Object bean = applicationContext.getBean(declaringClass);try {method.invoke(bean, message);} catch (IllegalAccessException | InvocationTargetException e) {e.printStackTrace();}});}private StreamOffset<String> getOffset(String streamKey, boolean pending) {if (pending) {// 获取尚未 ack 的消息return StreamOffset.create(streamKey, ReadOffset.from("0"));}// 指定消费最新的消息return StreamOffset.create(streamKey, ReadOffset.lastConsumed());}/*** 校验消费组是否存在，不存在则创建，否则会出现异常* Error in execution; nested exception is io.lettuce.core.RedisCommandExecutionException: NOGROUP No such key 'streamKey' or consumer group 'consumerGroup' in XREADGROUP with GROUP option** @param streamKey     streamKey* @param consumerGroup consumerGroup*/private void checkAndCreatGroup(String streamKey, String consumerGroup) {if (Boolean.TRUE.equals(redisTemplate.hasKey(streamKey))) {StreamOperations<Object, Object, Object> streamOperations = redisTemplate.opsForStream();StreamInfo.XInfoGroups groups = streamOperations.groups(streamKey);if (groups.isEmpty() || groups.stream().noneMatch(data -> consumerGroup.equals(data.groupName()))) {creatGroup(streamKey, consumerGroup);} else {groups.stream().forEach(g -> {log.info("XInfoGroups:{}", g);StreamInfo.XInfoConsumers consumers = streamOperations.consumers(streamKey, g.groupName());log.info("XInfoConsumers:{}", consumers);});}} else {creatGroup(streamKey, consumerGroup);}}private void creatGroup(String key, String group) {StreamOperations<Object, Object, Object> streamOperations = redisTemplate.opsForStream();String groupName = streamOperations.createGroup(key, group);log.info("创建组成功:{}", groupName);}
}

生产者


import cn.hutool.json.JSONUtil;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;
import java.util.UUID;@RestController
@RequestMapping("/test")
public class ProductController {@ResourceRedisTemplate<Object, Object> redisTemplate;@GetMapping("/listMessage")public void listMessage() {Message<String> message = new Message<>();// 随机生成idmessage.setId(UUID.randomUUID().toString());// 消息内容message.setContent("0001");redisTemplate.convertAndSend("pubsub.122", JSONUtil.toJsonStr(message));
//        redisTemplate.opsForList().leftPush("topic1", JSONUtil.toJsonStr(message));}@GetMapping("/test-stream")public String stream() {Message<String> message = new Message<>();message.setId(UUID.randomUUID().toString());message.setContent("0001");ObjectRecord<Object, Object> record = ObjectRecord.create("streamKey", JSONUtil.toJsonStr(message));// 将消息添加至消息队列中RecordId recordId = redisTemplate.opsForStream().add(record);return recordId == null ? "null" : recordId.getValue();}
}