RabbitMQ——SpringBoot配置选项

Bertha 。 2023-09-29 13:05 13阅读 0赞

##### 基本选项 #####

`spring.rabbitmq.host`: 服务Host  
`spring.rabbitmq.port`: 服务端口  
`spring.rabbitmq.username`: 用户名  
`spring.rabbitmq.password`: 密码  
`spring.rabbitmq.virtual-host`: RabbitMQ的虚拟主机  
`spring.rabbitmq.addresses`: 指定client连接到的server的地址，多个以逗号分隔(优先取addresses，然后再取host)  
`spring.rabbitmq.requested-heartbeat`: 指定心跳超时，单位秒，0为不指定；默认60s  
`spring.rabbitmq.publisher-confirms`: 开启Publisher Confirm机制  
`spring.rabbitmq.publisher-returns`: 开启publisher Return机制  
`spring.rabbitmq.connection-timeout`: 连接超时，单位毫秒，0表示无穷大，不超时

##### SSL #####

`spring.rabbitmq.ssl.enabled`: 是否支持ssl  
`spring.rabbitmq.ssl.key-store`: 指定持有SSL certificate的key store的路径  
`spring.rabbitmq.ssl.key-store-password`: 指定访问key store的密码  
`spring.rabbitmq.ssl.trust-store`: 指定持有SSL certificates的Trust store  
`spring.rabbitmq.ssl.trust-store-password`: 指定访问trust store的密码  
`spring.rabbitmq.ssl.algorithm`: ssl使用的算法，例如，TLSv1.1

##### cache #####

`spring.rabbitmq.cache.channel.size`: 缓存中保持的channel数量

`spring.rabbitmq.cache.channel.checkout-timeout`: 当缓存数量被设置时，从缓存中获取一个channel的超时时间，单位毫秒；如果为0，则总是创建一个新channel

channel不是线程安全的，线程并发的去访问同一个channel会出问题。这里有几种处理方式：
    
    1 全局公用一个channel且使用全局锁,让操作channel排队.这种明显性能是不行的；
    2 一个线程对应创建一个新的channel，但是要处理好一个连接能支撑的最大channel数量；
    3 一个线程对应一个channel,但是是从channel池子拿的,不是每次都创建新的.一旦一个线程完成了一个channel的使用，它将返回到池中，从而使该channel可用于另一个线程。
    
    量不大的话,使用第二种方式就可以了。量大的话，建议使用第三种方式，毕竟创建和销毁channel也是耗时耗资源的.在spring amqp中，提供了一个缓存channel的方案。可以在创建CachingConnectionFactory时指定缓存的模式。
    
    connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
    connectionFactory.setChannelCacheSize(25);
    
    上面的两行代码,表示channel共用唯一的一个连接,且缓存了25个channel,注意这里的25个并不是说，这个连接里只能最多创建25个channel,而是说最多缓存25个channel。举个例子，假设并发发送100条消息，在CachingConnectionFactory.CacheMode.CHANNEL模式下，瞬间会创建100个channel的，然后往缓存里放25个channel，当流量下去了，刚刚创建的多余的channel会自动关闭掉的，缓存里只保留25个。
    
    使用这种方式的话，要注意缓存的channel数量，不能太小，不然流量一大,仍然会造成频繁关闭channel的情况。当然我们也不能说有多少并发，就创建多少个channel，还是要限制一下，这个时候可以使用：
    connectionFactory.setChannelCheckoutTimeout(1000);
    当ChannelCheckoutTimeout的值大于0的时候，ChannelCacheSize的值就是最大的channel数量了，一旦从缓存中获取不到channel，等待ChannelCheckoutTimeout毫秒后,如果还是获取不到的，就会抛AmqpTimeoutException了。
    链接：https://juejin.cn/post/6844904015508029453

`spring.rabbitmq.cache.connection.size`: 缓存的连接数，只有是CONNECTION模式时生效  
`spring.rabbitmq.cache.connection.mode`: 连接工厂缓存模式：CHANNEL 和 CONNECTION

