bean 静态绑定

    1. @Getter
    2. @Configuration
    3. @ConfigurationProperties(prefix = "application.rabbit")
    4. public class RabbitMQProperties {
    6.     @ApiModelProperty("死信交换机默认配置")
    7.     private final DefaultMessage defaultMessage = new DefaultMessage();
    9.     @Data
    10.     @Component
    11.     public static class DefaultMessage {
    12.         private final CustomerExchange exchange = new CustomerExchange();
    13.         private final CustomerQueue queue = new CustomerQueue();
    14.         private final CustomerBinding binding = new CustomerBinding();
    15.     }
    17.     @Data
    18.     @Component
    19.     @ApiModel("消费者交换机")
    20.     public static class CustomerExchange {
    22.         @ApiModelProperty("交换机名称")
    23.         private String name = "defaultExchange";
    25.         @ApiModelProperty("交换机是否持久化")
    26.         private Boolean durable = true;
    28.         @ApiModelProperty("是否自动删除空闲的交换机")
    29.         private Boolean autoDelete = false;
    31.         @ApiModelProperty("对交换机属性进行配置")
    32.         private Map<String, Object> arguments = new HashMap<>(10);
    33.     }
    35.     @Data
    36.     @Component
    37.     @ApiModel("消费者队列")
    38.     public static class CustomerQueue {
    40.         @ApiModelProperty("队列名称")
    41.         private String name = "defaultQueue";
    43.         @ApiModelProperty("队列是否持久化")
    44.         private Boolean durable = true;
    46.         @ApiModelProperty("是否连接独占")
    47.         private Boolean exclusive = false;
    49.         @ApiModelProperty("是否自动删除空闲的交换机")
    50.         private Boolean autoDelete = false;
    52.         @ApiModelProperty("对消息属性进行配置")
    53.         private Map<String, Object> arguments = new HashMap<>(10);
    54.     }
    56.     @Data
    57.     @Component
    58.     public static class CustomerBinding {
    59.         /**
    60.          * 默认路由(#:表示拦截任意字符的路由,*:表示拦截任意的单个字符,其他拦截比如：user.ABC 等自定义路由)
    61.          */
    62.         private String routingKey = "#";
    63.     }
    64. }

    yml

    1.   rabbit:
    2.     enabled: false
    3.     default-message:
    4.       exchange:
    5.         name: defaultExchange
    6.       queue:
    7.         name: defaultQueue
    8.       binding:
    9.         routing-key: "#"
    1. @Configuration
    2. @EnableRabbit
    3. @ConditionalOnExpression("'${application.rabbit.enabled}'.equals('true')")
    4. public class RabbitMqConfiguration {
    6.     private static RabbitMqConfiguration rabbitMqConfiguration;
    8.     @Autowired
    9.     private CachingConnectionFactory connectionFactory;
    11.     @Autowired
    12.     private ObjectMapper objectMapper;
    14.     @Autowired
    15.     private RabbitMQProperties rabbitMQProperties;
    17.     @Autowired
    18.     private Map<String, MessageQueueConfigInterface> messageQueueConfigInterfaceMap;
    20.     @PostConstruct
    21.     public void initTestCreateQueue() {
    22.         messageQueueConfigInterfaceMap.values().forEach(MessageQueueConfigInterface::init);
    23.     }
    25.     @Bean
    26.     public DefaultClassMapper classMapper() {
    27.         DefaultClassMapper classMapper = new DefaultClassMapper();
    28.         classMapper.setTrustedPackages("*");
    29.         return classMapper;
    30.     }
    32.     /**
    33.      * 开启多例模式 用于实现多个不同的生产者回调不同
    34.      *
    35.      * @return RabbitTemplate
    36.      */
    37.     @Bean
    38.     @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    39.     public RabbitTemplate rabbitTemplate() {
    40.         RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    41.         rabbitTemplate.setMessageConverter(jsonMessageConverter());
    42.         rabbitTemplate.setMandatory(true);
    43.         return rabbitTemplate;
    44.     }
    46.     /**
    47.      * producer 消息转换器
    48.      */
    49.     @Bean
    50.     public MessageConverter jsonMessageConverter() {
    51.         Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter(objectMapper);
    52.         messageConverter.setClassMapper(classMapper());
    53.         return messageConverter;
    54.     }
    56.     /**
    57.      * consumer 消息转换器
    58.      */
    59.     @Bean
    60.     public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
    61.         SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    62.         factory.setConnectionFactory(connectionFactory);
    63.         factory.setMessageConverter(new Jackson2JsonMessageConverter());
    64.         return factory;
    65.     }
    67.     /**
    68.      * 1. 创建死信交换机
    69.      * 2. 创建死信队列
    70.      * 3. 绑定关系
    71.      * <p>
    72.      * 死信队列 DLQ Queue (默认消息队列)
    73.      * 消息来源:
    74.      * 消息被拒绝（basic.reject或basic.nack）并且requeue=false.
    75.      * 消息TTL过期
    76.      * 队列达到最大长度（队列满了，无法再添加数据到mq中）
    77.      * 处理方式:
    78.      * 丢弃，如果不是很重要，可以选择丢弃
    79.      * 记录死信入库，然后做后续的业务分析或处理
    80.      * 通过死信队列，由负责监听死信的应用程序进行处理
    81.      *
    82.      * @return Queue
    83.      */
    84.     @Bean
    85.     @Order(1)
    86.     public TopicExchange deadExchange() {
    87.         RabbitMQProperties.CustomerExchange exchange = rabbitMQProperties.getDefaultMessage().getExchange();
    88.         return new TopicExchange(exchange.getName(), exchange.getDurable(), exchange.getAutoDelete(), exchange.getArguments());
    89.     }
    91.     @Bean
    92.     @Order(2)
    93.     public Queue deadQueue() {
    94.         RabbitMQProperties.CustomerQueue queue = rabbitMQProperties.getDefaultMessage().getQueue();
    95.         return new Queue(queue.getName(), queue.getDurable(), queue.getExclusive(), queue.getAutoDelete(), queue.getArguments());
    96.     }
    98.     @Bean
    99.     @Order(3)
    100.     public Binding deadBinding() {
    101.         RabbitMQProperties.CustomerBinding binding = rabbitMQProperties.getDefaultMessage().getBinding();
    102.         return BindingBuilder.bind(deadQueue()).to(deadExchange()).with(binding.getRoutingKey());
    103.     }
    105.     /**
    106.      * 通用交换机配置
    107.      */
    108.     public final static String EXCHANGE_NAME = "CommonMessageExchange";
    109.     public final static String QUEUE_NAME = "CommonMessageQueue";
    110.     public final static String ROUTING_KEY_NAME = "message.ABC";
    112.     @Bean
    113.     public TopicExchange comExchange() {
    114.         return new TopicExchange(EXCHANGE_NAME, true, false);
    115.     }
    117.     @Bean
    118.     public Queue comQueue() {
    119.         Map<String, Object> arguments = new HashMap<>(10);
    120.         // 绑定死信交换机 & 设置超时时间
    121.         arguments.put("x-dead-letter-exchange", rabbitMQProperties.getDefaultMessage().getExchange().getName());
    122.         arguments.put("x-dead-letter-routing-key", rabbitMQProperties.getDefaultMessage().getBinding().getRoutingKey());
    123.         arguments.put("x-message-ttl", 10000);
    124.         arguments.put("x-queue-mode", "lazy");
    125.         return new Queue(QUEUE_NAME, true, false, false, arguments);
    126.     }
    128.     @Bean
    129.     public Binding comBinding() {
    130.         return BindingBuilder.bind(comQueue()).to(comExchange()).with(ROUTING_KEY_NAME);
    131.     }
    132. }

    interface 接口动态绑定

    1. public interface MessageQueueConfigInterface {
    2.     void init();
    3. }

    impl

    2. @Slf4j
    3. @Component
    4. public class EmailQueueImpl implements MessageQueueConfigInterface {
    6.     @Autowired
    7.     private AmqpAdmin amqpAdmin;
    9.     @Autowired
    10.     private RabbitMQProperties rabbitMQProperties;
    12.     // Exchange
    13.     public final static String EMAIL_EXCHANGE_NAME = "EmailMessageExchange";
    14.     // Queue
    15.     public final static String EMAIL_QUEUE_NAME = "EmailMessageQueue";
    16.     // RoutingKey
    17.     public final static String EMAIL_ROUTING_KEY_NAME = "message.email";
    19.     @Override
    20.     public void init() {
    21.         Exchange exchange = new DirectExchange(EMAIL_EXCHANGE_NAME);
    22.         amqpAdmin.declareExchange(exchange);
    23.         amqpAdmin.declareQueue(new Queue(EMAIL_QUEUE_NAME, true));
    24.         //创建绑定规则
    25.         amqpAdmin.declareBinding(new Binding(EMAIL_QUEUE_NAME, Binding.DestinationType.QUEUE, EMAIL_EXCHANGE_NAME, EMAIL_ROUTING_KEY_NAME, getArguments()));
    26.     }
    28.     public Map<String, Object> getArguments() {
    29.         Map<String, Object> arguments = new HashMap<>(10);
    30.         // 绑定死信交换机 & 设置超时时间
    31.         arguments.put("x-dead-letter-exchange", rabbitMQProperties.getDefaultMessage().getExchange().getName());
    32.         arguments.put("x-dead-letter-routing-key", rabbitMQProperties.getDefaultMessage().getBinding().getRoutingKey());
    33.         arguments.put("x-message-ttl", 10000);
    34.         arguments.put("x-queue-mode", "lazy");
    35.         return arguments;
    36.     }
    37. }
    2. @ApiModel("短信交换机队列")
    3. @Configuration
    4. @ConditionalOnExpression("'${application.rabbit.enabled}'.equals('true')")
    5. public class SmsQueueImpl implements MessageQueueConfigInterface{
    7.     @Autowired
    8.     private AmqpAdmin amqpAdmin;
    10.     @Autowired
    11.     private RabbitMQProperties rabbitMQProperties;
    13.     // Exchange
    14.     public final static String SMS_EXCHANGE_NAME = "SmsMessageExchange";
    15.     // Queue
    16.     public final static String SMS_QUEUE_NAME = "SmsMessageQueue";
    17.     // RoutingKey
    18.     public final static String SMS_ROUTING_KEY_NAME = "message.sms";
    20.     @Autowired
    21.     public void init() {
    22.         Exchange exchange = new DirectExchange(SMS_EXCHANGE_NAME);
    23.         amqpAdmin.declareExchange(exchange);
    24.         amqpAdmin.declareQueue(new Queue(SMS_QUEUE_NAME, true));
    25.         //创建绑定规则
    26.         amqpAdmin.declareBinding(new Binding(SMS_QUEUE_NAME, Binding.DestinationType.QUEUE, SMS_EXCHANGE_NAME, SMS_ROUTING_KEY_NAME, getArguments()));
    27.     }
    29.     public Map<String, Object> getArguments() {
    30.         Map<String, Object> arguments = new HashMap<>(10);
    31.         // 绑定死信交换机 & 设置超时时间
    32.         arguments.put("x-dead-letter-exchange", rabbitMQProperties.getDefaultMessage().getExchange().getName());
    33.         arguments.put("x-dead-letter-routing-key", rabbitMQProperties.getDefaultMessage().getBinding().getRoutingKey());
    34.         arguments.put("x-message-ttl", 10000);
    35.         arguments.put("x-queue-mode", "lazy");
    36.         return arguments;
    37.     }
    38. }