示例项目地址:
https://git.code.tencent.com/xinzhang0618/oa2.git
feature/xinzhang_rabbitmq分支

参考文档:
mq概念(这俩篇必读):
https://www.yuque.com/fvy7xd/xinzhang/toi6on
https://www.yuque.com/fvy7xd/xinzhang/if5xfk
延时队列: https://zhuanlan.zhihu.com/p/130417736
基本使用: https://zhuanlan.zhihu.com/p/265732383

思路

rabbitmq的玩法比较丰富, 包括但不限于: 异步, 微服务通信, 多线程处理提高性能, 延时等.
难点在于一是本身的概念配置丰富, 二是不同的业务场景玩法比较花, 本篇着重于梳理mq的最佳简易配置, 并指出各功能点针对的应用场景.
在看本篇之前, 建议熟悉rabbitmq中概念, 并熟悉基本使用.

解析

部署略, 注意:

  1. 需要添加管理界面用户
  2. 启动管理插件
  3. 安装并启用死信交换机插件
  4. 当exchange/queue为durable时, 若有配置变更, 需要删掉原来的使程序重新生成

常用命令

  1. - rabbitmqctl 
  2. rabbitmqctl --help 查看帮助
  3. rabbitmqctl status 查看状态
  4. rabbitmqctl shutdown/stop/stop_app 停止
  5. rabbitmqctl start_app 启动
  7. - rabbitmq-plugins  
  8. rabbitmq-plugins --help 查看帮助
  9. rabbitmq-plugins list 查看插件列表
  10. rabbitmq-plugins enable <pluginName> 启用插件
  11. rabbitmq-plugins disable <pluginName> 禁用插件
  13. 部署补充:
  14. 1.需启用管理插件, rabbitmq_plugins enable rabbitmq_management
  15. 2.需安装并启用死信交换机插件 
  16. - 上 https://www.rabbitmq.com/community-plugins.html 找 rabbitmq_delayed_message_exchange
  17. - 下载rabbitmq_delayed_message_exchange-3.8.0.ez扔到/usr/lib/rabbitmq/plugins/下
  18. - 启用rabbitmq-plusgins enable rabbitmq_delayed_message_exchange

配置

注: 示例项目为单体多模块, 交换机都使用topic类型, 项目结构如下图

依赖

oa-core, oa-consumer, oa-biz三个模块都需配置

  1.  <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter-web</artifactId>
  4.         </dependency>
  5.         <dependency>
  6.             <groupId>org.springframework.boot</groupId>
  7.             <artifactId>spring-boot-starter-amqp</artifactId>
  8.         </dependency>
  9.         <dependency>
  10.             <groupId>org.springframework.boot</groupId>
  11.             <artifactId>spring-boot-starter-test</artifactId>
  12.         </dependency>
  13.         <dependency>
  14.             <groupId>com.alibaba</groupId>
  15.             <artifactId>fastjson</artifactId>
  16.         </dependency>

基础配置

以下配置都在oa-core模块下, 这部分主要是对mq的通用配置以及template的封装

Message抽象类, 包含优先级, 延迟分钟, 租户等属性, 具体实现类需要指定交换机以及路由key

  1. package top.xinzhang0618.oa.amqp;
  4. import top.xinzhang0618.oa.BizContext;
  6. /**
  7.  * MQ消息抽象类 实现类需要提供消息发送的exchange，避免发送到默认exchange 实现类需要提供消息的routingKey，用于消息路由
  8.  */
  9. public abstract class Message {
  11.     /**
  12.      * 优先级
  13.      */
  14.     private Integer priority;
  15.     /**
  16.      * 延迟分钟
  17.      */
  18.     private Integer delayMinutes;
  19.     /**
  20.      * 租户ID
  21.      */
  22.     private Long tenantId;
  24.     public Message() {
  25.         this.tenantId = BizContext.getTenantId();
  26.     }
  28.     public Long getTenantId() {
  29.         return tenantId;
  30.     }
  32.     public void setTenantId(Long tenantId) {
  33.         this.tenantId = tenantId;
  34.     }
  36.     public Integer getPriority() {
  37.         return priority;
  38.     }
  40.     public void setPriority(Integer priority) {
  41.         this.priority = priority;
  42.     }
  44.     public Integer getDelayMinutes() {
  45.         return delayMinutes;
  46.     }
  48.     public void setDelayMinutes(Integer delayMinutes) {
  49.         this.delayMinutes = delayMinutes;
  50.     }
  52.     /**
  53.      * 返回消息对应的路由器
  54.      *
  55.      * @return ExchangeName
  56.      */
  57.     public String exchange() {
  58.         return "oa";
  59.     }
  61.     /**
  62.      * 返回消息路由键
  63.      *
  64.      * @return RoutingKey
  65.      */
  66.     public abstract String routingKey();
  68.     @Override
  69.     public String toString() {
  70.         return "Message{" +
  71.                 "priority=" + priority +
  72.                 ", delayMinutes=" + delayMinutes +
  73.                 ", tenantId=" + tenantId +
  74.                 '}';
  75.     }
  76. }

MqProducer消息发布接口

  1. package top.xinzhang0618.oa.amqp;
  3. import java.util.Collection;
  5. /**
  6.  * 消息生产者.
  7.  */
  8. public interface MqProducer {
  10.   /**
  11.    * 发送消息.
  12.    */
  13.   void send(Message message);
  15.   /**
  16.    * 批量发送消息.
  17.    */
  18.   void send(Collection<? extends Message> messages);
  19. }

