SpringBoot整合RabbitMQ学习记录

た入场券 2023-06-22 13:46 17阅读 0赞

### SpringBoot整合RabbitMQ学习记录 ###

*  RabbitMQ简介
 *  交换机类型介绍
 *   *  Direct Exchange：
     *  Topic：
     *  Fanout：
     *  Headers：
 *  show you code
 *   *  Direct Exchange
     *  Topic Exchange
     *  Fanout Exchang
 *  RabbitMQ消息的确认机制
 *   *  生产者消息确认机制：
     *  消费者消息确认机制：

RabbitMQ安装过程在此就不说了，去官网下载安装启动运行即可  
[RabbitMQ下载地址][RabbitMQ]  
启动后可进入后台管理控制台查看如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIwNTE0OA_size_16_color_FFFFFF_t_70]

# RabbitMQ简介 #

RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种语言平台的客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

通常我们谈到消息队列, 会有三个概念： 消息生产者(Provider)、队列(Queue)、消息消费者(Consumer)，RabbitMQ 在这个基本概念上, 多做了一层抽象, 在消息生产者和队列之间, 加入了交换器 (Exchange)。这样消息生产者和队列就没有直接联系, 变成消息生产者把消息发送给交换器, 交换器根据调度策略再把消息发送给队列。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIwNTE0OA_size_16_color_FFFFFF_t_70 1]

1.  左侧P代表消息生产者，也就是往RabbitMQ发消息的程序。
2.  中间即是RabbitMQ，其中包括交换机(Exchange)和队列(Queue)。
3.  右侧C代表消费者，也就是往RabbitMQ拿消息的程序。

其中比较重要的概念有：**虚拟主机(Virtual Host)**、**交换机(Exchange)**、**队列(Queue)**、**绑定(Binding)**。

*  虚拟主机(Virtual Hosts)：  
    在上面已经说明如何为一个用户创建一个Virtual Host，一个虚拟主机持有一组交换机、队列和绑定。在RabbitMQ当中，用户只能在虚拟主机这个粒度上进行权限的控制。 如果需要禁止A组访问B组的交换机/队列/绑定，必须为A和B分别创建一个虚拟主机。每一个RabbitMQ服务器都有一个默认的虚拟主机“/”。
 *  交换机(Exchange)：  
    交换机的功能主要是接收消息并且根据转发策略转发到对应的队列，交换机不存储消息，在启用ack模式后，交换机找不到队列会返回错误，这个ack模式后面再详细讨论。交换机有四种类型：Direct, topic, Headers and Fanout
 *  队列(Queue)：  
    队列用于存放消息的载体，一般是和交换机进行绑定，交换机根据转发策略把消息转发到队列里。
 *  绑定(Binding)：  
    也就是交换机需要和队列相绑定，这其中如上图所示，是多对多的关系。

# 交换机类型介绍 #

## Direct Exchange： ##

![在这里插入图片描述][20191216155014812.jpg]  
direct 类型的行为是”先匹配, 再投送”. 即在绑定时设定一个**binding\_key**, 消息的**routing\_key**与**binding\_key**匹配时, 才会被交换器投送到绑定的队列中去.

## Topic： ##

转发消息主要是根据通配符。 在这种交换机下，队列和交换机的绑定会定义一种路由模式，通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息。

