• Redis消息队列

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

    基于List结构模拟消息队列

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

    架构

    架构如下：

    引入依赖

    1.         
    2. 			cn.hutool
    3. 			hutool-all
    4. 			5.8.10
    5. 		
    6. 		
    7. 			org.projectlombok
    8. 			lombok
    9. 			true
    10. 		
    11. 		
    12. 		
    13. 			org.springframework.boot
    14. 			spring-boot-starter-data-redis
    15. 		
    16. 		
    17. 		
    18. 			org.springframework.boot
    19. 			spring-boot-starter-aop
    20.

    properties

    1. #redis
    2. spring.redis.host=127.0.0.1
    3. spring.redis.port=6379
    4. spring.redis.timeout=10s
    5. spring.redis.password=123

    自定义注解 

    消费者注解
    1. import org.springframework.core.annotation.AliasFor;
    2. import org.springframework.stereotype.Component;
    3.  
    4. import java.lang.annotation.*;
    5.  
    6. /**
    7.  * 消费者注解
    8.  */
    9. @Target({ElementType.TYPE})
    10. @Retention(RetentionPolicy.RUNTIME)
    11. @Documented
    12. @Component
    13. public @interface MessageConsumer {
    14.     @AliasFor(annotation = Component.class)
    15.     String value() default "";
    16. }
    处理器注解
    1. import org.springframework.stereotype.Component;
    2.  
    3. import java.lang.annotation.*;
    4.  
    5. /**
    6.  * 处理器注解，不同的类型使用不同的注解标准
    7.  */
    8. @Target({ElementType.TYPE})
    9. @Retention(RetentionPolicy.RUNTIME)
    10. @Documented
    11. @Component
    12. public @interface MessageHandler {
    13.     MessageListener.Mode value() default MessageListener.Mode.TOPIC;
    14.  
    15. }
    监听注解
    1.  
    2. import java.lang.annotation.*;
    3.  
    4. /**
    5.  * 监听注解
    6.  */
    7. @Target({ElementType.METHOD})
    8. @Retention(RetentionPolicy.RUNTIME)
    9. @Documented
    10. public @interface MessageListener {
    11.  
    12.     String value() default "";
    13.  
    14.     String topic() default "";
    15.  
    16.     String channel() default "";
    17.  
    18.     String streamKey() default "";
    19.  
    20.     /**
    21.      * group-name是关联到流的消费者组的名称。
    22.      */
    23.     String consumerGroup() default "";
    24.  
    25.     /**
    26.      * consumer-name是客户端用于在消费者组内标识自己的字符串
    27.      */
    28.     String consumerName() default "";
    29.  
    30.     /**
    31.      * 读取未ack
    32.      *
    33.      */
    34.     boolean pending() default false;
    35.  
    36.     Mode mode() default Mode.TOPIC;
    37.  
    38.     enum Mode {
    39.         /**
    40.          * topic 模式，主题订阅
    41.          */
    42.         TOPIC(),
    43.         /**
    44.          * pub/sub 模式 订阅发布 广播
    45.          */
    46.         PUBSUB(),
    47.         /**
    48.          * stream 模式
    49.          */
    50.         STREAM()
    51.     }
    52. }

    消费者

    1.  
    2. import lombok.extern.slf4j.Slf4j;
    3. import org.springframework.data.redis.connection.stream.ObjectRecord;
    4. import org.springframework.data.redis.core.RedisTemplate;
    5.  
    6. import javax.annotation.Resource;
    7.  
    8. /**
    9.  * 消费者
    10.  */
    11. @MessageConsumer
    12. @Slf4j
    13. public class RedisMqConsumer {
    14.     @Resource
    15.     RedisTemplate redisTemplate;
    16.  
    17.     @MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)
    18.     public void testTopic1(Message message) {
    19.         log.info("工作者1===> " + message);
    20.     }
    21.  
    22.     @MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)
    23.     public void testTopic11(Message message) {
    24.         log.info("工作者2===> " + message);
    25.     }
    26.  
    27.     @MessageListener(topic = "topic2", mode = MessageListener.Mode.TOPIC)
    28.     public void testTopic2(Message message) {
    29.         log.info("topic2===> " + message);
    30.     }
    31.  
    32.     @MessageListener(topic = "topic3", mode = MessageListener.Mode.TOPIC)
    33.     public void testTopic3(Message message) {
    34.         log.info("topic3===> " + message);
    35.     }
    36.  
    37. }

    处理者

    处理接口

    1.  
    2. import lombok.NonNull;
    3. import lombok.extern.slf4j.Slf4j;
    4. import cn.hutool.core.util.CharsetUtil;
    5. import cn.hutool.json.JSONUtil;
    6. import org.springframework.context.ApplicationContext;
    7. import org.springframework.context.ApplicationContextAware;
    8. import org.springframework.data.redis.connection.RedisConnection;
    9. import org.springframework.data.redis.core.RedisTemplate;
    10.  
    11.  
    12. import java.lang.reflect.InvocationTargetException;
    13. import java.lang.reflect.Method;
    14. import java.util.Set;
    15.  
    16. /**
    17.  * 处理接口
    18.  */
    19. @Slf4j
    20. public abstract class AbstractMessageHandler implements ApplicationContextAware {
    21.  
    22.     // 用于获取应用程序上下文的成员变量。
    23.     protected ApplicationContext applicationContext;
    24.  
    25.     protected RedisTemplate redisTemplate;
    26.  
    27.     /**
    28.      * 通过反射 执行 Method 方法
    29.      * @param method 方法
    30.      * @param message 消息
    31.      * @param bean bean 对象 本案例中是 RedisMqConsumer
    32.      */
    33.     protected void invokeMethod(Method method, Message message, Object bean) {
    34.         try {
    35.             method.invoke(bean, message);
    36.         } catch (IllegalAccessException | InvocationTargetException e) {
    37.             e.printStackTrace();
    38.         }
    39.     }
    40.     // 将字节数组转换为字符串，并使用 JSONUtil 将其转换为消息对象。
    41.     protected Message getMessage(byte[] bytes) {
    42.         String s = new String(bytes, CharsetUtil.CHARSET_UTF_8);
    43.         return JSONUtil.toBean(s, Message.class);
    44.     }
    45.     // 构造函数，接收 RedisTemplate 对象作为参数，并将其赋值给 redisTemplate 成员变量。
    46.     public AbstractMessageHandler(RedisTemplate redisTemplate) {
    47.         this.redisTemplate = redisTemplate;
    48.     }
    49.  
    50.     /**
    51.      * 获取 Redis 连接的方法，返回 RedisConnection 对象。
    52.      * @return RedisConnection
    53.      */
    54.     protected RedisConnection getConnection() {
    55.         return redisTemplate.getRequiredConnectionFactory().getConnection();
    56.     }
    57.  
    58.     /**
    59.      * 执行消息监听逻辑
    60.      * @param method 方法
    61.      */
    62.     public abstract void invokeMessage(Method method);
    63.     // 实现 ApplicationContextAware 接口的方法，用于设置应用程序上下文。
    64.     @Override
    65.     public void setApplicationContext(@NonNull ApplicationContext applicationContext) {
    66.         this.applicationContext = applicationContext;
    67.     }
    68.     // 消息消费方法，从字节数组中获取消息对象，并根据消息的唯一标识进行消费判断，然后调用 invokeMethod 方法执行指定方法。
    69.     protected void consumer(Method method, Set consumers, Object bean, byte[] message) {
    70.         Message msg = getMessage(message);
    71.         // 已经消费过的消息的 id 集合，用于判断消息是否已经被处理过。
    72.         if (consumers.add(msg.getId())) {
    73.             invokeMethod(method, msg, bean);
    74.         } else {
    75.             log.error("消息已经被消费 {}",msg);
    76.         }
    77.     }
    78. }
    工作队列
    1.  
    2. import cn.hutool.core.collection.CollectionUtil;
    3. import cn.hutool.core.util.StrUtil;
    4.  
    5. import org.springframework.data.redis.connection.RedisConnection;
    6. import org.springframework.data.redis.core.RedisTemplate;
    7.  
    8.  
    9. import java.lang.reflect.Method;
    10. import java.util.HashSet;
    11. import java.util.List;
    12. import java.util.Set;
    13.  
    14. /**
    15.  * topic处理方式 实现 工作队列
    16.  * 基于 list 的 lPush/brPop
    17.  */
    18. @MessageHandler(value = MessageListener.Mode.TOPIC)
    19. public class TopicMessageHandler extends AbstractMessageHandler {
    20.  
    21.     public TopicMessageHandler(RedisTemplate redisTemplate) {
    22.         super(redisTemplate);
    23.     }
    24.  
    25.     @Override
    26.     public void invokeMessage(Method method) {
    27.         Set consumers = new HashSet<>();
    28.         // 获取了MessageListener注解的相关信息
    29.         MessageListener annotation = method.getAnnotation(MessageListener.class);
    30.         // 主题（topic）
    31.         String topic = getTopic(annotation);
    32.         // 获取到一个RedisConnection对象
    33.         RedisConnection connection = getConnection();
    34.         // 通过反射获取到方法所在的类和方法所在的Bean对象。
    35.         Class declaringClass = method.getDeclaringClass();
    36.         Object bean = applicationContext.getBean(declaringClass);
    37.         /**
    38.          * 通过调用connection.bRPop(1, topic.getBytes())方法阻塞地等待从Redis的列表中获取消息。
    39.          * 方法参数中的1表示阻塞时间为1秒，topic.getBytes()将主题转换为字节数组形式。
    40.          */
    41.         while (true) {
    42.             List<byte[]> bytes = connection.bRPop(1, topic.getBytes());
    43.             if (CollectionUtil.isNotEmpty(bytes)) {
    44.                 if (null != bytes.get(1)) {
    45.                     consumer(method, consumers, bean, bytes.get(1));
    46.                 }
    47.             }
    48.         }
    49.     }
    50.  
    51.     private String getTopic(MessageListener annotation) {
    52.         // 获取注解中的 value 属性值
    53.         String value = annotation.value();
    54.         // 获取注解中的 topic 属性值
    55.         String topic = annotation.topic();
    56.         // 如果 topic 为空，则使用 value，否则使用 topic
    57.         return StrUtil.isBlank(topic) ? value : topic;
    58.     }
    59.  
    60. }

    消费者注册中心

    1.  
    2. import cn.hutool.core.util.ArrayUtil;
    3. import lombok.NonNull;
    4. import lombok.extern.slf4j.Slf4j;
    5. import org.springframework.beans.BeansException;
    6. import org.springframework.boot.ApplicationArguments;
    7. import org.springframework.boot.ApplicationRunner;
    8. import org.springframework.context.ApplicationContext;
    9. import org.springframework.context.ApplicationContextAware;
    10. import org.springframework.stereotype.Component;
    11.  
    12. import java.lang.reflect.Method;
    13. import java.util.Arrays;
    14. import java.util.Map;
    15. import java.util.stream.Collectors;
    16.  
    17. /**
    18.  * 启动配置，在项目启动时自动运行，主要用于消费者的注册
    19.  */
    20. @Component
    21. @Slf4j
    22. public class MessageConsumerStater implements ApplicationRunner, ApplicationContextAware {
    23.  
    24.     private ApplicationContext applicationContext;
    25.  
    26.     /**
    27.      * 项目启动时，获取所有的 标注 MessageConsumer 注解的方法 ，开启消息监听逻辑
    28.      *
    29.      * @param args ApplicationArguments
    30.      */
    31.     @Override
    32.     public void run(ApplicationArguments args) {
    33.         // 通过 getInvokers() 方法获取所有的 AbstractMessageHandler 实例
    34.         Map invokers = getInvokers();
    35.         /**
    36.          * 使用 applicationContext.getBeansWithAnnotation(MessageConsumer.class) 获取所有标注了 MessageConsumer 注解的 Bean，
    37.          * 并使用 values() 方法获取这些 Bean 的集合。
    38.          */
    39.         applicationContext.getBeansWithAnnotation(MessageConsumer.class).values().parallelStream().forEach(consumer -> {
    40.             /**
    41.              * 通过 consumer.getClass().getMethods() 获取该 Bean 类的所有方法，并存储在 methods 数组中。
    42.              * 如果 methods 数组非空，就通过并行流遍历这些方法，并对每个方法执行以下操作：
    43.              * 调用 startMessageListener(method, invokers) 方法，传入该方法和之前获取的消息处理器映射对象 invokers。这个方法用于启动消息监听逻辑。
    44.              */
    45.             Method[] methods = consumer.getClass().getMethods();
    46.             if (ArrayUtil.isNotEmpty(methods)) {
    47.                 Arrays.stream(methods).parallel().forEach(method -> startMessageListener(method, invokers));
    48.             }
    49.         });
    50.     }
    51.  
    52.     /**
    53.      * 找到对应的处理方式来处理消息的消费逻辑
    54.      *
    55.      * @param method     消费者方法 -> 本案例对应的是 MqConsumer 中的方法
    56.      * @param handlerMap 所有的处理方式集合
    57.      */
    58.     private void startMessageListener(Method method, Map handlerMap) {
    59.         MessageListener listener = method.getAnnotation(MessageListener.class);
    60.         if (null == listener) {
    61.             return;
    62.         }
    63.         MessageListener.Mode mode = listener.mode();
    64.         AbstractMessageHandler handler = handlerMap.get(mode);
    65.         if (handler == null) {
    66.             log.error("invoker is null");
    67.             return;
    68.         }
    69.         handler.invokeMessage(method);
    70.     }
    71.  
    72.     private Map getInvokers() {
    73.         // 通过调用 getBeansWithAnnotation() 方法从应用程序上下文中获取所有带有 MessageHandler 注解的 Bean
    74.         Map beansWithAnnotation = applicationContext.getBeansWithAnnotation(MessageHandler.class);
    75.         /**
    76.          * 使用 .collect(Collectors.toMap(k -> k.getClass().getAnnotation(MessageHandler.class).value(), k -> (AbstractMessageHandler) k)) 进行收集操作。具体来说：
    77.          * k -> k.getClass().getAnnotation(MessageHandler.class).value() 是用于获取每个 Bean 的 MessageHandler 注解上的 value 属性值作为 Map 的键。
    78.          * k -> (AbstractMessageHandler) k 是将 Bean 强制转换为 AbstractMessageHandler 类型后作为 Map 的值。
    79.          */
    80.         Map collect =
    81.                 beansWithAnnotation.values().stream().collect(Collectors
    82.                         .toMap(k -> k.getClass().getAnnotation(MessageHandler.class).value(), k -> (AbstractMessageHandler) k));
    83.         return collect;
    84.     }
    85.  
    86.     public MessageConsumerStater() {
    87.  
    88.     }
    89.  
    90.     @Override
    91.     public void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException {
    92.         this.applicationContext = applicationContext;
    93.     }
    94. }

    RedisConfig

    1.  
    2. import org.springframework.context.annotation.Bean;
    3. import org.springframework.context.annotation.Configuration;
    4. import org.springframework.data.redis.connection.RedisConnectionFactory;
    5. import org.springframework.data.redis.connection.stream.ObjectRecord;
    6. import org.springframework.data.redis.core.RedisTemplate;
    7. import org.springframework.data.redis.serializer.RedisSerializer;
    8. import org.springframework.data.redis.stream.StreamMessageListenerContainer;
    9. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    10. import java.time.Duration;
    11.  
    12. @Configuration
    13. public class RedisConfig{
    14.  
    15.  
    16.  
    17.     @Bean
    18.     @SuppressWarnings("all")
    19.     public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) {
    20.         RedisTemplate redisTemplate = new RedisTemplate<>();
    21.         redisTemplate.setConnectionFactory(redisConnectionFactory);
    22.         RedisSerializer redisSerializer = RedisSerializer.string();
    23.         redisTemplate.setValueSerializer(redisSerializer);
    24.         redisTemplate.setKeySerializer(redisSerializer);
    25.         redisTemplate.setHashKeySerializer(redisSerializer);
    26.         redisTemplate.setDefaultSerializer(redisSerializer);
    27.         redisTemplate.setStringSerializer(redisSerializer);
    28.         return redisTemplate;
    29.     }
    30.  
    31.  
    32.  
    33.  
    34.     @Bean
    35.     public StreamMessageListenerContainer> redisListenerContainer(ThreadPoolTaskExecutor executor, RedisTemplate redisTemplate) {
    36.         StreamMessageListenerContainer.StreamMessageListenerContainerOptions> options =
    37.                 StreamMessageListenerContainer.StreamMessageListenerContainerOptions
    38.                         .builder()
    39.                         // 拉取消息超时时间
    40.                         .pollTimeout(Duration.ofSeconds(5))
    41.                         // 批量抓取消息
    42.                         .batchSize(1)
    43.                         // 传递的数据类型
    44.                         .targetType(String.class)
    45.                         .executor(executor)
    46.                         .build();
    47.         RedisConnectionFactory connectionFactory = redisTemplate.getConnectionFactory();
    48.         assert connectionFactory != null;
    49.         StreamMessageListenerContainer> container = StreamMessageListenerContainer
    50.                 .create(connectionFactory, options);
    51.         container.start();
    52.         return container;
    53.     }
    54. }

    消息实体

    1.  
    2. import lombok.AllArgsConstructor;
    3. import lombok.Builder;
    4. import lombok.Data;
    5. import lombok.NoArgsConstructor;
    6.  
    7. import java.io.Serializable;
    8.  
    9. /**
    10.  * 消息实体
    11.  *
    12.  */
    13. @Data
    14. @Builder
    15. @NoArgsConstructor
    16. @AllArgsConstructor
    17. public class Message implements Serializable {
    18.  
    19.     private String id;
    20.  
    21.     private T content;
    22. }

    生产者

    1.  
    2. import cn.hutool.json.JSONUtil;
    3. import org.springframework.data.redis.core.RedisTemplate;
    4. import org.springframework.web.bind.annotation.GetMapping;
    5. import org.springframework.web.bind.annotation.RequestMapping;
    6. import org.springframework.web.bind.annotation.RestController;
    7.  
    8. import javax.annotation.Resource;
    9. import java.util.UUID;
    10.  
    11. @RestController
    12. @RequestMapping("/test")
    13. public class ProductController {
    14.  
    15.  
    16.  
    17.     @Resource
    18.     RedisTemplate redisTemplate;
    19.     
    20.     @GetMapping("/listMessage")
    21.     public void listMessage() {
    22.         Message message = new Message<>();
    23.         // 随机生成id
    24.         message.setId(UUID.randomUUID().toString());
    25.         // 消息内容
    26.         message.setContent("0001");
    27.         redisTemplate.opsForList().leftPush("topic1", JSONUtil.toJsonStr(message));
    28.     }
    29. }

    测试

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

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

    总结 

    优点：

    • 利用Redis存储，不受限于JVM内存上限

    • 基于Redis的持久化机制，数据安全性有保证

    • 可以满足消息有序性

    缺点：

    • 无法避免消息丢失

    • 只支持单消费者

    基于PubSub的消息队列

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

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

    在原有的基础上实现。

    架构

    架构如下：

    消费者

    1.  
    2. import lombok.extern.slf4j.Slf4j;
    3. import org.springframework.data.redis.connection.stream.ObjectRecord;
    4. import org.springframework.data.redis.core.RedisTemplate;
    5.  
    6. import javax.annotation.Resource;
    7.  
    8. /**
    9.  * 消费者
    10.  */
    11. @MessageConsumer
    12. @Slf4j
    13. public class RedisMqConsumer {
    14.     @Resource
    15.     RedisTemplate redisTemplate;
    16.  
    17.     @MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)
    18.     public void testTopic1(Message message) {
    19.         log.info("工作者1===> " + message);
    20.     }
    21.  
    22.     @MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)
    23.     public void testTopic11(Message message) {
    24.         log.info("工作者2===> " + message);
    25.     }
    26.  
    27.     @MessageListener(topic = "topic2", mode = MessageListener.Mode.TOPIC)
    28.     public void testTopic2(Message message) {
    29.         log.info("topic2===> " + message);
    30.     }
    31.  
    32.     @MessageListener(topic = "topic3", mode = MessageListener.Mode.TOPIC)
    33.     public void testTopic3(Message message) {
    34.         log.info("topic3===> " + message);
    35.     }
    36.  
    37.     @MessageListener(channel = "pubsub", mode = MessageListener.Mode.PUBSUB)
    38.     public void testPubsub1(Message message) {
    39.         log.info("pubsub1===> " + message);
    40.     }
    41.  
    42.     @MessageListener(channel = "pubsub", mode = MessageListener.Mode.PUBSUB)
    43.     public void testPubsub2(Message message) {
    44.         log.info("pubsub2===> " + message);
    45.     }
    46.  
    47. }

    处理者

    1.  
    2. import org.springframework.data.redis.connection.RedisConnection;
    3. import org.springframework.data.redis.core.RedisTemplate;
    4. import cn.hutool.core.util.StrUtil;
    5.  
    6. import java.lang.reflect.Method;
    7. import java.util.HashSet;
    8. import java.util.Set;
    9.  
    10. /**
    11.  * Pub/sub 处理方式 实现 广播消息
    12.  */
    13. @MessageHandler(value = MessageListener.Mode.PUBSUB)
    14. public class PubsubMessageHandler extends AbstractMessageHandler {
    15.  
    16.     public PubsubMessageHandler(RedisTemplate redisTemplate) {
    17.         super(redisTemplate);
    18.     }
    19.  
    20.     @Override
    21.     public void invokeMessage(Method method) {
    22.         Set consumers = new HashSet<>();
    23.         MessageListener listener = method.getAnnotation(MessageListener.class);
    24.         String channel = getChannel(listener);
    25.         RedisConnection connection = getConnection();
    26.         /**
    27.          * 模式匹配订阅
    28.          */
    29.         if (listener.channel().contains("*")) {
    30.             connection.pSubscribe((message, pattern) -> {
    31.                 Class declaringClass = method.getDeclaringClass();
    32.                 Object bean = applicationContext.getBean(declaringClass);
    33.                 byte[] body = message.getBody();
    34.                 consumer(method, consumers, bean, body);
    35.             }, channel.getBytes());
    36.  
    37.         }
    38.         /**
    39.          * 普通订阅
    40.          */
    41.         else {
    42.             connection.subscribe((message, pattern) -> {
    43.                 Class declaringClass = method.getDeclaringClass();
    44.                 Object bean = applicationContext.getBean(declaringClass);
    45.                 byte[] body = message.getBody();
    46.                 consumer(method, consumers, bean, body);
    47.             }, channel.getBytes());
    48.         }
    49.     }
    50.  
    51.     private String getChannel(MessageListener annotation) {
    52.         String value = annotation.value();
    53.         String channel = annotation.channel();
    54.         return StrUtil.isBlank(channel) ? value : channel;
    55.     }
    56. }

    生产者

    1.  
    2. import cn.hutool.json.JSONUtil;
    3. import org.springframework.data.redis.core.RedisTemplate;
    4. import org.springframework.web.bind.annotation.GetMapping;
    5. import org.springframework.web.bind.annotation.RequestMapping;
    6. import org.springframework.web.bind.annotation.RestController;
    7.  
    8. import javax.annotation.Resource;
    9. import java.util.UUID;
    10.  
    11. @RestController
    12. @RequestMapping("/test")
    13. public class ProductController {
    14.  
    15.  
    16.  
    17.     @Resource
    18.     RedisTemplate redisTemplate;
    19.  
    20.     @GetMapping("/listMessage")
    21.     public void listMessage() {
    22.         Message message = new Message<>();
    23.         // 随机生成id
    24.         message.setId(UUID.randomUUID().toString());
    25.         // 消息内容
    26.         message.setContent("0001");
    27.         redisTemplate.convertAndSend("pubsub", JSONUtil.toJsonStr(message));
    28. //        redisTemplate.opsForList().leftPush("topic1", JSONUtil.toJsonStr(message));
    29.     }
    30. }

    测试

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

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

    总结

    优点：

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

    缺点：

    • 不支持数据持久化

    • 无法避免消息丢失

    • 消息堆积有上限，超出时数据丢失

    基于Stream的消息队列

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

    消息队列相关命令

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

    架构

    架构如下： 

    消费者

    1.  
    2. import lombok.extern.slf4j.Slf4j;
    3. import org.springframework.data.redis.connection.stream.ObjectRecord;
    4. import org.springframework.data.redis.core.RedisTemplate;
    5.  
    6. import javax.annotation.Resource;
    7.  
    8. /**
    9.  * 消费者
    10.  */
    11. @MessageConsumer
    12. @Slf4j
    13. public class RedisMqConsumer {
    14.     @Resource
    15.     RedisTemplate redisTemplate;
    16.  
    17.     @MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)
    18.     public void testTopic1(Message message) {
    19.         log.info("工作者1===> " + message);
    20.     }
    21.  
    22.     @MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)
    23.     public void testTopic11(Message message) {
    24.         log.info("工作者2===> " + message);
    25.     }
    26.  
    27.     @MessageListener(topic = "topic2", mode = MessageListener.Mode.TOPIC)
    28.     public void testTopic2(Message message) {
    29.         log.info("topic2===> " + message);
    30.     }
    31.  
    32.     @MessageListener(topic = "topic3", mode = MessageListener.Mode.TOPIC)
    33.     public void testTopic3(Message message) {
    34.         log.info("topic3===> " + message);
    35.     }
    36.  
    37.     @MessageListener(channel = "pubsub.test", mode = MessageListener.Mode.PUBSUB)
    38.     public void testPubsub1(Message message) {
    39.         log.info("pubsub1===> " + message);
    40.     }
    41.  
    42.     @MessageListener(channel = "pubsub.*", mode = MessageListener.Mode.PUBSUB)
    43.     public void testPubsub2(Message message) {
    44.         log.info("pubsub2===> " + message);
    45.     }
    46.     @MessageListener(streamKey = "streamKey",mode = MessageListener.Mode.STREAM)
    47.     public void testStream1(ObjectRecord message) {
    48.         log.info("testStream消费消息 ===> " + message);
    49.         // 只获取最新一条
    50.         redisTemplate.opsForStream().trim("streamKey",1);
    51.     }
    52.  
    53. }

     处理者

    1. import lombok.extern.slf4j.Slf4j;
    2. import org.springframework.data.redis.connection.stream.*;
    3. import org.springframework.data.redis.core.RedisTemplate;
    4. import org.springframework.data.redis.core.StreamOperations;
    5. import org.springframework.data.redis.stream.StreamMessageListenerContainer;
    6.  
    7.  
    8. import javax.annotation.Resource;
    9. import java.lang.reflect.InvocationTargetException;
    10. import java.lang.reflect.Method;
    11.  
    12. /**
    13.  * Stream方式 实现消息队列
    14.  */
    15. @MessageHandler(value = MessageListener.Mode.STREAM)
    16. @Slf4j
    17. public class StreamMessageHandler extends AbstractMessageHandler {
    18.     @Resource
    19.     StreamMessageListenerContainer> redisListenerContainer;
    20.  
    21.     public StreamMessageHandler(RedisTemplate redisTemplate) {
    22.         super(redisTemplate);
    23.     }
    24.  
    25.     @Override
    26.     public synchronized void invokeMessage(Method method) {
    27.         // 获取注解中的相关信息
    28.         MessageListener annotation = method.getAnnotation(MessageListener.class);
    29.         String streamKey = annotation.streamKey();
    30.  
    31.  
    32.  
    33.  
    34.         StreamMessageListenerContainer.StreamReadRequest streamReadRequest =
    35.                 StreamMessageListenerContainer.StreamReadRequest.builder(
    36.                                 StreamOffset.create(streamKey, ReadOffset.lastConsumed()))
    37.                         .errorHandler((error) -> log.error("redis stream 异常 ：{}", error.getMessage()))
    38.                         .cancelOnError(e -> false)
    39.                         .build();
    40.  
    41.  
    42.         // 指定消费者对象注册到容器中去
    43.         redisListenerContainer.register(streamReadRequest, (message) -> {
    44.             Class declaringClass = method.getDeclaringClass();
    45.             Object bean = applicationContext.getBean(declaringClass);
    46.             try {
    47.                 method.invoke(bean, message);
    48.             } catch (IllegalAccessException | InvocationTargetException e) {
    49.                 e.printStackTrace();
    50.             }
    51.         });
    52.     }
    53.  
    54.  
    55. }

    生产者

    1. import cn.hutool.json.JSONUtil;
    2. import org.springframework.data.redis.connection.stream.ObjectRecord;
    3. import org.springframework.data.redis.connection.stream.RecordId;
    4. import org.springframework.data.redis.core.RedisTemplate;
    5. import org.springframework.web.bind.annotation.GetMapping;
    6. import org.springframework.web.bind.annotation.RequestMapping;
    7. import org.springframework.web.bind.annotation.RestController;
    8.  
    9. import javax.annotation.Resource;
    10. import java.util.UUID;
    11.  
    12. @RestController
    13. @RequestMapping("/test")
    14. public class ProductController {
    15.  
    16.  
    17.  
    18.     @Resource
    19.     RedisTemplate redisTemplate;
    20.  
    21.     @GetMapping("/listMessage")
    22.     public void listMessage() {
    23.         Message message = new Message<>();
    24.         // 随机生成id
    25.         message.setId(UUID.randomUUID().toString());
    26.         // 消息内容
    27.         message.setContent("0001");
    28.         redisTemplate.convertAndSend("pubsub.122", JSONUtil.toJsonStr(message));
    29. //        redisTemplate.opsForList().leftPush("topic1", JSONUtil.toJsonStr(message));
    30.     }
    31.     @GetMapping("/test-stream")
    32.     public String stream() {
    33.         Message message = new Message<>();
    34.         message.setId(UUID.randomUUID().toString());
    35.         message.setContent("0001");
    36.         ObjectRecord record = ObjectRecord.create("streamKey", JSONUtil.toJsonStr(message));
    37.         // 将消息添加至消息队列中
    38.         RecordId recordId = redisTemplate.opsForStream().add(record);
    39.         return recordId == null ? "null" : recordId.getValue();
    40.     }
    41. }

    测试

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

    总结

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

    • 消息可回溯

    • 一个消息可以被多个消费者读取

    • 可以阻塞读取

    • 有消息漏读的风险  

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

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

     消费者组相关命令

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

    架构

    架构如下：

     

    消费者

    1.  
    2. import lombok.extern.slf4j.Slf4j;
    3. import org.springframework.data.redis.connection.stream.ObjectRecord;
    4. import org.springframework.data.redis.core.RedisTemplate;
    5.  
    6. import javax.annotation.Resource;
    7.  
    8. /**
    9.  * 消费者
    10.  */
    11. @MessageConsumer
    12. @Slf4j
    13. public class RedisMqConsumer {
    14.     @Resource
    15.     RedisTemplate redisTemplate;
    16.  
    17.     @MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)
    18.     public void testTopic1(Message message) {
    19.         log.info("工作者1===> " + message);
    20.     }
    21.  
    22.     @MessageListener(topic = "topic1", mode = MessageListener.Mode.TOPIC)
    23.     public void testTopic11(Message message) {
    24.         log.info("工作者2===> " + message);
    25.     }
    26.  
    27.     @MessageListener(topic = "topic2", mode = MessageListener.Mode.TOPIC)
    28.     public void testTopic2(Message message) {
    29.         log.info("topic2===> " + message);
    30.     }
    31.  
    32.     @MessageListener(topic = "topic3", mode = MessageListener.Mode.TOPIC)
    33.     public void testTopic3(Message message) {
    34.         log.info("topic3===> " + message);
    35.     }
    36.  
    37.     @MessageListener(channel = "pubsub.test", mode = MessageListener.Mode.PUBSUB)
    38.     public void testPubsub1(Message message) {
    39.         log.info("pubsub1===> " + message);
    40.     }
    41.  
    42.     @MessageListener(channel = "pubsub.*", mode = MessageListener.Mode.PUBSUB)
    43.     public void testPubsub2(Message message) {
    44.         log.info("pubsub2===> " + message);
    45.     }
    46.  
    47.  
    48.     @MessageListener(streamKey = "streamKey", consumerGroup = "consumerGroup", consumerName = "consumerName", mode = MessageListener.Mode.STREAM)
    49.     public void testStream1(ObjectRecord message) {
    50.         log.info("testStream1 组 consumerGroup 名 consumerName 消费消息 ===> " + message);
    51.         // 手动 ack
    52.         redisTemplate.opsForStream().acknowledge("consumerGroup", message);
    53.     }
    54.  
    55.     /**
    56.      * 同一个组，不同的 consumerName 都可以进行消息的消费，与上面 testStream() 为竞争关系，消息仅被其中一个消费
    57.      *
    58.      * @param message msg
    59.      */
    60.     @MessageListener(streamKey = "streamKey", consumerGroup = "consumerGroup", consumerName = "consumerName2", mode = MessageListener.Mode.STREAM)
    61.     public void testStream2(ObjectRecord message) {
    62.         log.info("testStream2  组 consumerGroup 名 consumerName2 消费消息 ===> " + message);
    63. //        redisTemplate.opsForStream().acknowledge("consumerGroup", message);
    64.     }
    65.  
    66.     /**
    67.      * 不同组的消息，可以正常消费
    68.      *
    69.      * @param message message
    70.      */
    71.     @MessageListener(streamKey = "streamKey", consumerGroup = "consumerGroup3", consumerName = "consumerName3", mode = MessageListener.Mode.STREAM)
    72.     public void testStream3(ObjectRecord message) {
    73.         log.info("testStream3 组 consumerGroup3 名 consumerName3 消费消息 ===> " + message);
    74.     }
    75.  
    76.     /**
    77.      * 消费 streamKey 中，没有进行 ack 的消息
    78.      *
    79.      * @param message message
    80.      */
    81.     @MessageListener(
    82.             streamKey = "streamKey",
    83.             consumerGroup = "consumerGroup",
    84.             consumerName = "consumerName",
    85.             pending = true,
    86.             mode = MessageListener.Mode.STREAM)
    87.     public void testStreamPending(ObjectRecord message) {
    88.         log.info("testStreamPending 组 consumerGroup 名 consumerName 消费未 ack 消息 ===> " + message);
    89.         redisTemplate.opsForStream().acknowledge("consumerGroup", message);
    90.     }
    91. }

     处理者

    1. import lombok.extern.slf4j.Slf4j;
    2. import org.springframework.data.redis.connection.stream.*;
    3. import org.springframework.data.redis.core.RedisTemplate;
    4. import org.springframework.data.redis.core.StreamOperations;
    5. import org.springframework.data.redis.stream.StreamMessageListenerContainer;
    6.  
    7.  
    8. import javax.annotation.Resource;
    9. import java.lang.reflect.InvocationTargetException;
    10. import java.lang.reflect.Method;
    11.  
    12. /**
    13.  * Stream方式 实现消息队列
    14.  */
    15. @MessageHandler(value = MessageListener.Mode.STREAM)
    16. @Slf4j
    17. public class StreamMessageHandler extends AbstractMessageHandler {
    18.     @Resource
    19.     StreamMessageListenerContainer> redisListenerContainer;
    20.  
    21.     public StreamMessageHandler(RedisTemplate redisTemplate) {
    22.         super(redisTemplate);
    23.     }
    24.  
    25.     @Override
    26.     public synchronized void invokeMessage(Method method) {
    27.         // 获取注解中的相关信息
    28.         MessageListener annotation = method.getAnnotation(MessageListener.class);
    29.         String streamKey = annotation.streamKey();
    30.         String consumerGroup = annotation.consumerGroup();
    31.         String consumerName = annotation.consumerName();
    32.         boolean pending = annotation.pending();
    33.  
    34.         // 检测组是否存在
    35.         this.checkAndCreatGroup(streamKey, consumerGroup);
    36.  
    37.         // 获取 StreamOffset
    38.         StreamOffset offset = this.getOffset(streamKey, pending);
    39.  
    40.         // 创建消费者
    41.         Consumer consumer = Consumer.from(consumerGroup, consumerName);
    42.  
    43.         StreamMessageListenerContainer.StreamReadRequest streamReadRequest =
    44.                 StreamMessageListenerContainer
    45.                         .StreamReadRequest.builder(offset)
    46.                         .errorHandler((error) -> log.error("redis stream 异常 ：{}", error.getMessage()))
    47.                         .cancelOnError(e -> false)
    48.                         .consumer(consumer)
    49.                         //关闭自动ack确认
    50.                         .autoAcknowledge(false)
    51.                         .build();
    52.  
    53.         // 指定消费者对象注册到容器中去
    54.         redisListenerContainer.register(streamReadRequest, (message) -> {
    55.             Class declaringClass = method.getDeclaringClass();
    56.             Object bean = applicationContext.getBean(declaringClass);
    57.             try {
    58.                 method.invoke(bean, message);
    59.             } catch (IllegalAccessException | InvocationTargetException e) {
    60.                 e.printStackTrace();
    61.             }
    62.         });
    63.     }
    64.  
    65.     private StreamOffset getOffset(String streamKey, boolean pending) {
    66.         if (pending) {
    67.             // 获取尚未 ack 的消息
    68.             return StreamOffset.create(streamKey, ReadOffset.from("0"));
    69.         }
    70.         // 指定消费最新的消息
    71.         return StreamOffset.create(streamKey, ReadOffset.lastConsumed());
    72.     }
    73.  
    74.     /**
    75.      * 校验消费组是否存在，不存在则创建，否则会出现异常
    76.      * Error in execution; nested exception is io.lettuce.core.RedisCommandExecutionException: NOGROUP No such key 'streamKey' or consumer group 'consumerGroup' in XREADGROUP with GROUP option
    77.      *
    78.      * @param streamKey     streamKey
    79.      * @param consumerGroup consumerGroup
    80.      */
    81.     private void checkAndCreatGroup(String streamKey, String consumerGroup) {
    82.         if (Boolean.TRUE.equals(redisTemplate.hasKey(streamKey))) {
    83.             StreamOperations streamOperations = redisTemplate.opsForStream();
    84.             StreamInfo.XInfoGroups groups = streamOperations.groups(streamKey);
    85.             if (groups.isEmpty() || groups.stream().noneMatch(data -> consumerGroup.equals(data.groupName()))) {
    86.                 creatGroup(streamKey, consumerGroup);
    87.             } else {
    88.                 groups.stream().forEach(g -> {
    89.                     log.info("XInfoGroups:{}", g);
    90.                     StreamInfo.XInfoConsumers consumers = streamOperations.consumers(streamKey, g.groupName());
    91.                     log.info("XInfoConsumers:{}", consumers);
    92.                 });
    93.             }
    94.         } else {
    95.             creatGroup(streamKey, consumerGroup);
    96.         }
    97.     }
    98.  
    99.     private void creatGroup(String key, String group) {
    100.         StreamOperations streamOperations = redisTemplate.opsForStream();
    101.         String groupName = streamOperations.createGroup(key, group);
    102.         log.info("创建组成功:{}", groupName);
    103.     }
    104. }

    生产者

    1.  
    2. import cn.hutool.json.JSONUtil;
    3. import org.springframework.data.redis.connection.stream.ObjectRecord;
    4. import org.springframework.data.redis.connection.stream.RecordId;
    5. import org.springframework.data.redis.core.RedisTemplate;
    6. import org.springframework.web.bind.annotation.GetMapping;
    7. import org.springframework.web.bind.annotation.RequestMapping;
    8. import org.springframework.web.bind.annotation.RestController;
    9.  
    10. import javax.annotation.Resource;
    11. import java.util.UUID;
    12.  
    13. @RestController
    14. @RequestMapping("/test")
    15. public class ProductController {
    16.  
    17.  
    18.  
    19.     @Resource
    20.     RedisTemplate redisTemplate;
    21.  
    22.     @GetMapping("/listMessage")
    23.     public void listMessage() {
    24.         Message message = new Message<>();
    25.         // 随机生成id
    26.         message.setId(UUID.randomUUID().toString());
    27.         // 消息内容
    28.         message.setContent("0001");
    29.         redisTemplate.convertAndSend("pubsub.122", JSONUtil.toJsonStr(message));
    30. //        redisTemplate.opsForList().leftPush("topic1", JSONUtil.toJsonStr(message));
    31.     }
    32.     @GetMapping("/test-stream")
    33.     public String stream() {
    34.         Message message = new Message<>();
    35.         message.setId(UUID.randomUUID().toString());
    36.         message.setContent("0001");
    37.         ObjectRecord record = ObjectRecord.create("streamKey", JSONUtil.toJsonStr(message));
    38.         // 将消息添加至消息队列中
    39.         RecordId recordId = redisTemplate.opsForStream().add(record);
    40.         return recordId == null ? "null" : recordId.getValue();
    41.     }
    42. }

    测试

    生产者发送消息，消费组1和消费者2获取到消息，并指派一个消费者去消费

    总结

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

    • 消息可回溯
    • 可以多消费者争抢消息，加快消费速度
    • 可以阻塞读取
    • 没有消息漏读的风险
    • 有消息确认机制，保证消息至少被消费一次
  • 相关阅读:
    VUE页面添加水印
    Vue插槽slot详解
    【Proteus仿真】【STM32单片机】STM32脉搏血氧仪
    Nacos官网 镜像及使用记录
    编译原理--基本块的划分
    Cpp/Qt-day040920Qt
    （一）H264视频解码问题：出现部分绿屏问题的解决
    网络安全（黑客）自学
    基于SSM的酒店客房管理系统设计与实现
    实时数仓：美团点评Flink的实时数仓应用分享
  • 原文地址：https://blog.csdn.net/qq_63431773/article/details/133890638