RabbitMqProducer, rabbitmq消息发布的实现
注意: 这里包装了一层SpringEvent来处理mq事务(是否性能更佳存疑…)
rabbbitmq的事务: https://blog.csdn.net/u013256816/article/details/55515234
applicationEvent事务: https://blog.csdn.net/crowhyc/article/details/96433001
rabbitmq原生事务性能差, 因此不用; ApplicationEvent的消费端注解@TransactionalEventListener默认绑定的监听阶段是 TransactionPhase.AFTER_COMMIT即事务提交后, 如果事务回滚了, 消费端是接收不到消息的.

同时, 也aplicationEvent也解决了”生产端事务未提交, 消费端从数据库中查不到数据”这一问题

  1. package top.xinzhang0618.oa.amqp;
  3. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
  6. import org.springframework.context.ApplicationEventPublisher;
  7. import org.springframework.stereotype.Component;
  8. import org.springframework.transaction.event.TransactionalEventListener;
  9. import top.xinzhang0618.oa.Assert;
  11. import javax.annotation.Resource;
  12. import java.util.Collection;
  13. import java.util.List;
  15. /**
  16.  * 消息模板包装类.
  17.  * 这里利用applicationEvent的事务来取代rabbitmq的事务
  18.  */
  19. @Component
  20. @ConditionalOnBean(RabbitTemplate.class)
  21. public class RabbitMqProducer implements MqProducer {
  23.     @Autowired
  24.     private ApplicationEventPublisher applicationEventPublisher;
  26.     /**
  27.      * 发送消息.
  28.      *
  29.      * @param message 消息
  30.      */
  31.     @Override
  32.     public void send(Message message) {
  33.         applicationEventPublisher.publishEvent(new MessageSendingEvent(message.routingKey(), message));
  34.     }
  36.     @Override
  37.     public void send(Collection<? extends Message> messages) {
  38.         messages.forEach(this::send);
  39.     }
  42.     @Component
  43.     public static class RabbitMqMessageSender {
  45.         @Resource
  46.         private RabbitTemplate rabbitTemplate;
  48.         @TransactionalEventListener(fallbackExecution = true, classes = MessageSendingEvent.class)
  49.         public void sendMessage(MessageSendingEvent messageSendingEvent) {
  50.             List<Message> messages = messageSendingEvent.getMessages();
  51.             if (Assert.isEmpty(messages)) {
  52.                 return;
  53.             }
  54.             send(messages);
  55.         }
  57.         private void send(Message message) {
  58.             if (message.getPriority() == null && message.getDelayMinutes() == null) {
  59.                 rabbitTemplate.convertAndSend(message.exchange(), message.routingKey(), message);
  60.                 return;
  61.             }
  62.             rabbitTemplate.convertAndSend(message.exchange(), message.routingKey(), message, msg -> {
  63.                 if (message.getPriority() != null) {
  64.                     msg.getMessageProperties().setPriority(message.getPriority());
  65.                 }
  66.                 if (message.getDelayMinutes() != null) {
  67.                     msg.getMessageProperties().setDelay(1000 * 60 * message.getDelayMinutes());
  68.                 }
  69.                 return msg;
  70.             });
  71.         }
  73.         private void send(List<Message> messages) {
  74.             messages.forEach(this::send);
  75.         }
  76.     }
  77. }

MessageSendingEvent, ApplicationEvent的实现

  1. package top.xinzhang0618.oa.amqp;
  3. import org.springframework.context.ApplicationEvent;
  5. import java.util.Arrays;
  6. import java.util.List;
  8. /**
  9.  * 消息事件.
  10.  */
  11. public class MessageSendingEvent extends ApplicationEvent {
  13.   private final List<Message> messages;
  15.   public MessageSendingEvent(Object source, Message... messages) {
  16.     super(source);
  17.     this.messages = Arrays.asList(messages);
  18.   }
  20.   public List<Message> getMessages() {
  21.     return messages;
  22.   }
  24.   @Override
  25.   public String toString() {
  26.     return "MessageSendingEvent{" +
  27.             "messages=" + messages +
  28.             '}';
  29.   }
  30. }

另外
BaseMqConfig, 整个项目的总交换机配置

  1. package top.xinzhang0618.oa.amqp;
  3. import org.springframework.amqp.core.TopicExchange;
  4. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
  5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  6. import org.springframework.context.annotation.Bean;
  7. import org.springframework.context.annotation.Configuration;
  9. /**
  10.  * @author xinzhang
  11.  * @date 2021/2/8 14:05
  12.  */
  13. @Configuration
  14. public class BaseMqConfig {
  15.     /**
  16.      * 总交换机.
  17.      */
  18.     private static final String EXCHANGE_OA = "oa";
  20.     /**
  21.      * OA总交换机.
  22.      */
  23.     @Bean
  24.     public TopicExchange oaExchange() {
  25.         return new TopicExchange(EXCHANGE_OA);
  26.     }
  28.     @Bean
  29.     public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
  30.         final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
  31.         rabbitTemplate.setMessageConverter(new FastJson2JsonMessageConverter());
  32.         return rabbitTemplate;
  33.     }
  34. }

