在上一篇文章中，笔者整理了从消息生产出来到消费结束的整个生命周期过程中，为了确保消息能够可靠到达或者消费，我们需要在哪些环节进行哪些处理，同时也展示了使用Java原生代码怎么样在这些环节进行处理。本文主要介绍使用spring boot集成RabbitMQ的方式时，针对这些环节应该进行怎样的处理。

一、创建Exchange、Queue和Binding

        首先，需要创建待测试的交换机、队列和绑定。相对于原生代码，spring boot对ConnectionFactory、Channel这些对象的创建和销毁进行了封装，使得我们不再需要手动创建Connection或者Channel，也不需要手动进行释放。这样做的一个好处就是：我们不必再关心这些系统资源的生命周期，从而简化了开发，而且避免了因忘记释放资源造成的内存泄露。RabbitMQ的Exchange、Queue和Binding这些组件的创建，只需要创建相应的Bean即可，注入到IOC中。

        比如，笔者通过一个RabbitConfig的自动配置类，对这些Bean进行了注入：

@Configuration
public class RabbitConfig {
    @Value("${spring.rabbitmq.host}")
    private String host;
    @Value("${spring.rabbitmq.port}")
    private int port;
    @Value("${spring.rabbitmq.username}")
    private String username;
    @Value("${spring.rabbitmq.password}")
    private String password;
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setPublisherConfirms(true);
        return connectionFactory;
    }
    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        return template;
    }
    @Bean
    public Queue TestDirectQueue() {
        return new Queue("TestDirectQueue", true, false, false);
    }
    @Bean
    DirectExchange TestDirectExchange() {
        return new DirectExchange("TestDirectExchange", true, false);
    }
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectQueue");
    }
}

第3~10行：是RabbitMQ连接参数的配置

第11~19行：根据配置的连接参数，创建链接工厂ConnectionFactory ，这里我们只需要创建ConnectionFactory ，spring boot会自动创建和管理Connection以及Channel。

第20~24行：创建RabbitTemplate对象。RabbitTemplate是采用模板方法模式进行消息发送的一个模板，后面我们在发送消息时就是使用RabbitTemplate的相应的方法。实际上是在RabbitTemplate内部对Connection和Channel的创建进行了封装，而且只有我们在使用其第一次发送消息时，才会真正在RabbitMQ的broker上创建声明好的Exchange、Queue、Binding等组件。

第25~28行：创建名称为TestDirectQueue的队列

第29~32行：创建名称为TestDirectExchange的直连交换机

第33~36行：创建绑定，将上面创建的交换机和队列绑定在一起，路由键为队列的名称。

二、生产者端处理连接异常

        在前面我们已经提到，spring boot集成RabbitMQ之后，使用RabbitTemplate对象进行消息的发送，所以生产者端的异常处理需要在调用RabbitTemplate对象发送消息的代码上。我们在处理Java原生代码调用RabbitMQ要处理的异常主要是：

• IOException —— 客户端连不上broker的情况，抛出的异常。
• TimeoutException —— 客户端连接broker超时抛出的异常。
• ShutdownSignalException —— broker的交换机不存在时出现的异常。

        但是在spring boot的框架中，引入了AmqpConnectException异常，实际上一旦发生MQ掉线或者超时的情况，AmqpConnectException异常取代了IOException 和TimeoutException。所以完整的异常处理的代码如下：

try {
    Message message = new Message(str.getBytes());
    rabbitTemplate.send("TestDirectExchange", "TestDirectQueue", message);
    User user = new User("张三", 18);
    rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectQueue", user);
}  catch (ShutdownSignalException e) {
    System.out.println("交换机故障或者不存在。");
    e.printStackTrace();
}catch (AmqpConnectException e) {
    System.out.println("服务器连接失败。");
    e.printStackTrace();
}

第3行和第5行：分别调用rabbitTemplate的send和convertAndSend来发送消息，两者的不同在于前者发送的参数为Message对象，而该对象中封装了发送的消息的字节数组；后者发送的是一个自定义的类对象，但是要注意类需要实现Serializable接口。

第6~8行：对ShutdownSignalException 类型异常的处理，当交换机不存在时会产生此类异常。

第9~11行：对AmqpConnectException类型的异常的处理，当客户端连接不上Broker时会抛出此类的异常。

