springboot 整合 RabbitMQ

背景

多线程读取大文件后需要分时段进行第三方接口的调用【由于外部接口限制API30s才可以调用一次】，所以我们将使用RabbitMQ的延时消费的方式进行外部接口的调用，以保证项目程序的正常运行

RabbitMQ介绍

RabbitMQ介绍
RabbitMQ 快速下载安装

详情看上 ，本篇文章只解决背景问题，整合springboot+RabbitMQ

springboot 整合 RabbitMQ[延时队列]

延时队列

延时队列是存储延时消息的队列，延时消息就是生产者发送了一条消息，但是不希望该消息不要被立即消费，而是设置一个延时时间，等过了这个时间再消费消息

延时队列安装插件

在RabbitMQ3.6版本以前一般采用死信队列 + TTL过期时间的方式来实现延迟队列。
从3.6以后的版本，RabbitMQ官方提供了延迟对列的插件官方下载地址rabbitmq_delayed_message_exchange-3.8.0.ez 这个插件放到 RabbitMQ 安装目录的 plugins文件中

RabbitMQ 安装 sbin 文件中用 cmd 执行命令:

  1. rabbitmq-plugins enable rabbitmq_delayed_message_exchange

然后登陆web页面，可以查看到交换机tab下新增了x-delayed-message

插件原理

springboot 整合 RabbitMQ - 图1

  • 原始的 死信队列+ TTL 的模式，消息首先会路由到一个正常的队列，根据设置的 TTL 进入死信队列，与之不同的是通过 x-delayed-message 声明的交换机，它的消息在发布之后不会立即进入队列，先将消息保存至 Mnesia（一个分布式数据库管理系统，适合于电信和其它需要持续运行和具备软实时特性的 Erlang 应用）
  • 这个插件将会尝试确认消息是否过期，首先要确保消息的延迟范围是 Delay > 0, Delay =< ?ERL_MAX_T（在 Erlang 中可以被设置的范围为 (2^32)-1 毫秒），如果消息过期通过 x-delayed-type 类型标记的交换机投递至目标队列，整个消息的投递过程也就完成了

代码具体实现

pom.xml

  1. <!--springboot  2.3.1.RELEASE-->
  2. <dependency>
  3.      <groupId>org.springframework.boot</groupId>
  4.      <artifactId>spring-boot-starter-amqp</artifactId>
  5.  </dependency>
  6.  <dependency>
  7.      <groupId>org.springframework.boot</groupId>
  8.      <artifactId>spring-boot-starter-web</artifactId>
  9.  </dependency>

application.properties

  2. # 由于MQ只在本项目中，未跨项目 所有配置文件一份，跨项目则需要分开配置
  3. spring.rabbitmq.host=127.0.0.1
  4. spring.rabbitmq.port=5672
  5. spring.rabbitmq.username=guest
  6. spring.rabbitmq.password=guest
  7. #开启消息的return机制
  8. spring.rabbitmq.publisher-confirms=true
  9. #确认消息已发送到队列
  10. spring.rabbitmq.publisher-returns=true
  11. #开启消息的confirm机制 correlated:开启;NONE:关闭
  12. spring.rabbitmq.publisher-confirms-type=correlated
  13. #在需要使用消息的return机制时候，此参数必须设置为true
  14. spring.rabbitmq.template.mandatory=true
  16. spring.rabbitmq.listener.type=simple
  17. #消费方消息确认：手动确认
  18. spring.rabbitmq.listener.simple.acknowledge-mode=manual
  19. #决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系）
  20. spring.rabbitmq.listener.simple.default-requeue-rejected=false

静态常量抽取

  1. /**
  2.  *  MQ相关常量
  3.  **/
  4. public class MqConstant {
  6.     // 延时交换机
  7.     public static final String MQ_WEBSITE_FILM_DELAY_EXCHANGE = "website_film_delay_exchange";
  9.     // 延时队列名称
  10.     public static final String MQ_WEBSITE_FILM_QUEUE = "website_film_delay_queue";
  12.     // 普通交换机
  13.     public static final String MQ_WEBSITE_NORMAL_EXCHANGE = "website_normal_exchange";
  15.     // 普通队列名称
  16.     public static final String MQ_WEBSITE_NORMAL_QUEUE = "website_normal_queue";
  18.     // 普通交换机路由键
  19.     public static final String MQ_WEBSITE_NORMAL_ROUTING_KEY = "website_normal_routing_key";
  21. }

