【RabbitMQ】基本使用：Spring AMQP配置使用及SpringBoot整合

灰太狼 2022-12-22 00:56 255阅读 0赞

在上一篇 [【RabbitMQ】Spring AMQP核心组件解析][RabbitMQ_Spring AMQP]我们介绍了 Spring AMQP ，那它具体该如何使用呢？

## 1.xml配置方式 ##

<rabbit:connection-factory id="connectionFactory" virtual-host="/" username="guest" 
                               password="guest" host="127.0.0.1" port="5672" />
    
    
    <rabbit:admin id="connectAdmin" connection-factory="connectionFactory" />

### 1.1 Direct Exchange ###

<rabbit:queue name="MY_FIRST_QUEUE" durable="true" auto-delete="false" exclusive="false" 
                  declared-by="connectAdmin" />
    
    
    <rabbit:direct-exchange name="MY_DIRECT_EXCHANGE" durable="true" auto-delete="false" 
                            declared-by="connectAdmin">
        <rabbit:bindings>
            <rabbit:binding queue="MY_FIRST_QUEUE" key="FirstKey">
            </rabbit:binding>
        </rabbit:bindings>
    </rabbit:direct-exchange>
    
    
    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" 
                     exchange="MY_DIRECT_EXCHANGE" />
    
    
    <bean id="messageReceiver" class="com.mymq.consumer.FirstConsumer"></bean>
    
    
    <rabbit:listener-container connection-factory="connectionFactory">
        <rabbit:listener queues="MY_FIRST_QUEUE" ref="messageReceiver" />
    </rabbit:listener-container>
    
    
    <rabbit:queue name="MY_SECOND_QUEUE" durable="true" auto-delete="false" exclusive="false" 
                  declared-by="connectAdmin" />
    
    
    <rabbit:direct-exchange name="MY_DIRECT_EXCHANGE" durable="true" auto-delete="false"
                            declared-by="connectAdmin">
        <rabbit:bindings>
            <rabbit:binding queue="MY_SECOND_QUEUE" key="SecondKey"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:direct-exchange>
    
    
    <bean id="receiverSecond" class="com.mymq.consumer.SecondConsumer"></bean>
    
    
    <rabbit:listener-container connection-factory="connectionFactory">
        <rabbit:listener queues="MY_SECOND_QUEUE" ref="receiverSecond" />
    </rabbit:listener-container>

### 1.2 Topic Exchange ###

<rabbit:queue name="MY_THIRD_QUEUE" durable="true" auto-delete="false" exclusive="false" 
                  declared-by="connectAdmin" />
    
    
    <rabbit:topic-exchange name="MY_TOPIC_EXCHANGE" durable="true" auto-delete="false" 
                           declared-by="connectAdmin">
        <rabbit:bindings>
            <rabbit:binding queue="MY_THIRD_QUEUE" pattern="#.Third.#"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:topic-exchange>
    
    
    <rabbit:template id="amqpTemplate2" connection-factory="connectionFactory" 
                     exchange="MY_TOPIC_EXCHANGE" />
    
    
    <bean id="receiverThird" class="com.mymq.consumer.ThirdConsumer"></bean>
    
    
    <rabbit:listener-container connection-factory="connectionFactory">
        <rabbit:listener queues="MY_THIRD_QUEUE" ref="receiverThird" />
    </rabbit:listener-container>

### 1.3 Fanout Exchange ###

<rabbit:queue name="MY_FOURTH_QUEUE" durable="true" auto-delete="false" exclusive="false" 
                  declared-by="connectAdmin" />
    
    
    <rabbit:fanout-exchange name="MY_FANOUT_EXCHANGE" auto-delete="false" durable="true"
                            declared-by="connectAdmin" >
        <rabbit:bindings>
            <rabbit:binding queue="MY_FIRST_QUEUE"></rabbit:binding>
            <rabbit:binding queue="MY_FOURTH_QUEUE"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:fanout-exchange>
    
    
    <bean id="receiverFourth" class="com.mymq.consumer.FourthConsumer"></bean>
    
    
    <rabbit:listener-container connection-factory="connectionFactory">
        <rabbit:listener queues="MY_FOURTH_QUEUE" ref="receiverFourth" />
    </rabbit:listener-container>