FastJson2JsonMessageConverter, 序列化转换器, 在BaseMqConfig的生产端rabbitTemplate的Bean中有配置, 消费端的factory也需要配置, 后文会提到

  1. package top.xinzhang0618.oa.amqp;
  3. import com.alibaba.fastjson.JSON;
  4. import org.springframework.amqp.core.Message;
  5. import org.springframework.amqp.core.MessageProperties;
  6. import org.springframework.amqp.support.converter.AbstractMessageConverter;
  7. import org.springframework.amqp.support.converter.ClassMapper;
  8. import org.springframework.amqp.support.converter.DefaultClassMapper;
  9. import org.springframework.amqp.support.converter.MessageConversionException;
  11. /**
  12.  * @author xinzhang
  13.  * @date 2021/2/8 11:13
  14.  */
  15. public class FastJson2JsonMessageConverter extends AbstractMessageConverter {
  16.     public static final String DEFAULT_CHARSET = "UTF-8";
  17.     public static final String DEFAULT_CONTENT_TYPE = "application/json";
  18.     private static final ClassMapper CLASS_MAPPER = new FastJson2JsonMessageConverter.FastJsonClassMapper();
  20.     public FastJson2JsonMessageConverter() {
  21.     }
  23.     public ClassMapper getClassMapper() {
  24.         return CLASS_MAPPER;
  25.     }
  27.     @Override
  28.     protected Message createMessage(Object object, MessageProperties messageProperties) {
  29.         byte[] bytes = JSON.toJSONBytes(object);
  30.         messageProperties.setContentType(DEFAULT_CONTENT_TYPE);
  31.         messageProperties.setContentEncoding(DEFAULT_CHARSET);
  32.         messageProperties.setContentLength(bytes.length);
  33.         this.getClassMapper().fromClass(object.getClass(), messageProperties);
  34.         return new Message(bytes, messageProperties);
  35.     }
  37.     @Override
  38.     public Object fromMessage(Message message) throws MessageConversionException {
  39.         Object content = null;
  40.         MessageProperties properties = message.getMessageProperties();
  41.         if (properties != null) {
  42.             String contentType = properties.getContentType();
  43.             if (DEFAULT_CONTENT_TYPE.equals(contentType)) {
  44.                 Class<?> targetClass = this.getClassMapper().toClass(message.getMessageProperties());
  45.                 content = JSON.parseObject(message.getBody(), targetClass);
  46.             }
  47.         }
  49.         if (content == null) {
  50.             content = message.getBody();
  51.         }
  53.         return content;
  54.     }
  56.     public static class FastJsonClassMapper extends DefaultClassMapper {
  57.         public FastJsonClassMapper() {
  58.             this.setTrustedPackages("*");
  59.         }
  60.     }
  61. }

消费端配置

以下配置在oa-consumer模块, 这部分主要是对rabbitmq监听容器的配置

配置文件

  1. spring:
  2.     rabbitmq:
  3.     host: 47.94.148.180
  4.     port: 5672
  5.     username: bite
  6.     password: bite
  7.     virtual-host: xinzhang
  8. oa:
  9.   consumer:
  10.     demo:
  11.       exchange:
  12.         concurrent: 5
  13.         maxConcurrent: 10

启动类添加注解

  1. @EnableAsync
  2. @EnableRabbit
  3. @SpringBootApplication(scanBasePackages = "top.xinzhang0618.oa")
  4. public class ConsumerApplication {
  6.     public static void main(String[] args) {
  7.         SpringApplication.run(ConsumerApplication.class, args);
  8.     }
  10. }

ConsumerConfig, 消费端factory配置
注意:

  1. 核心参数是prefetchCount, concurrent, maxConcurrent, 后两个从配置文件中读取
  2. factory配置了fastJson序列化
  3. (confirm默认关闭, ack默认开启)这里开启了手动ack, 正常场景直接自动ack就行, ack的应用场景下文会特别提到 ```java package top.xinzhang0618.oa.config;

import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import top.xinzhang0618.oa.amqp.FastJson2JsonMessageConverter;

/**

  • @author xinzhang

  • @date 2021/2/8 15:27 */ @Configuration public class ConsumerConfig { @Value(“${oa.consumer.demo.exchange.concurrent:1}”) private Integer concurrent; @Value(“${oa.consumer.demo.exchange.maxConcurrent:1}”) private Integer maxConcurrent;

    @Bean public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {

    1.  SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    2.  factory.setConnectionFactory(connectionFactory);
    3.  factory.setMessageConverter(new FastJson2JsonMessageConverter());
    4.  factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    5.  // 参考: https://www.cnblogs.com/throwable/p/13834465.html
    6.  // prefetchCount表示手动ack情况下队列未ack的预取或说阻塞的消息的最大数, 该参数用来使消费端负载均衡, 默认250, 官方推荐30
    7.  factory.setPrefetchCount(30);
    8.  // 并发消费者数, 即开多线程消费, 比如2个consumer则默认单线程即2个channel, 将concurrent设置为5后, 变为10个channel
    9.  factory.setConcurrentConsumers(concurrent);
    10.  factory.setMaxConcurrentConsumers(maxConcurrent);
    11.  return factory;

    } }

  1. **AbstractConsumer, consumer基类**<br />作用: 1. 做统一异常以及日志处理;  2. 做手动ack等
  2. ```java
  3. package top.xinzhang0618.oa.consumer;
  5. import com.alibaba.fastjson.JSON;
  6. import com.rabbitmq.client.Channel;
  7. import org.slf4j.Logger;
  8. import org.slf4j.LoggerFactory;
  9. import top.xinzhang0618.oa.amqp.Message;
  11. import java.io.IOException;
  13. /**
  14.  * @author xinzhang
  15.  * @date 2021/2/19 13:59
  16.  */
  17. public abstract class AbstractConsumer {
  18.     private static final Logger LOGGER = LoggerFactory.getLogger(AbstractConsumer.class);
  20.     protected <T extends Message> void run(T message, final Channel channel, long tag,
  21.                                            Runnable runnable) throws IOException {
  22.         try {
  23.             if (LOGGER.isDebugEnabled()) {
  24.                 LOGGER.debug("{}处理消息：{}", this.getClass().getName(), message);
  25.             }
  26.             runnable.run();
  27.             channel.basicAck(tag, false);
  28.         } catch (Exception e) {
  29.             LOGGER.error("消息处理异常！{},{}", this.getClass().getName(), JSON.toJSONString(message));
  30.             LOGGER.error("消息处理异常！", e);
  31.             channel.basicReject(tag, false);
  32.         }
  33.     }
  34. }