配置类

RabbitConfig

主要是队列，交换机的配置绑定

  1. import org.springframework.amqp.core.*;
  2. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
  3. import org.springframework.amqp.rabbit.core.RabbitAdmin;
  4. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  5. import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
  6. import org.springframework.context.annotation.Bean;
  7. import org.springframework.context.annotation.Configuration;
  9. import java.util.HashMap;
  10. import java.util.Map;
  12. /**
  13.  * RabbitMQ的配置 主要是队列，交换机的配置绑定
  14.  **/
  15. @Configuration
  16. public class RabbitConfig {
  18.     //https://blog.csdn.net/weixin_38192427/article/details/120509586
  19.     /**
  20.      * durable="true" 持久化 rabbitmq重启的时候不需要创建新的队列
  21.      * exclusive 表示该消息队列是否只在当前connection生效,默认是false
  22.      * auto-delete 表示消息队列没有在使用时将被自动删除 默认是false
  23.      */
  24.     // 延时队列
  25.     @Bean
  26.     public Queue delayQueue() {
  27.         return new Queue(MqConstant.MQ_WEBSITE_FILM_QUEUE, true,false,false);
  28.     }
  30.     /**
  31.      * 交换机说明:
  32.      * durable="true" rabbitmq重启的时候不需要创建新的交换机
  33.      * auto-delete 表示交换机没有在使用时将被自动删除 默认是false
  34.      * direct交换器相对来说比较简单，匹配规则为：如果路由键匹配，消息就被投送到相关的队列
  35.      * topic交换器你采用模糊匹配路由键的原则进行转发消息到队列中
  36.      * fanout交换器中没有路由键的概念，他会把消息发送到所有绑定在此交换器上面的队列中。
  37.      */
  38.     // 延时交换机 设置
  39.     public FanoutExchange delayExchange() {
  40.         //使用自定义交换器
  41.         Map<String, Object> args = new HashMap<>();
  42.         args.put("x-delayed-type", "direct");
  43.         FanoutExchange fanoutExchange = new FanoutExchange(MqConstant.MQ_WEBSITE_FILM_DELAY_EXCHANGE, true, false, args);
  44.         fanoutExchange.setDelayed(true);
  45.         return fanoutExchange;
  46.     }
  48.     // 绑定延时队列与延时交换机
  49.     @Bean
  50.     public Binding delayBind() {
  51.         return BindingBuilder.bind(delayQueue()).to(delayExchange());
  52.     }
  54.     // ------------------------普通队列------------------------
  55.     // 普通队列
  56.     @Bean
  57.     public Queue normalQueue() {
  58.         return new Queue(MqConstant.MQ_WEBSITE_NORMAL_QUEUE, true);
  59.     }
  61.     // 普通交换机
  62.     @Bean
  63.     public DirectExchange normalExchange() {
  64.         return new DirectExchange(MqConstant.MQ_WEBSITE_NORMAL_EXCHANGE, true, false);
  65.     }
  67.     // 绑定普通消息队列
  68.     @Bean
  69.     public Binding normalBind() {
  70.         return BindingBuilder.bind(normalQueue()).to(normalExchange()).with(MqConstant.MQ_WEBSITE_NORMAL_ROUTING_KEY);
  71.     }
  73.     // 定义消息转换器
  74.     @Bean
  75.     public Jackson2JsonMessageConverter jsonMessageConverter() {
  76.         return new Jackson2JsonMessageConverter();
  77.     }
  80.     /** ======================== 定制一些处理策略 =============================*/
  82.     /**
  83.      * 定制化amqp模版
  84.      * <p>
  85.      * Rabbit MQ的消息确认有两种。
  86.      * <p>
  87.      * 一种是消息发送确认：这种是用来确认生产者将消息发送给交换机，交换机传递给队列过程中，消息是否成功投递。
  88.      * 发送确认分两步：一是确认是否到达交换机，二是确认是否到达队列
  89.      * <p>
  90.      * 第二种是消费接收确认：这种是确认消费者是否成功消费了队列中的消息。
  91.      */
  92.     // 定义消息模板用于发布消息，并且设置其消息转换器
  93.     @Bean
  94.     public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
  95.         final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
  96.         // 消息发送失败返回到队列中, yml需要配置 publisher-returns: true
  97.         rabbitTemplate.setMandatory(true);
  99.         /**
  100.          * 使用该功能需要开启消息确认，yml需要配置 publisher-confirms: true
  101.          * 通过实现ConfirmCallBack接口，用于实现消息发送到交换机Exchange后接收ack回调
  102.          * correlationData  消息唯一标志
  103.          * ack              确认结果
  104.          * cause            失败原因
  105.          */
  106.         rabbitTemplate.setConfirmCallback(new MsgSendConfirmCallback());
  107.         /**
  108.          * 使用该功能需要开启消息返回确认，yml需要配置 publisher-returns: true
  109.          * 通过实现ReturnCallback接口，如果消息从交换机发送到对应队列失败时触发
  110.          * message    消息主体 message
  111.          * replyCode  消息主体 message
  112.          * replyText  描述
  113.          * exchange   消息使用的交换机
  114.          * routingKey 消息使用的路由键
  115.          */
  116.         rabbitTemplate.setReturnCallback(new MsgSendReturnCallback());
  117.         rabbitTemplate.setMessageConverter(jsonMessageConverter());
  118.         return rabbitTemplate;
  119.     }
  121.     // --------------------------使用RabbitAdmin启动服务便创建交换机和队列--------------------------
  122.     @Bean
  123.     public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
  124.         RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
  125.         // 只有设置为 true，spring 才会加载 RabbitAdmin 这个类
  126.         rabbitAdmin.setAutoStartup(true);
  127.         // 创建延时交换机和对列
  128.         rabbitAdmin.declareExchange(delayExchange());
  129.         rabbitAdmin.declareQueue(delayQueue());
  130.         // 创建普通交换机和对列
  131.         rabbitAdmin.declareExchange(normalExchange());
  132.         rabbitAdmin.declareQueue(normalQueue());
  133.         return new RabbitAdmin(connectionFactory);
  134.     }
  136. }