## 2.SpringBoot整合 ##

使用了SpringBoot后，我们不再需要些上面那么多的配置，因为在`RabbitAutoConfiguration`中已经对Bean做了自动配置。

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>

### 2.1 定义队列、交换机 ###

@Configuration
    public class RabbitConfig {
        
        // 声明Queue、Exchange、Binding.......
    }

**1.定义交换机**

@Bean("directExchange") // 定义的name会在绑定时用到
    public DirectExchange getDirectExchange() {
        
        return new DirectExchange("DIRECT_EXCHANGE")
    }
    
    @Bean("topicExchange")
    public TopicExchange getTopicExchange(){
        
        return new TopicExchange("TOPIC_EXCHANGE");
    }
    
    @Bean("fanoutExchange")
    public FanoutExchange getFanoutExchange(){
        
        return  new FanoutExchange("FANOUT_EXCHANGE");
    }

**2.定义队列**

@Bean("firstQueue")
    public Queue getFirstQueue(){
        
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-message-ttl",6000); // 设置超时
        Queue queue = new Queue("FIRST_QUEUE", false, false, true, args);
        return queue;
    }
    
    @Bean("secondQueue")
    public Queue getSecondQueue(){
        
        return new Queue("SECOND_QUEUE");
    }
    
    @Bean("thirdQueue")
    public Queue getThirdQueue(){
        
        return new Queue("THIRD_QUEUE");
    }

**3.定义绑定**

@Bean
    public Binding bindFirst(@Qualifier("firstQueue") Queue queue, // 队列
                              @Qualifier("directExchange") TopicExchange exchange){
         // 交换机
        return BindingBuilder.bind(queue).to(exchange).with("my"); // 路由键
    }
    
    @Bean
    public Binding bindSecond(@Qualifier("secondQueue") Queue queue, 
                              @Qualifier("topicExchange") TopicExchange exchange){
         
        return BindingBuilder.bind(queue).to(exchange).with("#.my.#"); // 路由键有通配符
    }
    
    @Bean
    public Binding bindThird(@Qualifier("thirdQueue") Queue queue,
                             @Qualifier("fanoutExchange") FanoutExchange exchange){
        
        return BindingBuilder.bind(queue).to(exchange); // fanout没有路由键
    }

### 2.2 Producer ###

void convertAndSend(String exchange, String routingKey, Object message) throws AmqpException;

@Component
    public class MyProvider {
        
    
        @Autowired
        AmqpTemplate amqpTemplate;
    
        public void send(){
        
            // 发送4条消息        
            amqpTemplate.convertAndSend("DIRCET_EXCHANGE","my","a direct msg"); 
    
            amqpTemplate.convertAndSend("TOPIC_EXCHANGE", "msg", "a topic msg1");
            amqpTemplate.convertAndSend("TOPIC_EXCHANGE", "a.my.b", "a topic msg2");
    
            amqpTemplate.convertAndSend("FANOUT_EXCHANGE","","a fanout msg");
        }
    }

### 2.3 Consumer ###

@Component
    @RabbitListener(queues = "FIRST_QUEUE") // 指定监听的队列，
    public class FirstConsumer {
        
    
        @RabbitHandler // 有消息时，就会自动调用当前方法
        public void process(String msg){
        
            System.out.println(" first queue received msg : " + msg);
        }
    }

@Component
    @RabbitListener(queues = {
        "SECOND_QUEUE","THIRD_QUEUE"}) // 配置一个消费者监听多个队列
    public class SecondConsumer {
        
    
        @RabbitHandler
        public void process(String msg){
        
            System.out.println(" second queue received msg : " + msg);
        }
    }

### 2.4 配置文件 ###

全部配置总体上分成三大类：连接类、消息发送类、消息消费类。

**1.连接配置**