1.  路由键必须是一串字符，用句号(.)隔开，比如说 topic.message，或者 topic.message.detail 等。
2.  路由模式必须包含一个星号(*)，主要用于匹配路由键指定位置的一个单词，比如说，一个路由模式是这样：topic.*，那么就只能匹配路由键是：topic.message、topic.other等，第一个单词是 topic，第二个单词可以是任意一个单词。 井号(\#)就表示一个或者多个单词，例如一个匹配模式是topic.\#，那么可以匹配到例如：topic.message、topic.message.detail等，以topic.开头的路由键都可以匹配到。

## Fanout： ##

Fanout类型类似于消息广播，不管路由键或者是路由模式，会把消息发给绑定给它的全部队列，如果配置了routing\_key会被忽略。

## Headers： ##

设置header attribute参数类型的交换机

# show you code #

本文是基于Springboot-1.5.15整合RabbitMQ来进行讲解，在真实工作中，生产者和消费者一般是在不同的项目里，各自负责不同的职责，这里为了模拟真实环境，创建两个不同的项目进行演示。创建两个maven项目，消息生产者mq-rabbit-provider和消息消费者mq-rabbit-consumer，两个项目的pom.xml文件添加相同依赖：

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

mq-rabbit-provider项目的application.properties内容如下：

server.port=8080
    spring.application.name=springboot-rabbitmq-provider
     
    spring.rabbitmq.host=10.211.55.3
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=root
    spring.rabbitmq.password=root
    #RabbitMQ的虚拟host
    spring.rabbitmq.virtual-host=CalonHost

mq-rabbit-consumer项目的application.properties内容如下：

server.port=9090
    spring.application.name=springboot-rabbitmq-consumer
     
    spring.rabbitmq.host=10.211.55.3
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=root
    spring.rabbitmq.password=root
    #RabbitMQ的虚拟host
    spring.rabbitmq.virtual-host=CalonHost

这里只是端口和应用名不同，其他都一样。

接下来分别介绍Direct、Topic、Fanout等3种不同交换机的使用例子。

## Direct Exchange ##

在mq-rabbit-provider项目建一个配置类DirectRabbitConfig.java，配置交换机、队列、BindingKey=CalonDirectRouting的绑定关系，代码如下：

@Configuration
    public class DirectRabbitConfig { 
     
        //队列
    	@Bean
        public Queue CalonDirectQueue() { 
            return new Queue("CalonDirectQueue",true);
        }
    	
        //Direct交换机
    	@Bean
        DirectExchange CalonDirectExchange() { 
            return new DirectExchange("CalonDirectExchange");
        }
    	
        //绑定
    	@Bean
        Binding bindingDirect() { 
            return BindingBuilder.bind(CalonDirectQueue()).to(CalonDirectExchange()).with("CalonDirectRouting");
        }
    }

创建一个实体类User.java，这里说明一下，*该实体类是消息的主体，所以必须实现Serializable接口，否则在消息消费者项目读取消息时会报错*，代码如下：

package mq.rabbit.entity;
     
    import java.io.Serializable;
     
    public class User implements Serializable{ 
    	
    	private static final long serialVersionUID = 1L;
    	
    	private String id;
    	private String username;
    	private String password;
    	private String type;
    	
    	public String getId() { 
    		return id;
    	}
    	public void setId(String id) { 
    		this.id = id;
    	}
    	public String getUsername() { 
    		return username;
    	}
    	public void setUsername(String username) { 
    		this.username = username;
    	}
    	public String getPassword() { 
    		return password;
    	}
    	public void setPassword(String password) { 
    		this.password = password;
    	}
    	
    	public String getType() { 
    		return type;
    	}
    	public void setType(String type) { 
    		this.type = type;
    	}
    	public User() { 
    		super();
    	}
    	public User(String id, String username, String password, String type) { 
    		super();
    		this.id = id;
    		this.username = username;
    		this.password = password;
    		this.type = type;
    	}
    }

下面创建一个Controller，利用http请求进行调试，CalonDirectExchange是上面配置的交换机标识，CalonDirectRouting就是上面绑定好的queue名字，由于上面已经配置好交换机和队列的绑定关系，这两个组合就可以知道消息最终是发送到队列CalonDirectQueue里面去了,Controller类的代码如下：

@Controller
    public class SendController { 
     
    	@Autowired
    	private RabbitTemplate template;
    	
    	@GetMapping("/sendDirect")
    	private @ResponseBody String sendDirect(String message) throws Exception { 
    		User user = new User(UUID.randomUUID().toString(), message, "123456", "sendDirect");
    		template.convertAndSend("CalonDirectExchange", "CalonDirectRouting", user);
    		return "OK,sendDirect:" + message;
    	}
    }

启动mq-rabbit-provider项目，在浏览器输入：

> http://localhost:8080/sendDirect?message=123

再去RabbitMQ的web管理后台查看，你会发现在Queue里找到刚刚添加的那个队列，后面的数字就是消息数量有变化，说明消息已经存储进去了：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIwNTE0OA_size_16_color_FFFFFF_t_70 2]

把mq-rabbit-provider项目里的User类和DirectRabbitConfig类复制到mq-rabbit-consumer项目，User类用于读取消息时接收消息对象，DirectRabbitConfig可以不复制，但是如果RabbitMQ里还没有被监听的队列时会报错，复制过来是为了让RabbitMQ里还没有被监听的队列时自动创建该队列，防止报错。  
创建队列监听类DirectReceiver.java，代码如下：