消费者示例
Test1Consumer

  1. package top.xinzhang0618.oa.consumer;
  3. import com.alibaba.fastjson.JSON;
  4. import com.rabbitmq.client.Channel;
  5. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.amqp.support.AmqpHeaders;
  8. import org.springframework.messaging.handler.annotation.Header;
  9. import org.springframework.stereotype.Component;
  10. import top.xinzhang0618.oa.amqp.demo.MqConstants;
  11. import top.xinzhang0618.oa.amqp.demo.TestMessage1;
  12. import top.xinzhang0618.oa.amqp.demo.TestMessage3;
  14. import java.io.IOException;
  15. import java.time.LocalDateTime;
  17. /**
  18.  * @author xinzhang
  19.  * @date 2021/2/8 15:10
  20.  */
  21. @Component
  22. @RabbitListener(queues = MqConstants.QUEUE_TEST1)
  23. public class Test1Consumer extends AbstractConsumer {
  25.     @RabbitHandler(isDefault = true)
  26.     public void test(TestMessage1 testMessage1, Channel channel,
  27.                      @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
  28.         run(testMessage1, channel, tag, () -> {
  29.             System.out.println("test1队列收到消息了, 消息内容为: " + JSON.toJSON(testMessage1) + ", 消息接收时间为: " + LocalDateTime.now());
  30.         });
  31.     }
  33.     @RabbitHandler
  34.     public void test3(TestMessage3 testMessage3, Channel channel,
  35.                       @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
  36.         run(testMessage3, channel, tag, () -> {
  37.             System.out.println("test1队列收到消息了, 消息内容为: " + JSON.toJSON(testMessage3) + ", 消息接收时间为: " + LocalDateTime.now());
  38.         });
  39.     }
  40. }

Test2Consumer

  1. package top.xinzhang0618.oa.consumer;
  3. import com.alibaba.fastjson.JSON;
  4. import com.rabbitmq.client.Channel;
  5. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.amqp.support.AmqpHeaders;
  8. import org.springframework.messaging.handler.annotation.Header;
  9. import org.springframework.stereotype.Component;
  10. import top.xinzhang0618.oa.amqp.demo.MqConstants;
  11. import top.xinzhang0618.oa.amqp.demo.TestMessage2;
  13. import java.io.IOException;
  14. import java.time.LocalDateTime;
  16. /**
  17.  * @author xinzhang
  18.  * @date 2021/2/8 15:10
  19.  */
  20. @Component
  21. @RabbitListener(queues = MqConstants.QUEUE_TEST2)
  22. public class Test2Consumer extends AbstractConsumer {
  24.     @RabbitHandler
  25.     public void test(TestMessage2 testMessage2, Channel channel,
  26.                      @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
  27.         run(testMessage2, channel, tag, () -> {
  28.             System.out.println("test2队列收到消息了, 消息内容为: " + JSON.toJSON(testMessage2) + ", 消息接收时间为: " + LocalDateTime.now());
  29.         });
  30.     }
  31. }

业务端配置

以下配置在oa-biz模块, 这部分跟业务紧密相关, 主要是业务交换机以及队列的配置