RabbitConfirmConfig

RabbitMQ的配置 消息发送到 exchange，queue 的回调函数

  1. import org.apache.logging.log4j.LogManager;
  2. import org.apache.logging.log4j.Logger;
  3. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
  4. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  9. /**
  10.  * RabbitMQ的配置 消息发送到 exchange，queue 的回调函数
  11.  **/
  12. @Configuration
  13. public class RabbitConfirmConfig {
  15.     private static Logger logger = LogManager.getLogger(RabbitConfirmConfig.class);
  17.     /** ======================== 定制一些处理策略 =============================*/
  18.     /**
  19.      * 定制化amqp模版
  20.      * <p>
  21.      * Rabbit MQ的消息确认有两种。
  22.      * <p>
  23.      * 一种是消息发送确认：这种是用来确认生产者将消息发送给交换机，交换机传递给队列过程中，消息是否成功投递。
  24.      * 发送确认分两步：一是确认是否到达交换机，二是确认是否到达队列
  25.      * <p>
  26.      * 第二种是消费接收确认：这种是确认消费者是否成功消费了队列中的消息。
  27.      * Springboot中使用ConfirmCallback和ReturnCallback
  28.      * 注意：
  29.      *  在需要使用消息的return机制时候，mandatory参数必须设置为true
  30.      * 新版本开启消息的confirm配置publisher-confirms已经过时，改为使用publisher-confirm-type参数设置（correlated:开启;NONE:关闭）
  31.      */
  32.     @Bean
  33.     public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
  34.         RabbitTemplate rabbitTemplate = new RabbitTemplate();
  35.         rabbitTemplate.setConnectionFactory(connectionFactory);
  36. //        设置开启Mandatory才能触发回调函数，无论消息推送结果怎么样都强制调用回调函数
  37.         rabbitTemplate.setMandatory(true);
  38.         /**
  39.          * 使用该功能需要开启消息确认 无论成功与否，需要配置 publisher-confirms: true
  40.          * 通过实现ConfirmCallBack接口，用于实现消息发送到交换机Exchange后接收ack回调
  41.          * correlationData  消息唯一标志
  42.          * ack              确认结果
  43.          * cause            失败原因
  44.          */
  45.         rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
  46.             logger.info("ConfirmCallback：" + "相关数据：" + correlationData);
  47.             logger.info("ConfirmCallback：" + "确认情况：" + ack);
  48.             logger.info("ConfirmCallback：" + "原因：" + cause);
  49.         });
  50.         /**
  51.          * 消息从Exchange路由到Queue失败的回调
  52.          * 使用该功能需要开启消息返回确认，需要配置 publisher-returns: true
  53.          * 通过实现ReturnCallback接口，如果消息从交换机发送到对应队列失败时触发
  54.          * message    消息主体 message
  55.          * replyCode  消息主体 message
  56.          * replyText  描述
  57.          * exchange   消息使用的交换机
  58.          * routingKey 消息使用的路由键
  59.          */
  60.         rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
  61.             logger.info("ReturnCallback：" + "消息：" + message);
  62.             logger.info("ReturnCallback：" + "回应码：" + replyCode);
  63.             logger.info("ReturnCallback：" + "回应信息：" + replyText);
  64.             logger.info("ReturnCallback：" + "交换机：" + exchange);
  65.             logger.info("ReturnCallback：" + "路由键：" + routingKey);
  66.         });
  67.         return rabbitTemplate;
  68.     }
  69. }