# rabbitmq连接基本配置
    spring.rabbitmq.addresses=43.105.136.120:5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    spring.rabbitmq.virtual-host=/
    spring.rabbitmq.connection-timeout=15000
    
    # 开启confirm机制
    spring.rabbitmq.publisher-confirms=true
    # 开启return模式
    spring.rabbitmq.publisher-returns=true
    # 配合return机制使用，表示接收路由不可达的消息
    spring.rabbitmq.template.mandatory=true

<table> 
 <thead> 
  <tr> 
   <th align="left">属性值</th> 
   <th align="left">说明</th> 
   <th align="left">默认值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">address</td> 
   <td align="left">客户端连接的地址，有多个的时候用逗号分隔<br>改地址可以IP与port的结合</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">host</td> 
   <td align="left">RabbitMQ的主机地址</td> 
   <td align="left">localhost</td> 
  </tr> 
  <tr> 
   <td align="left">port</td> 
   <td align="left">RabbitMQ的端口号</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">virtual-host</td> 
   <td align="left">连接到RabbitMQ的虚拟主机</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">username</td> 
   <td align="left">登录到RabbitMQ的用户名</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">password</td> 
   <td align="left">登录到RabbitMQ的密码</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">ssl.enabled</td> 
   <td align="left">启用SSL支持</td> 
   <td align="left">false</td> 
  </tr> 
  <tr> 
   <td align="left">ssl.key-store</td> 
   <td align="left">保存SSL证书的地址</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">ssl.key-store-password</td> 
   <td align="left">访问SSL证书的地址使用的密码</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">ssl.trust-store</td> 
   <td align="left">SSL的可信地址</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">ssl.trust-store-password</td> 
   <td align="left">访问SSL的可信地址的密码</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">ssl.algorithm</td> 
   <td align="left">SSL算法，默认使用Rabbit的客户端算法库</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">cache.channel.checkout-timeout</td> 
   <td align="left">当缓存已满时，获取Channel的等待时间，单位为毫秒</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">cache.channel.size</td> 
   <td align="left">缓存中保持的Channe丨数量</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">cache.connection.mode</td> 
   <td align="left">连接缓存的模式</td> 
   <td align="left">CHANNEL</td> 
  </tr> 
  <tr> 
   <td align="left">cache.connection.size</td> 
   <td align="left">缓存的连接数</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">connnection-timeout</td> 
   <td align="left">连接超时参数单位为毫秒：设置为“0”代表无穷大</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">dynamic</td> 
   <td align="left">默认创建—个AmqpAdmin的Beantrue</td> 
   <td align="left"></td> 
  </tr> 
 </tbody> 
</table>

> 注：Producer与Consumer都需要先配置RabbitMQ连接信息。

**2.消息发送配置（Producer）**

# 开启confirm机制
    spring.rabbitmq.publisher-confirms=true
    # 开启return模式
    spring.rabbitmq.publisher-returns=true
    # 配合return机制使用，表示接收路由不可达的消息
    spring.rabbitmq.template.mandatory=true

<table> 
 <thead> 
  <tr> 
   <th align="left">属性值</th> 
   <th align="left">说明</th> 
   <th align="left">默认值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">publisher-confirms</td> 
   <td align="left">开启 Publisher Confirm 机制</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">publisher-returns</td> 
   <td align="left">开启 Publisher Return 机制</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">template.mandatory</td> 
   <td align="left">启用强制信息</td> 
   <td align="left">false</td> 
  </tr> 
  <tr> 
   <td align="left">template.receive-timeout</td> 
   <td align="left">receive()方法的超时时间</td> 
   <td align="left">0</td> 
  </tr> 
  <tr> 
   <td align="left">template.reply-timeout</td> 
   <td align="left">sendAndReceive()方法的超时时间</td> 
   <td align="left">5000</td> 
  </tr> 
  <tr> 
   <td align="left">template.retry.enabled</td> 
   <td align="left">设置为true的时候RabbitTemplate能够实现重试</td> 
   <td align="left">false</td> 
  </tr> 
  <tr> 
   <td align="left">template.retry.initial-interval</td> 
   <td align="left">第一次与第二次发布消息的时间间隔</td> 
   <td align="left">1000</td> 
  </tr> 
  <tr> 
   <td align="left">template.retry.max-attempts</td> 
   <td align="left">尝试发布消息的最大数量</td> 
   <td align="left">3</td> 
  </tr> 
  <tr> 
   <td align="left">template.retry.max-interval</td> 
   <td align="left">尝试发布消息的最大时间间隔</td> 
   <td align="left">10000</td> 
  </tr> 
  <tr> 
   <td align="left">template.retry.multiplier</td> 
   <td align="left">上一次尝试时间间隔的乘数</td> 
   <td align="left">1.0</td> 
  </tr> 
 </tbody> 