示例配置
DemoMqConfig

  1. package top.xinzhang0618.oa.amqp.demo;
  3. import org.springframework.amqp.core.*;
  4. import org.springframework.beans.factory.annotation.Qualifier;
  5. import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
  6. import org.springframework.context.annotation.Bean;
  7. import org.springframework.context.annotation.Configuration;
  8. import top.xinzhang0618.oa.amqp.BaseMqConfig;
  10. /**
  11.  * 交换机配置
  12.  *
  13.  * @author xinzhang
  14.  * @date 2021/2/8 14:35
  15.  */
  16. @ConditionalOnBean(BaseMqConfig.class)
  17. @Configuration
  18. public class DemoMqConfig {
  19.     /**
  20.      * 示例交换机
  21.      */
  22.     @Bean
  23.     public Exchange demoExchange() {
  24.         return ExchangeBuilder.topicExchange(MqConstants.EXCHANGE_DEMO).delayed().build();
  25.     }
  27.     /**
  28.      * 示例交换机绑定总交换机, routingKey: oa.demo.#
  29.      */
  30.     @Bean
  31.     public Binding demoExchangeBindingOaExchange(@Qualifier("oaExchange") TopicExchange oaExchange,
  32.                                                  @Qualifier("demoExchange") Exchange demoExchange) {
  33.         return BindingBuilder.bind(demoExchange).to(oaExchange).with(MqConstants.EXCHANGE_DEMO_BINDING_KEY);
  34.     }
  36.     /**
  37.      * 测试1队列
  38.      */
  39.     @Bean
  40.     public Queue test1Queue() {
  41.         return QueueBuilder.durable(MqConstants.QUEUE_TEST1).build();
  42.     }
  44.     /**
  45.      * 测试1队列 绑定 demo交换机, routingKey: oms.demo.test1
  46.      */
  47.     @Bean
  48.     public Binding test1QueueBindingDemoExchange(@Qualifier("demoExchange") Exchange demoExchange,
  49.                                                  @Qualifier("test1Queue") Queue test1Queue) {
  50.         return BindingBuilder
  51.                 .bind(test1Queue)
  52.                 .to(demoExchange)
  53.                 .with(MqConstants.QUEUE_TEST1_BINDING_KEY)
  54.                 .noargs();
  55.     }
  57.     /**
  58.      * 测试2队列
  59.      */
  60.     @Bean
  61.     public Queue test2Queue() {
  62.         return QueueBuilder.durable(MqConstants.QUEUE_TEST2).build();
  63.     }
  65.     /**
  66.      * 测试2队列 绑定 demo交换机, routingKey: oms.demo.test2
  67.      */
  68.     @Bean
  69.     public Binding test2QueueBindingDemoExchange(@Qualifier("demoExchange") Exchange demoExchange,
  70.                                                  @Qualifier("test2Queue") Queue test1Queue) {
  71.         return BindingBuilder
  72.                 .bind(test1Queue)
  73.                 .to(demoExchange)
  74.                 .with(MqConstants.QUEUE_TEST2_BINDING_KEY)
  75.                 .noargs();
  76.     }
  78.     /**
  79.      * 测试3延时队列, 延迟10min
  80.      * 这个队列没有消费者, 消息过期后由死信交换机demo转发至oa.demo.test1
  81.      */
  82.     @Bean
  83.     public Queue test3DelayQueue() {
  84.         return QueueBuilder
  85.                 .durable(MqConstants.QUEUE_TEST3_DELAY)
  86.                 .withArgument("x-dead-letter-exchange", MqConstants.EXCHANGE_DEMO)
  87.                 .withArgument("x-dead-letter-routing-key", MqConstants.QUEUE_TEST1_BINDING_KEY)
  88.                 .withArgument("x-message-ttl", 1000 * 60 * 10)
  89.                 .build();
  90.     }
  92.     /**
  93.      * 测试3延时队列 绑定 demo交换机, routingKey: oms.demo.test3.delay
  94.      */
  95.     @Bean
  96.     public Binding test3DelayQueueBindingDemoExchange(@Qualifier("demoExchange") Exchange demoExchange,
  97.                                                       @Qualifier("test3DelayQueue") Queue test3DelayQueue) {
  98.         return BindingBuilder
  99.                 .bind(test3DelayQueue)
  100.                 .to(demoExchange)
  101.                 .with(MqConstants.QUEUE_TEST3_DELAY_BINDING_KEY)
  102.                 .noargs();
  103.     }
  104. }

MqConstants

  1. package top.xinzhang0618.oa.amqp.demo;
  3. /**
  4.  * @author xinzhang
  5.  * @date 2021/2/8 14:14
  6.  */
  7. public class MqConstants {
  9.     /**
  10.      * 示例交换机.
  11.      */
  12.     public static final String EXCHANGE_DEMO = "oa.demo";
  13.     public static final String EXCHANGE_DEMO_BINDING_KEY = "oa.demo.#";
  15.     /**
  16.      * 测试队列1
  17.      */
  18.     public static final String QUEUE_TEST1 = "oa.demo.test1";
  19.     public static final String QUEUE_TEST1_BINDING_KEY = "oa.demo.test1";
  21.     /**
  22.      * 测试队列2
  23.      */
  24.     public static final String QUEUE_TEST2 = "oa.demo.test2";
  25.     public static final String QUEUE_TEST2_BINDING_KEY = "oa.demo.test2";
  27.     /**
  28.      * 测试队列3, 延时队列
  29.      */
  30.     public static final String QUEUE_TEST3_DELAY = "oa.demo.test3.delay";
  31.     public static final String QUEUE_TEST3_DELAY_BINDING_KEY = "oa.demo.test3.delay";
  32. }

TestMessage1

  1. package top.xinzhang0618.oa.amqp.demo;
  3. import top.xinzhang0618.oa.amqp.Message;
  5. /**
  6.  * @author xinzhang
  7.  * @date 2021/2/4 18:15
  8.  */
  9. public class TestMessage1 extends Message {
  10.     private String msg;
  12.     @Override
  13.     public String exchange() {
  14.         return super.exchange();
  15.     }
  17.     @Override
  18.     public String routingKey() {
  19.         return "oa.demo.test1";
  20.     }
  22.     public String getMsg() {
  23.         return msg;
  24.     }
  26.     public void setMsg(String msg) {
  27.         this.msg = msg;
  28.     }
  30.     @Override
  31.     public String toString() {
  32.         return "TestMessage1{" +
  33.                 "msg='" + msg + '\'' +
  34.                 '}';
  35.     }
  36. }