##### listener #####

`spring.rabbitmq.listener.simple.auto-startup`: 是否启动时自动启动容器

使用情景1:
    rabbitMQ配置动态启动，rabbitMQ代理不正常时可以不启动项目中MQ的监听，主要解决，项目和MQ的启动顺序的问题。
    使用情景2:
    当启动程序时，并且MQ队列中有消息，这些消息的消费就会在应用程序完全启动之前开始，但是启动程序要初始化一些程序要用的数据，在没有初始化完成时消费这些消息会报错，所以希望在程序初始化完成后，再启动RabbitMQ的监听容器。
    
     可以通过控制层，测试去启动，生成环境是通过一个线程，一个去查看rabbit的代理服务器是否正常，正常的时候就开启rabbit的监听。
    package com.example.demo.main;
     
    import java.util.Set;
    import javax.annotation.Resource;
    import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
     
     
    @RestController
    @RequestMapping("rabbitmq/listener")
    public class RabbitMQController {
        
     
        @Resource
        private RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry;
     
        @RequestMapping("stop")
        public String stop(){
        
            rabbitListenerEndpointRegistry.stop();
            System.out.println("停止MQ监听");
            return "success";
        }
     
        @RequestMapping("start")
        public String start(){
        
            rabbitListenerEndpointRegistry.start();
            System.out.println("启动MQ监听");
            return "success";
        }
     
        @RequestMapping("setup")
        public String setup(int consumer, int maxConsumer){
        
            Set<String> containerIds = rabbitListenerEndpointRegistry.getListenerContainerIds();
            SimpleMessageListenerContainer container = null;
            for(String id : containerIds){
        
                container = (SimpleMessageListenerContainer) rabbitListenerEndpointRegistry.getListenerContainer(id);
                if(container != null){
        
                    container.setConcurrentConsumers(consumer);
                    container.setMaxConcurrentConsumers(maxConsumer);
                }
            }
            System.out.println("设置");
            return "success";
        }
    }

`spring.rabbitmq.listener.simple.acknowledge-mode`: 表示消息确认方式，其有三种配置方式，分别是none、manual和auto；默认auto  
`spring.rabbitmq.listener.simple.concurrency`: 最小的消费者数量

sprinboot2.0版本之后，可以通过注解设置该选项：  
![在这里插入图片描述][eca4c8c1fa9042d599fbb2eccb441cac.png]

`spring.rabbitmq.listener.simple.max-concurrency`: 最大的消费者数量

`spring.rabbitmq.listener.simple.prefetch`: 指定一个请求能处理多少个消息，如果有事务的话，必须大于等于transaction数量.

每个customer会在MQ预取一些消息放入内存的LinkedBlockingQueue中，这个值越高，消息传递的越快，但非顺序处理消息的风险更高。如果ack模式为none，则忽略。如有必要，将增加此值以匹配txSize或messagePerAck。从2.0开始默认为250；设置为1将还原为以前的行为。
    
    prefetch默认值以前是1，这可能会导致高效使用者的利用率不足。从spring-amqp 2.0版开始，默认的prefetch值是250，这将使消费者在大多数常见场景中保持忙碌，从而提高吞吐量。
    
    不过在有些情况下，尤其是处理速度比较慢的大消息，消息可能在内存中大量堆积，消耗大量内存；以及对于一些严格要求顺序的消息，prefetch的值应当设置为1。
    
    对于低容量消息和多个消费者的情况（也包括单listener容器的concurrency配置）希望在多个使用者之间实现更均匀的消息分布，建议在手动ack下并设置prefetch=1。

`spring.rabbitmq.listener.simple.transaction-size`: 指定一个事务处理的消息数量，最好是小于等于prefetch的数量.  
`spring.rabbitmq.listener.simple.default-requeue-rejected`: 决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系）  
`spring.rabbitmq.listener.simple.idle-event-interval`: 多少长时间发布空闲容器时间，单位毫秒