package mq.rabbit.receiver;
     
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
     
    import mq.rabbit.entity.User;
     
    @Component
    @RabbitListener(queues = "CalonDirectQueue")//CalonDirectQueue为队列名称
    public class DirectReceiver { 
    	
    	@RabbitHandler
        public void process(User user) { 
            System.out.println("DirectReceiver消费者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
        }
     
    }

启动mq-rabbit-consumer项目，就会收到之前发送到CalonDirectQueue队列的消息了，继续调用上面的请求/sendDirect，消息消费者会继续收到消息。

## Topic Exchange ##

在mq-rabbit-provider项目建一个配置类TopicRabbitConfig.java，配置交换机、队列、BindingKey的绑定关系，代码如下：

package mq.rabbit.config;
     
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
     
    /** * Topic Exchange类型交换机 * @author calon * */
    @Configuration
    public class TopicRabbitConfig { 
     
        public final static String first = "topic.first";
        public final static String second = "topic.second";
     
        @Bean
        public Queue firstQueue() { 
            return new Queue(TopicRabbitConfig.first);
        }
     
        @Bean
        public Queue secondQueue() { 
            return new Queue(TopicRabbitConfig.second);
        }
     
        @Bean
        TopicExchange exchange() { 
            return new TopicExchange("topicExchange");
        }
     
        //绑定topic.first队列到routingKey为topic.first，只有topic.first的routingKey消息才发送到此队列
        @Bean
        Binding bindingExchangeMessage() { 
            return BindingBuilder.bind(firstQueue()).to(exchange()).with(first);
        }
     
        //绑定topic.second队列到topic.#，凡是topic.开头的routingKey消息都发送到此队列
        @Bean
        Binding bindingExchangeMessage2() { 
            return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
        }
        
    }

Topic Exchange类型的交换机是基于模糊匹配规则，所以这里创建两个Queue，分别绑定到两个BindingKey：topic.first和topic.\#，用来测试消息进到哪个队列里。

在SendController类里添加两个request，代码如下：

@Controller
    public class SendController { 
     
    	@Autowired
    	private RabbitTemplate template;
    	
    	@GetMapping("/sendTopicFirst")
    	private @ResponseBody String sendTopicFirst(String message) { 
    		User user = new User(UUID.randomUUID().toString(), message, "123456", "sendTopicFirst");
    		template.convertAndSend("topicExchange", "topic.first", user);
    		return "OK,sendTopicFirst:" + message;
    	}
    	
    	@GetMapping("/sendTopicSecond")
    	private @ResponseBody String sendTopicSecond(String message) { 
    		User user = new User(UUID.randomUUID().toString(), message, "123456", "sendTopicSecond");
    		template.convertAndSend("topicExchange", "topic.second", user);
    		return "OK,sendTopicSecond:" + message;
    	}
    }

当我们调用/sendTopicFirst请求时，交换机为topicExchange，routingKey为topic.first，按照上面bindingKey的配置，可以匹配到topic.first和topic.\#规则，对应的队列是topic.first和topic.second,所以一条消息进到两个队列里。

当调用/sendTopicSecond请求时，交换机为topicExchange，routingKey为topic.second，匹配到topic.\#规则，对应的队列是topic.second，所以消息进到topic.second队列里，除了\#匹配规则，大家可以自行试试星号(\*)这个匹配规则，\*符号是匹配一个单词的。

把mq-rabbit-provider项目里的TopicRabbitConfig类复制到mq-rabbit-consumer项目，分别创建TopicFirstReceiver和TopicSecondReceiver消息监听类，代码如下：

package mq.rabbit.receiver;
     
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    import mq.rabbit.entity.User;
     
    @Component
    @RabbitListener(queues = "topic.first")
    public class TopicFirstReceiver { 
    	
    	@RabbitHandler
        public void process(User user) { 
    		System.out.println("TopicFirstReceiver消费者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
        }
    }

package mq.rabbit.receiver;
     
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    import mq.rabbit.entity.User;
     
    @Component
    @RabbitListener(queues = "topic.second")
    public class TopicSecondReceiver { 
    	
    	@RabbitHandler
        public void process(User user) { 
    		System.out.println("TopicSecondReceiver消费者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
        }
     
    }

启动mq-rabbit-consumer项目，会发现分别接收到各自监听的队列的消息。

## Fanout Exchang ##

在mq-rabbit-provider项目建一个配置类FanoutRabbitConfig.java，配置交换机、队列的绑定关系，代码如下：

package mq.rabbit.config;
     
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.FanoutExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
     
    @Configuration
    public class FanoutRabbitConfig { 
    	
    	@Bean
        public Queue AMessage() { 
            return new Queue("fanout.A");
        }
     
        @Bean
        public Queue BMessage() { 
            return new Queue("fanout.B");
        }
     
        @Bean
        public Queue CMessage() { 
            return new Queue("fanout.C");
        }
     
        @Bean
        FanoutExchange fanoutExchange() { 
            return new FanoutExchange("fanoutExchange");
        }
     
        @Bean
        Binding bindingExchangeA() { 
            return BindingBuilder.bind(AMessage()).to(fanoutExchange());
        }
     
        @Bean
        Binding bindingExchangeB() { 
            return BindingBuilder.bind(BMessage()).to(fanoutExchange());
        }
     
        @Bean
        Binding bindingExchangeC() { 
            return BindingBuilder.bind(CMessage()).to(fanoutExchange());
        }
    }

这里创建三个队列fanout.A、fanout.B、fanout.C，都绑定到FanoutExchange交换机fanoutExchange上。

在SendController类添加一个请求/sendFanout,代码如下：

package mq.rabbit.controller;
     
    import java.util.UUID;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.ResponseBody;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import mq.rabbit.entity.User;
     
    @Controller
    public class SendController { 
     
    	@Autowired
    	private RabbitTemplate template;
    	
    	@GetMapping("/sendFanout")
    	private @ResponseBody String sendFanout(String message) { 
    		User user = new User(UUID.randomUUID().toString(), message, "123456", "sendFanout");
    		template.convertAndSend("fanoutExchange", null, user);
    		return "OK,sendFanout:" + message;
    	}
    	
    }

当调用/sendFanout请求时，在RabbitMQ的web管理界面看到三个队列fanout.A、fanout.B、fanout.C都有一条消息，在Fanout交换机里，如果有设置BindingKey，Fanout交换机会忽略已设置的BindingKey，把消息发送到绑定该交换机的所有队列里。

把mq-rabbit-provider项目里的FanoutRabbitConfig类复制到mq-rabbit-consumer项目，分别创建FanoutReceiverA、FanoutReceiverB和FanoutReceiverC类，代码如下：

package mq.rabbit.receiver;
     
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    import mq.rabbit.entity.User;
     
    @Component
    @RabbitListener(queues = "fanout.A")
    public class FanoutReceiverA { 
     
        @RabbitHandler
        public void process(User user) { 
        	System.out.println("FanoutReceiverA消费者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
        }
     
    }

package mq.rabbit.receiver;
     
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    import mq.rabbit.entity.User;
     
    @Component
    @RabbitListener(queues = "fanout.B")
    public class FanoutReceiverB { 
     
        @RabbitHandler
        public void process(User user) { 
        	System.out.println("FanoutReceiverB消费者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
        }
     
    }

package mq.rabbit.receiver;
     
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    import mq.rabbit.entity.User;
     
    @Component
    @RabbitListener(queues = "fanout.C")
    public class FanoutReceiverC { 
     
        @RabbitHandler
        public void process(User user) { 
        	System.out.println("FanoutReceiverC消费者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
        }
     
    }

上面也可以在一个类里写3个方法来进行对队列的监听，不同的地方在于把@RabbitListener移到方法上即可。

启动mq-rabbit-consumer，即可收到队列的消息。

# RabbitMQ消息的确认机制 #

在使用RabbitMQ的时候，我们可以通过消息持久化操作来解决因为服务器的异常奔溃导致的消息丢失，除此之外我们还会遇到一个问题，当消息的生产者在将消息发送出去之后，消息到底有没有正确到达服务器？如果不进行特殊配置的话，默认情况下发布消息是不会返回任何信息给生产者的，也就是生产者是不知道消息有没有正确到达消息服务器，同理，消息消费者在接收消息后，如果在执行业务逻辑过程出现异常崩溃等情况，会导致消息丢失，所以我们需要对消息的发送和消费进行确认，确保消息能够被正确的存储和消费。RabbitMQ为我们提供了两种方式：**1、事务机制；2、确认机制**。下面介绍消息确认机制。

## 生产者消息确认机制： ##

先把例子跑起来，下面再做详细介绍。在mq-rabbit-provider项目的application.properties文件添加以下属性：

#确认消息已发送到交换机(Exchange)
    spring.rabbitmq.publisher-confirms=true
    #确认消息已发送到队列(Queue)
    spring.rabbitmq.publisher-returns=true

在mq-rabbit-provider项目创建配置类RabbitConfig.java，代码如下：

package mq.rabbit.config;
     
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.support.CorrelationData;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
     
    @Configuration
    public class RabbitConfig { 
    	
    	@Bean
        public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){ 
            RabbitTemplate rabbitTemplate = new RabbitTemplate();
            rabbitTemplate.setConnectionFactory(connectionFactory);
            rabbitTemplate.setMandatory(true);//必须设置为true，才能让下面的ReturnCallback函数生效
     
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { 
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) { 
                    System.out.println("=======ConfirmCallback=========");
                    System.out.println("correlationData = " + correlationData);
                    System.out.println("ack = " + ack);
                    System.out.println("cause = " + cause);
                    System.out.println("=======ConfirmCallback=========");
                }
            });
     
            rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() { 
                @Override
                public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { 
                    System.out.println("--------------ReturnCallback----------------");
                    System.out.println("message = " + message);
                    System.out.println("replyCode = " + replyCode);
                    System.out.println("replyText = " + replyText);
                    System.out.println("exchange = " + exchange);
                    System.out.println("routingKey = " + routingKey);
                    System.out.println("--------------ReturnCallback----------------");
                }
            });
     
            return rabbitTemplate;
        }
     
    }

RabbitMQ生产者是依赖两个回调函数来实现确认的，分别是ConfirmCallback和ConfirmCallback，如上面的代码。按以下4种情况进行回调：

1、消息找不到交换机(Exchange)时回调ConfirmCallback，返回ack=false，代码如下：

=======ConfirmCallback=========
    correlationData = null
    ack = false
    cause = channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'CalonDirectExchange1' in vhost 'calonHost', class-id=60, method-id=40)
    =======ConfirmCallback=========
    2018-08-30 09:59:37.892 ERROR 55704 --- [0.211.55.3:5672] o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'CalonDirectExchange1' in vhost 'calonHost', class-id=60, method-id=40)