TestMessage2

  1. package top.xinzhang0618.oa.amqp.demo;
  3. import top.xinzhang0618.oa.amqp.Message;
  5. /**
  6.  * @author xinzhang
  7.  * @date 2021/2/4 18:15
  8.  */
  9. public class TestMessage2 extends Message {
  10.     private String msg;
  12.     @Override
  13.     public String exchange() {
  14.         return super.exchange();
  15.     }
  17.     @Override
  18.     public String routingKey() {
  19.         return "oa.demo.test2";
  20.     }
  22.     public String getMsg() {
  23.         return msg;
  24.     }
  26.     public void setMsg(String msg) {
  27.         this.msg = msg;
  28.     }
  30.     @Override
  31.     public String toString() {
  32.         return "TestMessage2{" +
  33.                 "msg='" + msg + '\'' +
  34.                 '}';
  35.     }
  36. }

TestMessage3

  1. package top.xinzhang0618.oa.amqp.demo;
  3. import top.xinzhang0618.oa.amqp.Message;
  5. /**
  6.  * @author xinzhang
  7.  * @date 2021/2/4 18:15
  8.  */
  9. public class TestMessage3 extends Message {
  10.     private String msg;
  12.     @Override
  13.     public String exchange() {
  14.         return super.exchange();
  15.     }
  17.     @Override
  18.     public String routingKey() {
  19.         return "oa.demo.test3.delay";
  20.     }
  22.     public String getMsg() {
  23.         return msg;
  24.     }
  26.     public void setMsg(String msg) {
  27.         this.msg = msg;
  28.     }
  30.     @Override
  31.     public String toString() {
  32.         return "TestMessage2{" +
  33.                 "msg='" + msg + '\'' +
  34.                 '}';
  35.     }
  36. }

测试

至此， 基础的配置都已完成，后续我们再来针对一个个功能点做分析， 先做简单的消息测试，配置完成后结构如下
image.png
启动消费者服务, 发送TestMessage1和TestMessage2, 两个消费者能够正常接收到消息并消费

  1. @Test
  2.     public void test() {
  3.         TestMessage1 testMessage1 = new TestMessage1();
  4.         testMessage1.setMsg("这里是test1测试消息, 消息发送时间为: " + LocalDateTime.now());
  5.         TestMessage2 testMessage2 = new TestMessage2();
  6.         testMessage2.setMsg("这里是test2测试消息, 消息发送时间为: " + LocalDateTime.now());
  7.         producer.send(testMessage1);
  8.         producer.send(testMessage2);
  9.     }
  10. -----------------------------
  11. 14:35:37.214 DEBUG [SimpleAsyncTaskExecutor-4] t.x.oa.consumer.AbstractConsumer - top.xinzhang0618.oa.consumer.Test1Consumer处理消息：TestMessage1{msg='这里是test1测试消息, 消息发送时间为: 2021-02-19T14:35:37.089'}
  12. test1队列收到消息了, 消息内容为: {"msg":"这里是test1测试消息, 消息发送时间为: 2021-02-19T14:35:37.089"}, 消息接收时间为: 2021-02-19T14:35:37.214
  13. 14:35:37.214 DEBUG [SimpleAsyncTaskExecutor-5] t.x.oa.consumer.AbstractConsumer - top.xinzhang0618.oa.consumer.Test2Consumer处理消息：TestMessage2{msg='这里是test2测试消息, 消息发送时间为: 2021-02-19T14:35:37.089'}
  14. test2队列收到消息了, 消息内容为: {"msg":"这里是test2测试消息, 消息发送时间为: 2021-02-19T14:35:37.089"}, 消息接收时间为: 2021-02-19T14:35:37.214

消息的准确投递问题

oms参考方案如下:
消息全流程: 生产端—>broker—>消费端

  • 使用了applicationEvent取代mq事务来保证消息从生产端—>broker的准确投递(若事务回滚则mq不会发送消息)
  • 消费端若消费失败
    • 利用手动ack对特定异常, 进行消息回队, 等待再次消费
    • 利用定时器进行业务补偿, 比如因为缺货或并发导致配货失败, 会有”缺货重配”, “定时扫表再次配货”等定时任务

发布者确认以及消费者应答

参考文档: https://www.rabbitmq.com/confirms.html
https://www.zhihu.com/question/41976893

发布者确认一般用于事务控制, 默认关闭, 这里没用到
消费者应答默认开启, 这里采用手动ack的方式
ack机制能灵活的控制消息的消费, 拒绝后可以选择性的删除或者重新入队, 比如oms中使用consumer ack来实现
“如果消费端爆出版本异常(乐观锁), 则消息重新入队(后续再被消费), 其余情况则直接丢弃” 这一功能