MsgSendReturnCallback

消息从交换机发送到对应队列失败时触发

  1. import org.apache.logging.log4j.LogManager;
  2. import org.apache.logging.log4j.Logger;
  3. import org.springframework.amqp.core.Message;
  4. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  5. import org.springframework.stereotype.Component;
  7. @Component
  8. public class MsgSendReturnCallback implements RabbitTemplate.ReturnCallback {
  9.     private static Logger logger = LogManager.getLogger(MsgSendReturnCallback.class);
  11.     /**
  12.      * 使用该功能需要开启消息返回确认，yml需要配置 publisher-returns: true
  13.      * message    消息主体 message
  14.      * replyCode  消息主体 message
  15.      * replyText  描述
  16.      * exchange   消息使用的交换机
  17.      * routingKey 消息使用的路由键
  18.      * <p>
  19.      * PS：通过实现ReturnCallback接口，如果消息从交换机发送到对应队列失败时触发
  20.      * </p>
  21.      */
  22.     @Override
  23.     public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
  24.         if (exchange.equals(MqConstant.MQ_WEBSITE_FILM_DELAY_EXCHANGE)) {
  25.             // 如果配置了发送回调ReturnCallback，rabbitmq_delayed_message_exchange插件会回调该方法，因为发送方确实没有投递到队列上，只是在交换器上暂存，等过期/时间到了才会发往队列。
  26.             // 所以如果是延迟队列的交换器，则直接放过，并不是bug
  27.             return;
  28.         }
  29.         String correlationId = message.getMessageProperties().getCorrelationId();
  30.         logger.debug("消息：{} 发送失败, 应答码：{} 原因：{} 交换机: {}  路由键: {}", correlationId, replyCode, replyText, exchange, routingKey);
  31.     }
  32. }

AckListener