2、消息找到交换机(Exchange)但找不到队列(Queue)时回调ConfirmCallback和ReturnCallback，返回ack=true，replyCode = 312，replyText = NO\_ROUTE，代码如下：

--------------ReturnCallback----------------
    message = (Body:'[B@bf8af5b(byte[179])' MessageProperties [headers={ }, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=application/x-java-serialized-object, contentEncoding=null, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=null, receivedExchange=null, receivedRoutingKey=null, receivedDelay=null, deliveryTag=0, messageCount=null, consumerTag=null, consumerQueue=null])
    replyCode = 312
    replyText = NO_ROUTE
    exchange = CalonDirectExchange
    routingKey = CalonDirectRouting1
    --------------ReturnCallback----------------
    =======ConfirmCallback=========
    correlationData = null
    ack = true
    cause = null
    =======ConfirmCallback=========

3、消息既找不到交换机(Exchange)又找不到队列(Queue)时回调ConfirmCallback，返回ack=false，代码如下：

=======ConfirmCallback=========
    correlationData = null
    ack = false
    cause = channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'CalonDirectExchange1' in vhost 'calonHost', class-id=60, method-id=40)
    =======ConfirmCallback=========
    2018-08-30 10:03:22.204 ERROR 55704 --- [0.211.55.3:5672] o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'CalonDirectExchange1' in vhost 'calonHost', class-id=60, method-id=40)