`spring.rabbitmq.listener.simple.retry.enabled`: 监听重试是否可用  
`spring.rabbitmq.listener.simple.retry.max-attempts`: 最大重试次数  
`spring.rabbitmq.listener.simple.retry.initial-interval`: 第一次和第二次尝试发布或传递消息之间的间隔  
`spring.rabbitmq.listener.simple.retry.multiplier`: 应用于上一重试间隔的乘数  
`spring.rabbitmq.listener.simple.retry.max-interval`: 最大重试时间间隔  
`spring.rabbitmq.listener.simple.retry.stateless`: 重试是有状态or无状态

##### template #####

`spring.rabbitmq.template.mandatory`: 启用强制信息；默认false  
`spring.rabbitmq.template.receive-timeout`: receive() 操作的超时时间  
`spring.rabbitmq.template.reply-timeout`: sendAndReceive() 操作的超时时间  
`spring.rabbitmq.template.retry.enabled`: 发送重试是否可用  
`spring.rabbitmq.template.retry.max-attempts`: 最大重试次数  
`spring.rabbitmq.template.retry.initial-interva`l: 第一次和第二次尝试发布或传递消息之间的间隔  
`spring.rabbitmq.template.retry.multiplie`r: 应用于上一重试间隔的乘数  
`spring.rabbitmq.template.retry.max-interval`: 最大重试时间间隔

##### 全部参数 #####