三、生产者端手动确认消息

        在生产者端开启手动确认消息，可以保证由于RabbitMQ自身的原因导致消息路由到队列失败时，我们可以进行手动的消息重发。在java原生代码调用RabbitMQ时需要手动开启通道的confirm模式，而且提供了3种接受broker返回的ack的方法。在spring-boot-starter-amqp中，进一步对这个逻辑进行了封装，只保留了使用回调方法的方式。

        下面，先创建一个回调的类：

public class MyConfirmCallback implements RabbitTemplate.ConfirmCallback {

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("MyConfirmCallback:ack=" + ack);
    }
}

        可以看到，回调类实现了ConfirmCallback 接口，接口中有一个回调方法confirm，confirm具有三个参数：

• correlationData —— 发送消息时携带的附加数据，send方法和convertAndSend方法都具有携带这个参数的重载类型：

void send(String routingKey, Message message, CorrelationData correlationData);
void convertAndSend(String routingKey, Object object, CorrelationData correlationData);

• ack —— 消息是否成功发送到exchange
• cause —— 消息发送失败的原因

        有了上述回调类的实现，我们就可以在发送时指定回调类，比如上面的发送消息代码就可以改成这样：

try {
    Message message = new Message(str.getBytes());
    rabbitTemplate.setConfirmCallback(new MyConfirmCallback());
    rabbitTemplate.send("TestDirectExchange", "TestDirectQueue", message);
    User user = new User("张三", 18);
    rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectQueue", user);
}  catch (ShutdownSignalException e) {
    System.out.println("交换机故障或者不存在。");
    e.printStackTrace();
}catch (AmqpConnectException e) {
    System.out.println("服务器连接失败。");
    e.printStackTrace();
}

        在第3行中，我们为rabbitTemplate指定了回调的类对象，这里需要特别注意：

（1）一定要在发送消息之前，也就是在调用send方法或者convertAndSend方法之前调用setConfirmCallback指定回调对象

（2）只能为rabbitTemplate指定一次回调对象，否则会抛出异常

四、生产者端处理Return的消息

        在上一篇文章中，我们特别指出了一种场景：消息从交换机路由到队列中时，我们可以选择当队列缺失时，将消息返回给生产者。在spring-boot-starter-amqp中，对这种应用也进行了封装，与上一小节中的手动确认消息类似，我们需要先创建一个接受Return消息的回调类：

public class MyReturnCallback implements RabbitTemplate.ReturnsCallback {

    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        String str = String.format("消息发送失败-消息回退，应答码：{}，原因：{}，交换机：{}，路由键：{}",
                returnedMessage.getReplyCode(),
                returnedMessage.getReplyText(),
                returnedMessage.getExchange(),
                returnedMessage.getRoutingKey());
        System.out.println(str);
    }
}

        回调类实现了ReturnsCallback 接口，接口中有一个回调方法returnedMessage，方法只有一个参数ReturnedMessage，先看下ReturnedMessage类型的定义：

public class ReturnedMessage {
    private final Message message;
    private final int replyCode;
    private final String replyText;
    private final String exchange;
    private final String routingKey;
}

• message —— 细心的朋友可能已经观察到，这个参数的类型实际上就是rabbitTemplate.send方法中封装的消息的类型，当消息没有路由到队列被return时被原样返回。
• replyCode —— 返回时的应答码
• replyText —— 返回的原因
• exchange —— 交换机名称
• routingKey —— 路由键，如果在直连模式下就是队列名称

        在实现了回调类之后，就可以进行回调类对象的装配了：

try {
    Message message = new Message(str.getBytes());
    rabbitTemplate.setConfirmCallback(new MyConfirmCallback());
    rabbitTemplate.setMandatory(true);
    rabbitTemplate.setReturnsCallback(new MyReturnCallback());
    rabbitTemplate.send("TestDirectExchange", "TestDirectQueue", message);
    User user = new User("张三", 18);
    rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectQueue", user);
}  catch (ShutdownSignalException e) {
    System.out.println("交换机故障或者不存在。");
    e.printStackTrace();
}catch (AmqpConnectException e) {
    System.out.println("服务器连接失败。");
    e.printStackTrace();
}

        注意第4行和第5行是我们新增的代码，对于设置手动处理broker return的消息来说，第4行代码必不可少，意思是开启手动处理回退消息。对于Return消息的回调类的使用也需要注意：

（1）一定要在发送消息之前，也就是在调用send方法或者convertAndSend方法之前调用setReturnsCallback指定回调对象（注意是使用setReturnsCallback，而非setReturnCallback，后者已经被标注过时）

（2）只能为rabbitTemplate指定一次回调对象，否则会抛出异常

