本文共 12600 字，大约阅读时间需要 42 分钟。

代码实践

注意一点，在发送消息的时候对template进行配置mandatory=true保证监听有效

消费端核心配置

1.首先配置手工确认模式，用于ACK的手工处理，这样我们可以保证消息的可靠性送达，或者在消费端消费失败的时候可以做到重回队列、根据业务记录日志等处理

2可以设置消费端的监听个数和最大个数，用于控制消费端的并发情况

@RabbitListener注解的使用

消费端监听@RabbitListener注解，这个对于在实际工作中非常的好用

@RabbitListener是一个组合注解，里面可以注解配置（@QueueBinding、@Queue、@Exchange）直接通过这个组合注解一次性搞定消费端交换机、队列、绑定、路由、并且配置监听功能等

注：由于类配置写在代码里非常不友好，所以强烈建议大家使用配置文件配置

rabbitmq-common子项目

package com.zxp.rabbitmqcommon.entity;import java.io.Serializable;/** * @author笑笑 * @site www.xiaoxiao.com * @company * @create 2020-03-04 17:08 * * 支付完毕后下单操作 */public class Order implements Serializable {    private String id;    private String name;    public Order() {    }    public Order(String id, String name) {        super();        this.id = id;        this.name = name;    }    public String getId() {        return id;    }    public void setId(String id) {        this.id = id;    }    public String getName() {        return name;    }    public void setName(String name) {        this.name = name;    }}

rabbitmq-springcloud-consumer子项目

   
       
    
     4.0.0
        
            
     
      org.springframework.boot
             
     
      spring-boot-starter-parent
             
     
      2.2.5.RELEASE
             
      
         
        
    
     com.zxp
        
    
     rabbitmq-springcloud-consumer
        
    
     0.0.1-SNAPSHOT
        
    
     rabbitmq-springcloud-consumer
        
    
     Demo project for Spring Boot
        
            
     
      1.8
         
        
            
                 
      
       org.springframework.boot
                  
      
       spring-boot-starter
              
             
                 
      
       org.springframework.boot
                  
      
       spring-boot-starter-test
                  
      
       test
                  
                      
                           
        
         org.junit.vintage
                            
        
         junit-vintage-engine
                        
                   
              
             
                 
      
       com.zxp
                  
      
       rabbitmq-common
                  
      
       0.0.1-SNAPSHOT
              
             
                 
      
       org.springframework.boot
                  
      
       spring-boot-starter-amqp
              
             
                 
      
       junit
                  
      
       junit
                  
      
       4.12
                  
      
       test
              
         
        
            
                 
                      
       
        org.springframework.boot
                       
       
        spring-boot-maven-plugin
                   
              
         
    
   

Yml配置

spring.rabbitmq.addresses=192.168.23.130:5672spring.rabbitmq.username=guestspring.rabbitmq.password=guestspring.rabbitmq.virtual-host=/spring.rabbitmq.connection-timeout=15000spring.rabbitmq.listener.simple.acknowledge-mode=manualspring.rabbitmq.listener.simple.concurrency=5spring.rabbitmq.listener.simple.max-concurrency=10spring.rabbitmq.listener.order.queue.name=queue-2spring.rabbitmq.listener.order.queue.durable=truespring.rabbitmq.listener.order.exchange.name=exchange-2spring.rabbitmq.listener.order.exchange.durable=truespring.rabbitmq.listener.order.exchange.type=topicspring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=truespring.rabbitmq.listener.order.key=springboot.*

RabbitReceiver