</table>

**3.消息消费配置（Consumer）**

# 设置签收模式：AUTO(自动签收)、MANUAL(手工签收)、NONE(不签收，没有任何操作)
    spring.rabbitmq.listener.simple.acknowledge-mode=MANUAL
    # 设置当前消费者数量(线程数)
    spring.rabbitmq.listener.simple.concurrency=5
    # 设置消费者最大并发数量
    spring.rabbitmq.listener.simple.max-concurrency=10

<table> 
 <thead> 
  <tr> 
   <th align="left">属性值</th> 
   <th align="left">说明</th> 
   <th align="left">默认值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">listener.simple.acknowledge-mode</td> 
   <td align="left">容器的acknowledge模式</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">listener.simple.auto-startup</td> 
   <td align="left">肩动时自动启动容器</td> 
   <td align="left">true</td> 
  </tr> 
  <tr> 
   <td align="left">listener.simple.concurrency</td> 
   <td align="left">消费者的最小数量</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">listener.simple.default-requeue-rejected</td> 
   <td align="left">投递失败时是否重新排队</td> 
   <td align="left">true</td> 
  </tr> 
  <tr> 
   <td align="left">listener.simple.max-concurrency</td> 
   <td align="left">消费者的最大数量</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">listener.simple.missing-queues-fatal</td> 
   <td align="left">容器上声明的队列不可用时是否失敗</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">listener.simple.prefetch</td> 
   <td align="left">在单个请求中处理的消息个数，他应该大于等于事务数量</td> 
   <td align="left"></td> 
  </tr> 
  <tr> 
   <td align="left">listener.simple.retry.enabled</td> 
   <td align="left">不论是不是重试的发布</td> 
   <td align="left">false</td> 
  </tr> 
  <tr> 
   <td align="left">listener.simple.retry.initial-interval</td> 
   <td align="left">第一次与第二次投递尝试的时间间隔</td> 
   <td align="left">1000ms</td> 
  </tr> 
  <tr> 
   <td align="left">listener.simple.retry.max-attempts</td> 
   <td align="left">尝试投递消息的最大数量</td> 
   <td align="left">3</td> 
  </tr> 
  <tr> 
   <td align="left">listener.simple.retry.max-interval</td> 
   <td align="left">两次尝试的最大时间间隔</td> 
   <td align="left">10000ms</td> 
  </tr> 
  <tr> 
   <td align="left">listener.simple.retry.multiplier</td> 
   <td align="left">上一次尝试时间间隔的乘数</td> 
   <td align="left">1.0</td> 
  </tr> 
  <tr> 
   <td align="left">listener.simple.retry.stateless</td> 
   <td align="left">重试是有状态的还是无状态的</td> 
   <td align="left">true</td> 
  </tr> 
  <tr> 
   <td align="left">listener.simple.transaction-size</td> 
   <td align="left">在一个事务中处理的消息数量 = 为了获得最佳效果， <br>该值应设罝为小于等于每个请求中处理的消息个数，即 listener.prefetch 的值</td> 
   <td align="left"></td> 
  </tr> 
 </tbody> 
</table>

[RabbitMQ_Spring AMQP]: https://blog.csdn.net/weixin_43935927/article/details/109737660