使用ack示例代码如下:

  1. 先打开手动ack

    1. factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

  2. 在AbstractConsumer中针对特殊异常做处理 ```java package com.greatonce.oms.consumer;

import com.greatonce.core.util.JsonUtil; import com.greatonce.oms.domain.OmsException; import com.greatonce.oms.domain.SysExceptions; import com.greatonce.oms.message.Message; import com.greatonce.oms.util.consumer.MessageListenerContainerManager; import com.rabbitmq.client.Channel; import java.io.IOException; import javax.annotation.PostConstruct; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired;

/**

  • 消息处理抽象类. *
  • @author ginta
  • @author Shenzhen Greatonce Co Ltd

  • @version 2018/3/23 */ public abstract class AbstractConsumer {

    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractConsumer.class);

    @Autowired private MessageListenerContainerManager messageListenerContainerManager;

    @PostConstruct protected void init() { messageListenerContainerManager.registerContainer(containerId(), containerName()); }

    protected abstract String containerId();

    protected abstract String containerName();

    protected void run(T message, final Channel channel, long tag, Runnable runnable) throws IOException { try { if (LOGGER.isDebugEnabled()) {

    1.  LOGGER.debug("{}处理消息：{}", this.getClass().getName(), message);

    } runnable.run(); channel.basicAck(tag, false); } catch (OmsException e) { LOGGER.error(“消息处理异常！{},{},{}”, this.getClass().getName(), JsonUtil.toJson(message), e.getMessage()); LOGGER.error(“消息处理异常！”, e); if (SysExceptions.VERSION_CHANGED.equals(e.getMessage())

    1.    || SysExceptions.MALL_SECURITY_API_ERROR.equals(e.getMessage())) {
    2.  channel.basicReject(tag, true);

    } else {

    1.  channel.basicReject(tag, false);

    } } catch (Exception e) { LOGGER.error(“消息处理异常！{},{}”, this.getClass().getName(), JsonUtil.toJson(message)); LOGGER.error(“消息处理异常！”, e); channel.basicReject(tag, false); } } }

  1. > basicReject与basicNack方法区别: 
  2. > basicReject：是接收端告诉服务器这个消息我拒绝接收,不处理,可以设置是否放回到队列中还是丢掉，而且只能一次拒绝一个消息,官网中有明确说明不能批量拒绝消息，为解决批量拒绝消息才有了basicNack。
  3. > basicNack：可以一次拒绝N条消息，客户端可以设置basicNack方法的multiple参数为true，服务器会拒绝指定了delivery_tag的所有未确认的消息.
  4. > 
  5. > DeliveryTag作用:
  6. > 对于每个Channel来说，每个消息都会有一个DeliveryTag，一般用接收消息的顺序来表示：1,2,3,4 等等
  9. <a name="qvOni"></a>
  10. ### 消息的优先级
  11. 只有在消费端消息堆积的时候才有效果, 不然也没卵用, 比如oms中转化队列配置了"下载的销售单"比"从第三方系统同步的销售单"更优先进行转化
  13. 使用示例:
  15. 1. 先设置队列的优先级, 表示能容纳的消息的最大优先级, 数字越大越优先
  16. ```java
  17.  @Bean
  18.   public Queue salesTranslateQueue() {
  19.     return QueueBuilder
  20.         .durable(QUEUE_TRADE_SALES_ORDER_TRANSLATE)
  21.         .withArgument("x-max-priority", 2)
  22.         .build();
  23.   }

  1. 然后设置消息的优先级

    1. /**
    2. * 销售单从第三方系统同步消息.
    3. *
    4. * @author ginta
    5. * @author Shenzhen Greatonce Co Ltd
    6. * @version 2018/3/7
    7. */
    8. public class MallSalesOrderSynchronizedMessage extends MallSalesOrderMessage {
    10. public MallSalesOrderSynchronizedMessage(Long mallSalesOrderId, Long storeId, String tradeId) {
    11.  super(mallSalesOrderId, storeId, tradeId,"synchronized");
    12.  this.setPriority(1);
    13. }
    14. }
    1. /**
    2. * 销售单已下载消息.
    3. *
    4. * @author ginta
    5. * @author Shenzhen Greatonce Co Ltd
    6. * @version 2018/3/7
    7. */
    8. public class MallSalesOrderDownloadMessage extends MallSalesOrderMessage {
    10. public MallSalesOrderDownloadMessage(Long mallSalesOrderId, Long storeId, String tradeId) {
    11.  super(mallSalesOrderId, storeId, tradeId,"download");
    12.  this.setPriority(2);
    13. }
    14. }

  2. 在配置上, 上面的RabbitMqProducer示例代码中已经将priority属性设置进MessageProperties了

    1. rabbitTemplate.convertAndSend(message.exchange(), message.routingKey(), message, msg -> {
    2.              if (message.getPriority() != null) {
    3.                  msg.getMessageProperties().setPriority(message.getPriority());
    4.              }
    5.              if (message.getDelayMinutes() != null) {
    6.                  msg.getMessageProperties().setDelay(1000 * 60 * message.getDelayMinutes());
    7.              }
    8.              return msg;
    9.          });

延迟消息

利用队列的ttl以及死信队列实现

前言
这里也可以利用消息的ttl实现, 需要设置消息的expiration属性
补充: 当队列和消息同时有过期时间时, 更短的一个会生效

但有个严重问题是, 消息会在队列进行排队顺序消费, 若前一消息的ttl大于后一消息, 则后一消息到期后依然要等到前面消息消费完
比如按顺序发送”2min过期的消息”, “1min过期的消息”, 消费端会在2min后一起消费这两条消息,
因此在示例项目的配置中(message的封装上, rabbitmqProducer的封装上), 我完全摈弃了消息的ttl这一特性

  1.  if (this.delayMinutes != null) {
  2.         int delayTimemillis = 1000 * 60 * this.delayMinutes;
  3.         if (delayType == DelayType.EXCHANGE) {
  4.           message.getMessageProperties().setDelay(delayTimemillis);
  5.         } else {
  6.           message.getMessageProperties().setExpiration(String.valueOf(delayTimemillis));
  7.         }
  8.       }