Consumer ACK机制

  1. import com.rabbitmq.client.Channel;
  2. import org.apache.logging.log4j.LogManager;
  3. import org.apache.logging.log4j.Logger;
  4. import org.springframework.amqp.core.Message;
  5. import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
  6. import org.springframework.stereotype.Component;
  8. import java.util.Map;
  10. /**
  11. * Consumer ACK机制：
  12. *  1. 设置手动签收。acknowledge="manual"
  13. *  2. 让监听器类实现ChannelAwareMessageListener接口
  14. *  3. 如果消息成功处理，则调用channel的 basicAck()签收
  15. *  4. 如果消息处理失败，则调用channel的basicNack()拒绝签收，broker重新发送给consumer
  16. *    
  17. */
  19. @Component
  20. public class AckListener implements ChannelAwareMessageListener {
  22.     private static Logger logger = LogManager.getLogger(AckListener.class);
  24.     Map<Long,Integer> map;
  25.     @Override
  26.     public void onMessage(Message message, Channel channel) throws Exception {
  27.         long deliveryTag = message.getMessageProperties().getDeliveryTag();
  29.         try {
  30.             //1.接收转换消息
  31.             logger.info("MQ接收转换消息[{}]",new String(message.getBody()));
  33.             //2. 处理业务逻辑
  34.             logger.info("处理业务逻辑...");
  35.             int i = 3/0;//出现错误
  36.             //3. 手动签收
  37.             channel.basicAck(deliveryTag,false);
  38.         } catch (Exception e) {
  39.             //e.printStackTrace();
  40.             logger.error("MQ拒绝签收逻辑...");
  41.             /*
  42.             4.拒绝签收
  43.             第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会重新发送该消息给消费端
  44.             */
  45.             /*if(!map.containsKey(deliveryTag)){
  46.             map.put(deliveryTag,1);
  47.             channel.basicNack(deliveryTag,true,true);
  48.             }else{
  49.             //判断是否大于3次
  50.             if(>3){
  51.             channel.basicNack(deliveryTag,false,true);
  52.             //记录
  53.             }else{
  54.             channel.basicNack(deliveryTag,true,true);
  55.             }
  57.             }*/
  58.             channel.basicNack(deliveryTag,false,false);
  59.             //channel.basicReject(deliveryTag,true);
  60.         }
  61.     }
  62. }

MsgProductionService

发送消息到MQ 业务类 消费生产

  1. import org.springframework.amqp.core.MessageDeliveryMode;
  2. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  3. import org.springframework.stereotype.Service;
  5. import javax.annotation.Resource;
  7. /**
  8.  * 发送消息到MQ
  9.  **/
  10. @Service("msgProductionService")
  11. public class MsgProductionService {
  13.     @Resource
  14.     private RabbitTemplate rabbitTemplate;
  16.     // 发送延时信息
  17.     public void sendTimeoutMsg(String exchange,String content, String routingKey, int delay) {
  18.         // 通过广播模式发布延时消息，会广播至每个绑定此交换机的队列，这里的路由键没有实质性作用
  19.         rabbitTemplate.convertAndSend(exchange, routingKey, content, message -> {
  20.             message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
  21.             // 毫秒为单位，指定此消息的延时时长
  22.             message.getMessageProperties().setDelay(delay * 1000);
  23.             return message;
  24.         });
  25.     }
  26.     // 发送普通消息
  27.     public void sendMsg(String exchange,String routingKey, String content) {
  28.         // DirectExchange类型的交换机，必须指定对应的路由键
  29.         rabbitTemplate.convertAndSend(exchange, routingKey, content);
  30.     }
  31. }

MsgComsumerService