4、消息成功发送回调ConfirmCallback，返回ack=true，代码如下：

=======ConfirmCallback=========
    correlationData = null
    ack = true
    cause = null
    =======ConfirmCallback=========

## 消费者消息确认机制： ##

在mq-rabbit-consumer项目的DirectRabbitConfig配置类进行消息消费确认机制的配置，代码如下：

package mq.rabbit.config;
     
    import org.springframework.amqp.core.AcknowledgeMode;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import mq.rabbit.receiver.DirectAckReceiver;
     
    @Configuration
    public class DirectRabbitConfig { 
    	
    	@Bean
        public Queue CalonDirectQueue() { 
            return new Queue("CalonDirectQueue",true);
        }
    	
    	@Bean
        DirectExchange CalonDirectExchange() { 
            return new DirectExchange("CalonDirectExchange");
        }
    	
    	@Bean
        Binding bindingDirect() { 
            return BindingBuilder.bind(CalonDirectQueue()).to(CalonDirectExchange()).with("CalonDirectRouting");
        }
     
    	@Autowired
    	private CachingConnectionFactory connectionFactory;
    	@Autowired
    	private DirectAckReceiver directAckReceiver;//消息接收处理类
     
    	@Bean
    	public SimpleMessageListenerContainer simpleMessageListenerContainer() { 
    		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
    		container.setConcurrentConsumers(1);
    		container.setMaxConcurrentConsumers(1);
    		container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息
     
    		container.setQueues(CalonDirectQueue());
    		container.setMessageListener(directAckReceiver);
    		return container;
    	}
     
    }