使用流程
这一方案, 可以参考上面test3DelayQueue的配置, 这个队列10分钟过期, 没有消费者
消息过期后, 会由”x-dead-letter-exchange”通过”x-dead-letter-routing-key”将消息转发出去, 队列配置如下
因此使用示例:
发送消息到test3DelayQueue队列, 过10分钟后消息过期转发到路由oa.demo.test1, 然后Test1Consumer消费, 实现了消息延迟10分钟消费

  1. /**
  2.      * 测试3延时队列, 延迟10min
  3.      * 这个队列没有消费者, 消息过期后由死信交换机demo转发至oa.demo.test1
  4.      */
  5.     @Bean
  6.     public Queue test3DelayQueue() {
  7.         return QueueBuilder
  8.                 .durable(MqConstants.QUEUE_TEST3_DELAY)
  9.                 .withArgument("x-dead-letter-exchange", MqConstants.EXCHANGE_DEMO)
  10.                 .withArgument("x-dead-letter-routing-key", MqConstants.QUEUE_TEST1_BINDING_KEY)
  11.                 .withArgument("x-message-ttl", 1000 * 60 * 10)
  12.                 .build();
  13.     }
  15.     /**
  16.      * 测试3延时队列 绑定 demo交换机, routingKey: oms.demo.test3.delay
  17.      */
  18.     @Bean
  19.     public Binding test3DelayQueueBindingDemoExchange(@Qualifier("demoExchange") Exchange demoExchange,
  20.                                                       @Qualifier("test3DelayQueue") Queue test3DelayQueue) {
  21.         return BindingBuilder
  22.                 .bind(test3DelayQueue)
  23.                 .to(demoExchange)
  24.                 .with(MqConstants.QUEUE_TEST3_DELAY_BINDING_KEY)
  25.                 .noargs();
  26.     }

从管理界面上也能看到各个标签, 鼠标挪到标签上会有信息显示
image.png
image.png

利用死信交换机插件实现

这一实现更加简单, 安装并启用插件后, 声明交换机的delay属性

一个消息变成死信的条件有： 1.消费被拒绝（basic.reject 或者 basic.nack），并且参数 requeue = false 时 2.消息TTL（存活时间）过期 3.队列达到最大长度 rabbitMQ对于死信消息的处理是：如果配置了死信队列，成为死信的消息会被丢进死信队列，如果没有则被丢弃。

  1. /**
  2.      * 示例交换机
  3.      */
  4.     @Bean
  5.     public Exchange demoExchange() {
  6.         return ExchangeBuilder.topicExchange(MqConstants.EXCHANGE_DEMO).delayed().build();
  7.     }

管理界面也能看到交换机类型变为”x-delayed-message”
image.png
给消息设置一定的延迟时间, 这里也要注意是消息的”x-delay”属性
交换机会等待delay的时间, 然后再将消息转发到对应的队列中

  1. rabbitTemplate.convertAndSend(message.exchange(), message.routingKey(), message, msg -> {
  2.                 if (message.getPriority() != null) {
  3.                     msg.getMessageProperties().setPriority(message.getPriority());
  4.                 }
  5.                 if (message.getDelayMinutes() != null) {
  6.                     msg.getMessageProperties().setDelay(1000 * 60 * message.getDelayMinutes());
  7.                 }
  8.                 return msg;
  9.             });

测试如下

  1. @Test
  2.     public void testSimpleConsume() {
  3.         TestMessage1 testMessage1 = new TestMessage1();
  4.         testMessage1.setDelayMinutes(1);
  5.         testMessage1.setMsg("这里是test1测试消息, 消息发送时间为: " + LocalDateTime.now());
  6.         producer.send(testMessage1);
  7.     }
  8. -------------------
  9. 16:02:33.276 DEBUG [SimpleAsyncTaskExecutor-5] t.x.oa.consumer.AbstractConsumer - top.xinzhang0618.oa.consumer.Test1Consumer处理消息：TestMessage1{msg='这里是test1测试消息, 消息发送时间为: 2021-02-19T16:01:33.152'}
  10. test1队列收到消息了, 消息内容为: {"msg":"这里是test1测试消息, 消息发送时间为: 2021-02-19T16:01:33.152","delayMinutes":1}, 消息接收时间为: 2021-02-19T16:02:33.276

灵活控制消费者开启关闭

自定义注解

通过自定义注解能开启或关闭单个消费者, 由配置文件控制, 更改需要重启应用

  1. import java.lang.annotation.Documented;
  2. import java.lang.annotation.ElementType;
  3. import java.lang.annotation.Retention;
  4. import java.lang.annotation.RetentionPolicy;
  5. import java.lang.annotation.Target;
  6. import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
  8. @Target({ElementType.TYPE, ElementType.METHOD})
  9. @Retention(RetentionPolicy.RUNTIME)
  10. @Documented
  11. @ConditionalOnProperty(name = "oms.consumer.dispatch.enabled", havingValue = "true")
  12. public @interface DispatchOrderCondition {
  14. }

监听容器开关

根据业务区分不同的监听容器, 个业务组下的交换机注册到各监听容器中, 通过http请求方式直接关闭开启监听容器