MQ消费 消费者【小编这里是在同一个项目中实现的，生产和消费】

  1. import com.rabbitmq.client.Channel;
  2. import com.zy.website.service.FilmService;
  3. import org.apache.logging.log4j.LogManager;
  4. import org.apache.logging.log4j.Logger;
  5. import org.springframework.amqp.core.Message;
  6. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  7. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  8. import org.springframework.stereotype.Service;
  10. import javax.annotation.Resource;
  11. import java.io.IOException;
  13. /**
  14.  *  MQ 消费者
  15.  **/
  16. @Service("msgComsumerService")
  17. public class MsgComsumerService {
  19.     private static Logger logger = LogManager.getLogger(MsgComsumerService.class);
  21.     @Resource
  22.     FilmService filmService;//自己业务具体实现类注入
  24.     // 监听消费延时消息
  25.     @RabbitListener(queues = {"website_film_delay_queue"})
  26.     @RabbitHandler
  27.     public void process(String content, Message message, Channel channel) throws IOException {
  28.         try {
  29.             // 消息的可定确认，第二个参数如果为true将一次性确认所有小于deliveryTag的消息
  30.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
  31.             //调用方法消费消息 自己业务具体实现类
  32.             filmService.getFilmInfoByExternalApi(content);
  33.             logger.info("延迟队列消息[{}]被消费！！",content);
  34.         } catch (Exception e) {
  35.             logger.error("延迟队列消息 处理失败:{}", e.getMessage());
  36.             // 直接拒绝消费该消息，后面的参数一定要是false，否则会重新进入业务队列，不会进入死信队列
  37.             channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
  38.             /*// ack返回false，requeue-true并重新回到队列
  39.             channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);*/
  40.         }
  41.     }
  43.     // 消费普通消息
  44.     @RabbitListener(queues = {"website_normal_queue"})
  45.     @RabbitHandler
  46.     public void process1(String content, Message message, Channel channel) throws IOException {
  47.         try {
  48.             logger.info("普通队列的内容[{}]", content);
  49.             // 消息的可定确认，第二个参数如果为true将一次性确认所有小于deliveryTag的消息
  50.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
  51.             logger.info("普通信息处理完毕");
  52.         } catch (Exception e) {
  53.             logger.error("处理失败:{}", e.getMessage());
  54.             // 直接拒绝消费该消息，后面的参数一定要是false，否则会重新进入业务队列，不会进入死信队列
  55.             channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
  56.         }
  57.     }
  58. }
  1. **代码实现完成，这只是简单实现，复杂还需要研究。。。。。**

RabbitMQ的具体详细配置项

基础配置项

  1. spring.rabbitmq.host: 默认localhost
  2. spring.rabbitmq.port: 默认5672
  3. spring.rabbitmq.username: 用户名
  4. spring.rabbitmq.password: 密码
  5. spring.rabbitmq.virtual-host: 连接到代理时用的虚拟主机
  6. spring.rabbitmq.addresses: 连接到server的地址列表（以逗号分隔），先addresses后host 
  7. spring.rabbitmq.requested-heartbeat: 请求心跳超时时间，0为不指定，如果不指定时间单位默认为妙
  8. spring.rabbitmq.publisher-confirms: 是否启用【发布确认】，默认false
  9. spring.rabbitmq.publisher-returns: 是否启用【发布返回】，默认false
  10. spring.rabbitmq.connection-timeout: 连接超时时间，单位毫秒，0表示永不超时

SSL

  1. spring.rabbitmq.ssl.enabled: 是否支持ssl，默认false
  2. spring.rabbitmq.ssl.key-store: 持有SSL certificate的key store的路径
  3. spring.rabbitmq.ssl.key-store-password: 访问key store的密码
  4. spring.rabbitmq.ssl.trust-store: 持有SSL certificates的Trust store
  5. spring.rabbitmq.ssl.trust-store-password: 访问trust store的密码
  6. spring.rabbitmq.ssl.trust-store-type=JKS：Trust store 类型.
  7. spring.rabbitmq.ssl.algorithm: ssl使用的算法，默认由rabiitClient配置
  8. spring.rabbitmq.ssl.validate-server-certificate=true：是否启用服务端证书验证
  9. spring.rabbitmq.ssl.verify-hostname=true 是否启用主机验证

缓存cache

  1. spring.rabbitmq.cache.channel.size: 缓存中保持的channel数量
  2. spring.rabbitmq.cache.channel.checkout-timeout: 当缓存数量被设置时，从缓存中获取一个channel的超时时间，单位毫秒；如果为0，则总是创建一个新channel
  3. spring.rabbitmq.cache.connection.size: 缓存的channel数，只有是CONNECTION模式时生效
  4. spring.rabbitmq.cache.connection.mode=channel: 连接工厂缓存模式：channel 和 connection