spring.rabbitmq.addresses= # 以逗号分隔的客户端应连接的地址列表
    spring.rabbitmq.cache.channel.checkout-timeout= # 如果已达到缓存大小，则等待获取通道的持续时间
    spring.rabbitmq.cache.channel.size= # 要在缓存中保留的通道数
    spring.rabbitmq.cache.connection.mode=channel # 连接工厂缓存模式
    spring.rabbitmq.cache.connection.size= # 缓存的连接数
    spring.rabbitmq.connection-timeout= # 连接超时。将其设置为 0 以永远等待
    spring.rabbitmq.dynamic=true # 是否创建 AmqpAdmin bean
    spring.rabbitmq.host=localhost # RabbitMQ 主机
    spring.rabbitmq.listener.direct.acknowledge-mode= # 确认容器的模式
    spring.rabbitmq.listener.direct.auto-startup=true # 是否在启动时自动启动容器
    spring.rabbitmq.listener.direct.consumers-per-queue= # 每个队列的消费者数量
    spring.rabbitmq.listener.direct.default-requeue-rejected= # 默认情况下，拒绝交付是否重新排队
    spring.rabbitmq.listener.direct.idle-event-interval= # 应该多久发布一次空闲容器事件
    spring.rabbitmq.listener.direct.missing-queues-fatal=false # 如果容器声明的队列在代理上不可用，则是否失败
    spring.rabbitmq.listener.direct.prefetch= # 每个消费者可能未完成的最大未确认消息数
    spring.rabbitmq.listener.direct.retry.enabled=false # 是否启用发布重试
    spring.rabbitmq.listener.direct.retry.initial-interval=1000ms # 第一次和第二次尝试传递消息之间的持续时间
    spring.rabbitmq.listener.direct.retry.max-attempts=3 # 传递消息的最大尝试次数
    spring.rabbitmq.listener.direct.retry.max-interval=10000ms # 最长尝试次数
    spring.rabbitmq.listener.direct.retry.multiplier=1 # 乘数应用于先前的重试间隔
    spring.rabbitmq.listener.direct.retry.stateless=true # 重试是无国籍还是有状态
    spring.rabbitmq.listener.simple.acknowledge-mode= # 确认容器的模式
    spring.rabbitmq.listener.simple.auto-startup=true # 是否在启动时自动启动容器
    spring.rabbitmq.listener.simple.concurrency= # 侦听器调用者线程的最小数量
    spring.rabbitmq.listener.simple.default-requeue-rejected= # 默认情况下，拒绝交付是否重新排队
    spring.rabbitmq.listener.simple.idle-event-interval= # 应该多久发布一次空闲容器事件
    spring.rabbitmq.listener.simple.max-concurrency= # 侦听器调用者线程的最大数量。
    spring.rabbitmq.listener.simple.missing-queues-fatal=true # 如果容器声明的队列在代理上不可用，则是否失败和/或如果在运行时删除一个或多个队列，是否停止容器
    spring.rabbitmq.listener.simple.prefetch= # 每个消费者可能未完成的未确认消息的最大数量
    spring.rabbitmq.listener.simple.retry.initial-interval=1000ms # 第一次和第二次尝试传递消息之间的持续时间
    spring.rabbitmq.listener.simple.retry.max-attempts=3 # 传递消息的最大尝试次数
    spring.rabbitmq.listener.simple.retry.max-interval=10000ms # 尝试之间的最长持续时间
    spring.rabbitmq.listener.simple.retry.multiplier=1 # 乘数应用于上一个重试间隔
    spring.rabbitmq.listener.simple.retry.stateless=true # 重试是无状态还是有状态
    spring.rabbitmq.listener.simple.transaction-size= # 确认模式为AUTO时要在acks之间处理的消息数。如果大于预取，则预取将增加到此值
    spring.rabbitmq.listener.type=simple # Listener 容器类型
    spring.rabbitmq.password=guest # 登录以对代理进行身份验证
    spring.rabbitmq.port=5672 # RabbitMQ 端口
    spring.rabbitmq.publisher-confirms=false # 是否启用发布者确认
    spring.rabbitmq.publisher-returns=false # 是否启用发布者返回
    spring.rabbitmq.requested-heartbeat= # 请求心跳超时;零，没有。如果未指定持续时间后缀，则将使用秒
    spring.rabbitmq.ssl.algorithm= # SSL算法使用。默认情况下，由Rabbit客户端库配置
    spring.rabbitmq.ssl.enabled=false # 是否启用SSL支持
    spring.rabbitmq.ssl.key-store= # 保存SSL证书的密钥库的路径
    spring.rabbitmq.ssl.key-store-password= # 用于访问密钥库的密码
    spring.rabbitmq.ssl.key-store-type=PKCS12 # 密钥库类型
    spring.rabbitmq.ssl.trust-store= # 持有SSL证书的信任存储
    spring.rabbitmq.ssl.trust-store-password= # 用于访问信任库的密码
    spring.rabbitmq.ssl.trust-store-type=JKS # 信托商店类型
    spring.rabbitmq.ssl.validate-server-certificate=true # 是否启用服务器端证书验证
    spring.rabbitmq.ssl.verify-hostname=true # 是否启用主机名验证
    spring.rabbitmq.template.default-receive-queue= # 从明确指定none时接收消息的默认队列的名称
    spring.rabbitmq.template.exchange= # 用于发送操作的默认交换的名称
    spring.rabbitmq.template.mandatory= # 是否启用强制消息
    spring.rabbitmq.template.receive-timeout= # receive（）操作的超时时间
    spring.rabbitmq.template.reply-timeout= # sendAndReceive（）操作的超时时间
    spring.rabbitmq.template.retry.enabled=false # 是否启用发布重试
    spring.rabbitmq.template.retry.initial-interval=1000ms # 第一次和第二次尝试传递消息之间的持续时间
    spring.rabbitmq.template.retry.max-attempts=3 # 传递消息的最大尝试次数
    spring.rabbitmq.template.retry.max-interval=10000ms # 尝试之间的最长持续时间
    spring.rabbitmq.template.retry.multiplier=1 # 乘数应用于先前的重试间隔
    spring.rabbitmq.template.routing-key= # 用于发送操作的默认路由密钥的值
    spring.rabbitmq.username=guest # 登录用户以对代理进行身份验证
    spring.rabbitmq.virtual-host= # 连接到代理时使用的虚拟主机

[eca4c8c1fa9042d599fbb2eccb441cac.png]: https://img-blog.csdnimg.cn/eca4c8c1fa9042d599fbb2eccb441cac.png