（3）在使用setReturnsCallback之前要使用rabbitTemplate.setMandatory(true)。

        如果不出什么意外，当消息被return时，程序会打印类似以下的信息（红色圈出部分）：

五、消费者端手动ACK

        在前面我们梳理了RabbitMQ生产者端一些消息的可靠性传输的保证机制，下面再学习下消费者端怎么进行手动ack。消费者端的手动ack，首先要开启消费者端支持手动ack（默认是自动ack），开启的方式有两种：

• 第一种，使用配置文件开启手动ack：

spring:
  #配置rabbitMq 服务器
  rabbitmq:
    listener:
      type: simple
      simple:
        #simple关闭自动ack,手动ack
        acknowledge-mode: manual

第8行，修改消费者确认消息的模式为手动

• 第二种，自定义containerFactory，开启手动确认模式：

@Bean
@ConditionalOnClass
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(CachingConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    return factory;
}

        注意第6行，开启了手动确认模式。

        在完成了上述任意一种设置之后，然后配置监听器，并手动进行确认：

@RabbitListener(containerFactory = "rabbitListenerContainerFactory", bindings = @QueueBinding(value = @Queue(value = "TestDirectQueue", durable = "true"),
        exchange = @Exchange(name = "TestDirectQueue", durable = "true", type = "direct")))
public void consumer(@Payload Message message, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException {
    System.out.println("接受到的消息是：" + message.toString());
    channel.basicAck(deliveryTag,false);
    System.out.print("这里是接收者1答应消息： ");
    System.out.println("SYS_TOPIC_ORDER_CALCULATE_ZZ_FEE process1  : " + message);
}

第1行 —— 使用RabbitListener注解指定监听器，同时装配自定义的containerFactory（如果使用配置文件的方式手动确认，无需装配containerFactory），同时指定的还有exchange、queue和binding等等。

第3行 —— consumer包含了三个参数：message是传递过来的消息，deliveryTag是为每个消息指定的自增长的id（详细在上一篇文章中已经解释过），channel传递信息的信道。

第5行 —— 通过调用channel的basicAck进行消息的确认，参数的解释可以参照笔者上一篇文章。

        至此，整个消费者的手动确认方法也介绍完毕。

六、总结

        下面，针对本文的内容进行总结：

（1）spring-boot-starter-amqp中，针对生产者端或者消费者端连接不到broker的IOException异常和TimeoutException异常，重新封装了新的异常类型AmqpConnectException异常。

（2）生产者发送消息时如果交换机不存在，会抛出ShutdownSignalException异常。

（3）生产者端可以手动确认发送的消息正常到达了交换机，方法是实现ReturnsCallback接口，然后使用rabbitTemplate.setConfirmCallback()方法进行装配。

（4）如果broker出现内部异常或者目标队列不存在时，可以设置消息返还给生产者，方法是设置rabbitTemplate.setMandatory(true)，实现ReturnsCallback接口，然后使用rabbitTemplate.setReturnsCallback()进行装配。

（5）开启消费者手动ack有两种方法，一种是通过配置文件修改支持，另一种是创建自定义的SimpleRabbitListenerContainerFactory并注入到IOC中，在监听方法中调用channel.basicAck()。

七、附yml文件中常见的RabbitMQ相关配置：

spring:
  #配置rabbitMq 服务器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    #虚拟host 可以不设置,使用server默认host
    #virtual-host: JCcccHost
    publisher-confirm-type: correlated
    #发布者到达确认
    publisher-returns: true
    listener:
      type: simple
      simple:
        #simple关闭自动ack,手动ack
        acknowledge-mode: manual
        retry:
          ### 开启重试机制(调用监听方法失败时会重试，不是从队列中重复拿消息)
          enabled: true
          #最大重试传递次数
          max-attempts: 3
          #第一次和第二次尝试传递消息的间隔时间 单位毫秒
          initial-interval: 5000ms
          #最大重试时间间隔，单位毫秒
          max-interval: 300000ms
          #应用前一次重试间隔的乘法器，multiplier默认为1
          multiplier: 3
          #以上配置的间隔0s  5s  15s  45s
        #重试次数超过上面的设置之后是否丢弃(消费者listener抛出异常，是否重回队列，默认true：重回队列， false为不重回队列(结合死信交换机))
        default-requeue-rejected: true
    ### 模板配置
    ##设置为 true 后 消费者在消息没有被路由到合适队列情况下会被return监听，而不会自动删除
    template:
      mandatory: true