Listener

  1. spring.rabbitmq.listener.type=simple: 容器类型.simple或direct
  3. spring.rabbitmq.listener.simple.auto-startup=true: 是否启动时自动启动容器
  4. spring.rabbitmq.listener.simple.acknowledge-mode: 表示消息确认方式，其有三种配置方式，分别是none、manual和auto；默认auto
  5. spring.rabbitmq.listener.simple.concurrency: 最小的消费者数量
  6. spring.rabbitmq.listener.simple.max-concurrency: 最大的消费者数量
  7. spring.rabbitmq.listener.simple.prefetch: 一个消费者最多可处理的nack消息数量，如果有事务的话，必须大于等于transaction数量.
  8. spring.rabbitmq.listener.simple.transaction-size: 当ack模式为auto时，一个事务（ack间）处理的消息数量，最好是小于等于prefetch的数量.若大于prefetch， 则prefetch将增加到这个值
  9. spring.rabbitmq.listener.simple.default-requeue-rejected: 决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系）
  10. spring.rabbitmq.listener.simple.missing-queues-fatal=true 若容器声明的队列在代理上不可用，是否失败； 或者运行时一个多多个队列被删除，是否停止容器
  11. spring.rabbitmq.listener.simple.idle-event-interval: 发布空闲容器的时间间隔，单位毫秒
  12. spring.rabbitmq.listener.simple.retry.enabled=false: 监听重试是否可用
  13. spring.rabbitmq.listener.simple.retry.max-attempts=3: 最大重试次数
  14. spring.rabbitmq.listener.simple.retry.max-interval=10000ms: 最大重试时间间隔
  15. spring.rabbitmq.listener.simple.retry.initial-interval=1000ms:第一次和第二次尝试传递消息的时间间隔
  16. spring.rabbitmq.listener.simple.retry.multiplier=1: 应用于上一重试间隔的乘数
  17. spring.rabbitmq.listener.simple.retry.stateless=true: 重试时有状态or无状态
  19. spring.rabbitmq.listener.direct.acknowledge-mode= ack模式
  20. spring.rabbitmq.listener.direct.auto-startup=true 是否在启动时自动启动容器
  21. spring.rabbitmq.listener.direct.consumers-per-queue= 每个队列消费者数量.
  22. spring.rabbitmq.listener.direct.default-requeue-rejected= 默认是否将拒绝传送的消息重新入队.
  23. spring.rabbitmq.listener.direct.idle-event-interval= 空闲容器事件发布时间间隔.
  24. spring.rabbitmq.listener.direct.missing-queues-fatal=false若容器声明的队列在代理上不可用，是否失败.
  25. spring.rabbitmq.listener.direct.prefetch= 每个消费者可最大处理的nack消息数量.
  26. spring.rabbitmq.listener.direct.retry.enabled=false  是否启用发布重试机制.
  27. spring.rabbitmq.listener.direct.retry.initial-interval=1000ms # Duration between the first and second attempt to deliver a message.
  28. spring.rabbitmq.listener.direct.retry.max-attempts=3 # Maximum number of attempts to deliver a message.
  29. spring.rabbitmq.listener.direct.retry.max-interval=10000ms # Maximum duration between attempts.
  30. spring.rabbitmq.listener.direct.retry.multiplier=1 # Multiplier to apply to the previous retry interval.
  31. spring.rabbitmq.listener.direct.retry.stateless=true # Whether retries are stateless or stateful.

Template

  1. spring.rabbitmq.template.mandatory: 启用强制信息；默认false
  2. spring.rabbitmq.template.receive-timeout: receive() 操作的超时时间
  3. spring.rabbitmq.template.reply-timeout: sendAndReceive() 操作的超时时间
  4. spring.rabbitmq.template.retry.enabled=false: 发送重试是否可用 
  5. spring.rabbitmq.template.retry.max-attempts=3: 最大重试次数
  6. spring.rabbitmq.template.retry.initial-interva=1000msl: 第一次和第二次尝试发布或传递消息之间的间隔
  7. spring.rabbitmq.template.retry.multiplier=1: 应用于上一重试间隔的乘数
  8. spring.rabbitmq.template.retry.max-interval=10000: 最大重试时间间隔