在mq-rabbit-consumer项目新建消息监听类DirectAckReceiver.java，用于处理消息的确认操作，代码如下：

package mq.rabbit.receiver;
     
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.rabbitmq.client.Channel;
    import mq.rabbit.entity.User;
     
    @Component
    public class DirectAckReceiver implements ChannelAwareMessageListener { 
     
    	@Autowired
    	private ObjectMapper objectMapper;
     
    	@Override
    	public void onMessage(Message message, Channel channel) throws Exception { 
    		long deliveryTag = message.getMessageProperties().getDeliveryTag();
    		try { 
    			byte[] body = message.getBody();
    			User user = objectMapper.readValue(body, User.class);
    			System.out.println("DirectAckReceiver消费者收到消息 : " + user.getId()+","+user.getUsername()+","+user.getPassword()+","+user.getType());
    			channel.basicAck(deliveryTag, true);
    // channel.basicReject(deliveryTag, true);//为true会重新放回队列
    		} catch (Exception e) { 
    			channel.basicReject(deliveryTag, false);
    			e.printStackTrace();
    		}
    	}
     
    }

消息接收确认模式有以下3种：

*  AcknowledgeMode.NONE：不确认
 *  AcknowledgeMode.AUTO：自动确认
 *  AcknowledgeMode.MANUAL：手动确认

默认情况下是自动确认，如果消费端消费逻辑抛出异常，也就是消费端没有处理成功这条消息，那么就相当于丢失了消息，在实际应用中，我们希望每条消息都能够被正确消费而不是出现丢失的情况，上面代码是开启手动确认模式，下面看看手动确认都有哪几种方式：

*  成功确认：void basicAck(long deliveryTag, boolean multiple) throws IOException;  
    deliveryTag:该消息的index  
    multiple：是否批量. true：将一次性ack所有小于deliveryTag的消息。  
    消费者成功处理后，调用channel.basicAck(message.getMessageProperties().getDeliveryTag(), false)方法对消息进行确认。
 *  失败确认：void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException;  
    deliveryTag:该消息的index。  
    multiple：是否批量. true：将一次性拒绝所有小于deliveryTag的消息。  
    requeue：是否重新入队列。
 *  拒绝确认：void basicReject(long deliveryTag, boolean requeue) throws IOException;  
    deliveryTag:该消息的index。  
    requeue：被拒绝的是否重新入队列。  
    channel.basicNack 与 channel.basicReject 的区别在于basicNack可以批量拒绝多条消息，而basicReject一次只能拒绝一条消息。

**这里要注意一点的是，无论如何，必须对消息进行确认操作，如果不调用相关函数进行确认，则RabbitMQ会认为该程序处理能力弱，不会再发送消息到该监听程序。**

--------------------

[代码参考-mq-rabbit-provider.git][-mq-rabbit-provider.git]

[代码参考-mq-rabbit-consumer.git][-mq-rabbit-consumer.git]

[全文参考][Link 1]

[RabbitMQ]: http://www.rabbitmq.com/download.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIwNTE0OA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/201912161543109.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIwNTE0OA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIwNTE0OA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191216154544599.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIwNTE0OA==,size_16,color_FFFFFF,t_70
[20191216155014812.jpg]: https://img-blog.csdnimg.cn/20191216155014812.jpg
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIwNTE0OA_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20191216155848258.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTIwNTE0OA==,size_16,color_FFFFFF,t_70
[-mq-rabbit-provider.git]: https://github.com/fm1995/mq-rabbit-consumer
[-mq-rabbit-consumer.git]: https://github.com/fm1995/mq-rabbit-provider
[Link 1]: https://blog.csdn.net/kevinmcy/article/details/82221297