package com.zxp.rabbitmqspringcloudconsumer.conusmer;import com.zxp.rabbitmqcommon.entity.Order;import com.rabbitmq.client.Channel;import org.springframework.amqp.rabbit.annotation.*;import org.springframework.amqp.support.AmqpHeaders;import org.springframework.messaging.Message;import org.springframework.messaging.handler.annotation.Headers;import org.springframework.messaging.handler.annotation.Payload;import org.springframework.stereotype.Component;import java.util.Map;@Componentpublic class RabbitReceiver {    /**     * 将队列、交换机、以及他们之间的关系申明在代码中     *     * 模拟的是短信微服务c从MQ服务器中接收到的调第三方接口发短信的消息     */    @RabbitListener(bindings = @QueueBinding(            value = @Queue(value = "queue-1",                    durable="true"),            exchange = @Exchange(value = "exchange-1",                    durable="true",                    type= "topic",                    ignoreDeclarationExceptions = "true"),            key = "springboot.*"    )    )    @RabbitHandler    public void onMessage(Message message, Channel channel) throws Exception {        System.err.println("--------------------------------------");        System.err.println("消费端Payload: " + message.getPayload());        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);        //手工ACK        channel.basicAck(deliveryTag, false);    }    /**     *     * spring.rabbitmq.listener.order.queue.name=queue-2     * spring.rabbitmq.listener.order.queue.durable=true     * spring.rabbitmq.listener.order.exchange.name=exchange-1     * spring.rabbitmq.listener.order.exchange.durable=true     * spring.rabbitmq.listener.order.exchange.type=topic     * spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true     * spring.rabbitmq.listener.order.key=springboot.*     * @param order     * @param channel     * @param headers     * @throws Exception     *     *  将队列、交换机、以及他们之间的关系申明在全局配置文件中     *     */    @RabbitListener(bindings = @QueueBinding(            value = @Queue(value = "${spring.rabbitmq.listener.order.queue.name}",                    durable="${spring.rabbitmq.listener.order.queue.durable}"),            exchange = @Exchange(value = "${spring.rabbitmq.listener.order.exchange.name}",                    durable="${spring.rabbitmq.listener.order.exchange.durable}",                    type= "${spring.rabbitmq.listener.order.exchange.type}",                    ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"),            key = "${spring.rabbitmq.listener.order.key}"    )    )    @RabbitHandler    public void onOrderMessage(@Payload Order order,                               Channel channel,                               @Headers Map
   
     headers) throws Exception {        System.err.println("--------------------------------------");        System.err.println("消费端order: " + order.getId());//        实际项目开发中  this.orderDao.save(order)        Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);        // 手工ACK        channel.basicAck(deliveryTag, false);    }}
   

rabbitmq-springcloud-producer子项目

package com.zxp.rabbitmqspringcloudproducer;import org.springframework.context.annotation.ComponentScan;import org.springframework.context.annotation.Configuration;/** * @author笑笑 * @site www.xiaoxiao.com * @company * @create 2020-03-04 17:31 */@Configuration@ComponentScan({"com.zxp.rabbitmqspringcloudproducer.*"})public class MainConfig {}

Pom依赖

   
       
    
     4.0.0
        
            
     
      org.springframework.boot
             
     
      spring-boot-starter-parent
             
     
      2.2.5.RELEASE
             
      
         
        
    
     com.zxp
        
    
     rabbitmq-springcloud-producer
        
    
     0.0.1-SNAPSHOT
        
    
     rabbitmq-springcloud-producer
        
    
     Demo project for Spring Boot
        
            
     
      1.8
         
        
            
                 
      
       org.springframework.boot
                  
      
       spring-boot-starter
              
             
                 
      
       org.springframework.boot
                  
      
       spring-boot-starter-test
                  
      
       test
                  
                      
                           
        
         org.junit.vintage
                            
        
         junit-vintage-engine
                        
                   
              
             
                 
      
       com.zxp
                  
      
       rabbitmq-common
                  
      
       0.0.1-SNAPSHOT
              
             
                 
      
       org.springframework.boot
                  
      
       spring-boot-starter-amqp
              
             
                 
      
       junit
                  
      
       junit
                  
      
       4.12
                  
      
       test
              
         
        
            
                 
                      
       
        org.springframework.boot
                       
       
        spring-boot-maven-plugin
                   
              
         
    
   

Yml配置

spring.rabbitmq.addresses=192.168.23.130:5672spring.rabbitmq.username=guestspring.rabbitmq.password=guestspring.rabbitmq.virtual-host=/spring.rabbitmq.connection-timeout=15000spring.rabbitmq.publisher-confirms=truespring.rabbitmq.publisher-returns=truespring.rabbitmq.template.mandatory=true

RabbitSender.java

package com.zxp.rabbitmqspringcloudproducer.producer;import com.zxp.rabbitmqcommon.entity.Order;import org.springframework.amqp.rabbit.connection.CorrelationData;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.messaging.Message;import org.springframework.messaging.MessageHeaders;import org.springframework.messaging.support.MessageBuilder;import org.springframework.stereotype.Component;import java.util.Map;@Componentpublic class RabbitSender {    //自动注入RabbitTemplate模板类    @Autowired    private RabbitTemplate rabbitTemplate;    //回调函数: confirm确认    final ConfirmCallback confirmCallback = new ConfirmCallback() {        @Override        public void confirm(CorrelationData correlationData, boolean ack, String cause) {            System.err.println("correlationData: " + correlationData);            System.err.println("ack: " + ack);            if(!ack){                System.err.println("异常处理....");            }        }    };    //回调函数: return返回    final ReturnCallback returnCallback = new ReturnCallback() {        @Override        public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,                                    String exchange, String routingKey) {            System.err.println("return exchange: " + exchange + ", routingKey: "                    + routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);        }    };    //发送消息方法调用: 构建Message消息    public void send(Object message, Map
   
     properties) throws Exception {        MessageHeaders mhs = new MessageHeaders(properties);        Message msg = MessageBuilder.createMessage(message, mhs);        rabbitTemplate.setConfirmCallback(confirmCallback);        rabbitTemplate.setReturnCallback(returnCallback);        //id + 时间戳 全局唯一        CorrelationData correlationData = new CorrelationData("1234567890");        rabbitTemplate.convertAndSend("exchange-1", "springboot.abc", msg, correlationData);    }    //发送消息方法调用: 构建自定义对象消息    public void sendOrder(Order order) throws Exception {        rabbitTemplate.setConfirmCallback(confirmCallback);        rabbitTemplate.setReturnCallback(returnCallback);        //id + 时间戳 全局唯一        CorrelationData correlationData = new CorrelationData("0987654321");        rabbitTemplate.convertAndSend("exchange-2", "springboot.def", order, correlationData);    }}
   

测试代码

package com.zxp.rabbitmqspringcloudproducer;import com.zxp.rabbitmqcommon.entity.Order;import com.zxp.rabbitmqspringcloudproducer.producer.RabbitSender;import org.junit.Test;import org.junit.runner.RunWith;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.test.context.SpringBootTest;import org.springframework.test.context.junit4.SpringRunner;import java.text.SimpleDateFormat;import java.util.Date;import java.util.HashMap;import java.util.Map;@RunWith(SpringRunner.class)@SpringBootTestpublic class RabbitmqSpringcloudProducerApplicationTests {    @Autowired    private RabbitSender rabbitSender;    private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");    @Test    public void testSender1() throws Exception {        Map
   
     properties = new HashMap<>();        properties.put("number", "12345");        properties.put("send_time", simpleDateFormat.format(new Date()));        rabbitSender.send("让短信微服务调用第三方接口的消息!", properties);    }    @Test    public void testSender2() throws Exception {        Order order = new Order("001", "第一个订单");        rabbitSender.sendOrder(order);    }}
   

转载地址：http://iyrzi.baihongyu.com/