一、概述
二、安装
- 1. 通过docker安装
三、整合
- 3.1 SpringBoot整合
四、可靠投递
- 4.1 发送端确认
  - 4.1.1 确认回调
    - 1. 开启发送端消息抵达Broker确认
    - 2. 定制RabbitTemplate
  - 4.1.2 返回回调
    - 1. 开启发送端消息抵达队列确认
    - 2. 定制RabbitTemplate
- 4.2 消费端确认
  - 4.2.1 几种操作方式

一、概述

RabbitMQ是一个由erlang语言开发的AMQP(Advanved Message Queue Protocol)的开源实现。

1.1 核心概念

1.1.1 消息（Message）

消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。

1.1.2 生产者（Publisher）

消息的生产者，也是一个向交换器发布消息的客户端应用程序。

1.1.3 交换机（Exchange）

交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。 Exchange有4种类型：direct(默认)，fanout, topic, 和headers，不同类型的Exchange转发消息的策略有所区别。

1.1.4 消息队列（Queue）

消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

1.1.5 绑定（Binding）

绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。 Exchange 和Queue的绑定可以是多对多的关系。

1.1.6 连接（Connection）

网络连接，比如一个TCP连接。

1.1.7 信道（Channel）

信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。

1.1.8 消费者（Consumer）

消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。

1.1.9 虚拟主机（Virtual Host）

虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。
每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是/。

1.1.10 消息代理（Broker）

表示消息队列服务器实体。

1.2 工作流程梳理

AMQP 中消息的路由过程和 Java 开发者熟悉的 JMS 存在一些差别， AMQP 中增加了 Exchange 和 Binding 的角色。
生产者把消息发布到 Exchange 上，消息最终到达队列并被消费者接收，而 Binding 决定交换器的消息应该发送到那个队列。

1.3 交换机类型

Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct、 fanout、topic、headers 。
headers 匹配 AMQP 消息的 header 而不是路由键， headers 交换器和 direct 交换器完全一致，但性能差很多，目前几乎用不到了，所以直接看另外三种类型：

1.3.1 Direct Exchange

消息中的路由键（routing key）如果和 Binding 中的 binding key 一致，交换器就将消息发到对应的队列中。
路由键与队列名完全匹配，如果一个队列绑定到交换机要求路由键为“dog”，则只转发 routing key 标记为“dog”的消息，不会转发 “dog.puppy”，也不会转发“dog.guard” 等等。它是完全匹配、单播的模式。

1.3.2 Fanout Exchange

每个发到 fanout 类型交换器的消息都会分到所有绑定的队列上去。fanout 交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。
fanout 类型转发消息是最快的。

1.3.3 Topic Exchange

topic 交换器通过模式匹配分配消息的路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开。
它同样也会识别两个通配符：符号#和*。#匹配0个或多个单词，*匹配一个单词。

二、安装

1. 通过docker安装

1. 拉取镜像

docker pull rabbitmq:3.8.2-management

2. 创建挂载目录

mkdir /opt/docker/rabbitmq

3. 创建容器

docker run \
--name rabbit382 \
-p 5671:5671 \
-p 5672:5672 \
-p 4369:4369 \
-p 25672:25672 \
-p 15671:15671 \
-p 15672:15672 \
--mount type=bind,src=/opt/docker/rabbitmq,dst=/var/lib/rabbitmq \
--hostname rabbitOne \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=你的密码 \
-d rabbitmq:3.8.2-management

其中：

4369, 25672 (Erlang发现&集群端口)
5672, 5671 (AMQP端口)
15672 (web管理后台端口)
61613, 61614 (STOMP协议端口)
1883, 8883 (MQTT协议端口)

三、整合

3.1 SpringBoot整合

3.1.1 引入依赖

引入如下依赖，RabbitAutoConfiguration就会自动生效。

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

会给容器中自动配置下列Bean对象：

RabbitTemplate
AmqpAdmin
CachingConnectionFactory
RabbitMessagingTemplate

3.1.2 添加配置，开启支持

在配置文件中添加一下关于RabbitMQ的基础配置信息：

# RabbitMQ配置
spring.rabbitmq.host=
spring.rabbitmq.port=5672
spring.rabbitmq.username=
spring.rabbitmq.password=
# 虚拟主机配置
spring.rabbitmq.virtual-host=/

使用注解@EnableRabbit开启SprigBoot对RabbitMQ的支持，此注解在使用队列监听器注解@RabbitListener时，是必须加的。若是单纯的创建交换机、队列等操作则无需添加此注解。

3.1.3 创建相关组件

1. 注入AmqpAdmin对象

使用AmqpAdmin对象，可以用来做创建交换机、队列、绑定等操作。

@Autowired
private AmqpAdmin amqpAdmin;

2. 创建交换机

这里创建直接型交换机，存在参数如下：

String name：交换机名称
boolean durable：是否持久化
boolean autoDelete：是否自动删除

Map arguments：参数

/**
* 1、如何创建Exchange、Queue、Binding
*      1）、使用AmqpAdmin进行创建
* 2、如何收发消息
*/
public void createExchange() {
Exchange directExchange = new DirectExchange("hello-java-exchange",true,false);
amqpAdmin.declareExchange(directExchange);
log.info("Exchange[{}]创建成功：","hello-java-exchange");
}

3. 创建队列

创建队列时，可以传入以下参数：

String name：队列名称
boolean durable：是否持久化
boolean exclusive：是否排他
boolean autoDelete：是否自动删除

Map arguments：参数

@Test
public void testCreateQueue() {
Queue queue = new Queue("hello-java-queue",true,false,false);
amqpAdmin.declareQueue(queue);
log.info("Queue[{}]创建成功：","hello-java-queue");
}

4. 创建绑定

创建绑定时，可以传入一下参数：

String destination：目的地，传入目的地名称
DestinationType destinationType：目的地类型，如交换机、队列
String exchange：做绑定的交换机名称
String routingKey：路由键

Map arguments：参数

public void createBinding() {
Binding binding = new Binding("hello-java-queue",
       Binding.DestinationType.QUEUE,
       "hello-java-exchange",
       "hello.java",
       null);
amqpAdmin.declareBinding(binding);
log.info("Binding[{}]创建成功：","hello-java-binding");
}

3.1.4 发送及接收消息

1. 注入RabbitTemplate对象

@Autowired
private RabbitTemplate rabbitTemplate;

2. 发送消息

public void sendMessageTest() {
OrderReturnReasonEntity reasonEntity = new OrderReturnReasonEntity();
reasonEntity.setId(1L);
reasonEntity.setCreateTime(new Date());
reasonEntity.setName("reason");
reasonEntity.setStatus(1);
reasonEntity.setSort(2);
String msg = "Hello World";
//1、发送消息,如果发送的消息是个对象，会使用序列化机制，将对象写出去，对象必须实现Serializable接口
//2、发送的对象类型的消息，可以是一个json
rabbitTemplate.convertAndSend("hello-java-exchange","hello2.java",
       reasonEntity,new CorrelationData(UUID.randomUUID().toString()));
log.info("消息发送完成:{}",reasonEntity);
}

默认情况下，RabbitMQ会通过Java的序列化将对象进行序列化，再进行Base64编码发送给队列，这时对象需要实现Serializable接口。

另外，我们也可以给RabbitMQ配置专门的序列化工具来对对象进行序列化，转换成json对象，进行发送。

序列化工具配置方式如下：

@Configuration
public class MyRabbitConfig {
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

最终发送到队列中的消息如下：

3. 接收消息

1）方式一

通过方法上的@RabbitListener注解，指定监听的队列，一旦队列中有新的消息到达，就会获取到。

/**
 * 注解参数：
 *     queues：声明需要监听的队列
 *
 * 方法参数：
 *    Message message：原生消息实体，包含消息头 + 消息体
 *  T<发送的消息的类型> OrderReturnReasonEntity content：发送到队列中的对象消息对应的类型，可以便捷的反序列化
 *  Channel channel：当前传输数据的通道
 */
@RabbitListener(queues = {"hello-java-queue"})
public void revieveMessage(Message message,
                           OrderReturnReasonEntity content) {
    //拿到主体内容
    byte[] body = message.getBody();
    //拿到的消息头属性信息
    MessageProperties messageProperties = message.getMessageProperties();
    System.out.println("接受到的消息...内容" + message + "===内容：" + content);
}

注意：Queue可以很多人都来监听，只要收到消息，队列删除消息，而且只能有一个收到消息。

场景：

订单服务启动多个；同一个消息，只能有一个客户端收到。
只有一个消息完全处理完，方法运行结束，才可以接收到下一个消息。

2）方式二

在类上使用@RabbitListener注解，指定监听的队列，然后在具体的方法上使用@RabbitHandler注解，通过多个方法重载可以同时使用@RabbitHandler注解，接受不同类型的消息，便捷处理。

@Service("orderItemService")
@RabbitListener(queues = {"hello-java-queue"})
public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {
    /**
     * queues：声明需要监听的队列
     * channel：当前传输数据的通道
     */
    @RabbitHandler
    public void revieveMessage(Message message,
                               OrderReturnReasonEntity content) {
        //拿到主体内容
        byte[] body = message.getBody();
        //拿到的消息头属性信息
        MessageProperties messageProperties = message.getMessageProperties();
        System.out.println("接受到的消息...内容" + message + "===内容：" + content);
    }
    @RabbitHandler
    public void revieveMessage2(OrderEntity content) {
        System.out.println("接受到的消息...");
    }
}

四、可靠投递

在消息投递及接收环节，可能会出现业务失败、网络抖动等情况，导致消息丢失。为了保证消息可靠投递，不会出现丢失的情况，我们可以使用事务消息的方式来实现，但这种方式会使消息中间件的性能下降250倍左右，为此，引入了消息的确认机制，来实现高性能并保证消息可靠投递。

可以看到，消息投递的过程总体上可以分为三个阶段：

确认回调（confirmCallback）：消息的发送者（publisher）到消息代理（broker）之间的一次确认，是否成功发送到mq。比如网络抖动导致发送失败。
返回回调（returnCallback）：交换机（Exchange）到队列（Queue）之间的一次确认，是交换机是否将消息分发到对应的队列。比如路由键错误导致分发失败。
ack机制：队列（Queue）与消费者（Consumer）之间的一次确认，是消费者是否成功接收消息并消费的确认。比如网络抖动、业务执行失败等导致的消费失败。

前两种属于发送端的确认，后一种属于接收端的确认。

4.1 发送端确认

4.1.1 确认回调

在RabbitTemplate中有成员private ConfirmCallback confirmCallback;，可以配置确认回调方法内容。


    /**
     * A callback for publisher confirmations.
     *
     */
    @FunctionalInterface
    public interface ConfirmCallback {
        /**
         * Confirmation callback.
         * @param correlationData correlation data for the callback.    消息的关联数据，内含消息的唯一id
         * @param ack true for ack, false for nack    确认标识，正确收到为true，没有正确收到为false
         * @param cause An optional cause, for nack, when available, otherwise null. 没有正确收到时，返回原因
         */
        void confirm(@Nullable CorrelationData correlationData, boolean ack, @Nullable String cause);
    }

其中，CorrelationData可以在发送消息时做指定，其中可以存入一个唯一的id来做消息区分：

public class CorrelationData implements Correlation {
    private final SettableListenableFuture<Confirm> future = new SettableListenableFuture<>();
    /**
     *    可以给此字段存入唯一标识，做区分
     *
     */
    @Nullable
    private volatile String id;
    private volatile Message returnedMessage;
    ...
}

1. 开启发送端消息抵达Broker确认

若是非自动化配置，使用连接工厂的话，可以在创建 connectionFactory 的时候设置 PublisherConfirms(true) 选项，开启confirmcallback 。
自动化配置情况下，可以使用如下配置方式：

# 开启发送端消息抵达Broker确认
spring.rabbitmq.publisher-confirms=true

2. 定制RabbitTemplate

其实就是为RabbitTemplate配置专门的确认回调方法，执行指定的回调逻辑。

@Configuration
public class MyRabbitConfig {
    private RabbitTemplate rabbitTemplate;
    /**
    * 定制RabbitTemplate
    * 1、服务收到消息就会回调
    *      1、spring.rabbitmq.publisher-confirms: true
    *      2、设置确认回调
    * 2、消息正确抵达队列就会进行回调
    *      1、spring.rabbitmq.publisher-returns: true
    *         spring.rabbitmq.template.mandatory: true
    *      2、设置确认回调ReturnCallback
    *
    * 3、消费端确认(保证每个消息都被正确消费，此时才可以broker删除这个消息)
    *
    */
    @PostConstruct  //MyRabbitConfig对象创建完成以后，执行这个方法
    public void initRabbitTemplate() {
        /**
        * 1、只要消息抵达Broker就ack=true
        * correlationData：当前消息的唯一关联数据(这个是消息的唯一id)
        * ack：消息是否成功收到
        * cause：失败的原因
        */
        //设置确认回调
        rabbitTemplate.setConfirmCallback((correlationData,ack,cause) -> {
            System.out.println("confirm...correlationData["+correlationData+"]==>ack:["+ack+"]==>cause:["+cause+"]");
        });
    }
}

注意：
  - 回调函数中的ack，与消息是否被消费是无关的，只要消息正确抵达Broker就会返回true。
  - 消息只要被 broker 接收到就会执行 confirmCallback，如果是 cluster 模式，需要所有broker 接收到才会调用 confirmCallback。
  - 被 broker 接收到只能表示 message 已经到达服务器，并不能保证消息一定会被投递到目标 queue 里。所以需要用到接下来的 returnCallback 。

4.1.2 返回回调

在RabbitTemplate中，同样有成员private ReturnCallback returnCallback;，此是返回回调的调用逻辑。

/**
* A callback for returned messages.
*
*/
@FunctionalInterface
public interface ReturnCallback {
    /**
    * Returned message callback.
    * @param message the returned message.    消息原始数据
    * @param replyCode the reply code.    返回状态码
    * @param replyText the reply text.    返回状态文本
    * @param exchange the exchange.    来源交换机
    * @param routingKey the routing key.    路由键
    */
    void returnedMessage(Message message, int replyCode, String replyText,
                         String exchange, String routingKey);
}

confrim 模式只能保证消息到达 broker，不能保证消息准确投递到目标 queue 里。
在有些业务场景下，我们需要保证消息一定要投递到目标 queue 里，此时就需要用到return 退回模式。这样如果未能投递到目标 queue 里将调用 returnCallback ，可以记录下详细到投递数据，定期的巡检或者自动纠错都需要这些数据。

1. 开启发送端消息抵达队列确认

# 开启发送端消息抵达Queue确认
spring.rabbitmq.publisher-returns=true
# 只要消息抵达Queue，就会异步发送优先回调returnconfirm
spring.rabbitmq.template.mandatory=true

2. 定制RabbitTemplate

同样是配置指定的返回回调逻辑。

@Configuration
public class MyRabbitConfig {
    private RabbitTemplate rabbitTemplate;
    /**
     * 定制RabbitTemplate
     * 1、服务收到消息就会回调
     *      1、spring.rabbitmq.publisher-confirms: true
     *      2、设置确认回调
     * 2、消息正确抵达队列就会进行回调
     *      1、spring.rabbitmq.publisher-returns: true
     *         spring.rabbitmq.template.mandatory: true
     *      2、设置确认回调ReturnCallback
     *
     * 3、消费端确认(保证每个消息都被正确消费，此时才可以broker删除这个消息)
     *
     */
    @PostConstruct  //MyRabbitConfig对象创建完成以后，执行这个方法
    public void initRabbitTemplate() {
        /**
         * 只要消息没有投递给指定的队列，就触发这个失败回调
         * message：投递失败的消息详细信息
         * replyCode：回复的状态码
         * replyText：回复的文本内容
         * exchange：当时这个消息发给哪个交换机
         * routingKey：当时这个消息用哪个路邮键
         */
        rabbitTemplate.setReturnCallback((message,replyCode,replyText,exchange,routingKey) -> {
            System.out.println("Fail Message["+message+"]==>replyCode["+replyCode+"]" +
                    "==>replyText["+replyText+"]==>exchange["+exchange+"]==>routingKey["+routingKey+"]");
        });
    }
}

注意：只有当消息没有成功抵达队列时，此返回回调的逻辑才会执行。当消息正确到达队列时，此回调逻辑不会执行。

4.2 消费端确认

消费者获取到消息，成功处理，可以回复Ack给Broker：

basic.ack用于肯定确认；broker将移除此消息
basic.nack用于否定确认；可以指定broker是否丢弃此消息，可以批量
basic.reject用于否定确认；同上，但不能批量

默认是自动ack，消息被消费者收到之后，就会从broker的queue中移除。

queue若没有对应的消费者，消息依然会被存储，直到消费者将其消费掉。

消费者收到消息，默认会自动ack。但是如果无法确定此消息是否被处理完成，或者成功处理。我们可以开启手动ack模式：

消息处理成功，ack()，接受下一个消息，此消息broker就会移除
消息处理失败，nack()/reject()，重新发送给其他人进行处理，或者容错处理后ack
消息一直没有调用ack/nack方法，broker认为此消息正在被处理，不会投递给别人，此时客户端断开，消息不会被broker移除，会投递给别人

4.2.1 几种操作方式

首先是需要配置手动ack消息：

# 手动ack消息，不使用默认的消费端确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual

1. 确认消息消费成功


/**
 * Acknowledge one or several received
 * messages. Supply the deliveryTag from the {@link com.rabbitmq.client.AMQP.Basic.GetOk}
 * or {@link com.rabbitmq.client.AMQP.Basic.Deliver} method
 * containing the received message being acknowledged.
 * @see com.rabbitmq.client.AMQP.Basic.Ack
 * @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver} // 投递标记，唯一标识消息
 * @param multiple true to acknowledge all messages up to and    // 是否批量操作
 * including the supplied delivery tag; false to acknowledge just
 * the supplied delivery tag.
 * @throws java.io.IOException if an error is encountered
 */
void basicAck(long deliveryTag, boolean multiple) throws IOException;

2. 拒收消息（可批量操作）


    /**
     * Reject one or several received messages.
     *
     * Supply the <code>deliveryTag</code> from the {@link com.rabbitmq.client.AMQP.Basic.GetOk}
     * or {@link com.rabbitmq.client.AMQP.Basic.GetOk} method containing the message to be rejected.
     * @see com.rabbitmq.client.AMQP.Basic.Nack
     * @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver}    // 投递标记，表示确认哪条消息
     * @param multiple true to reject all messages up to and including    // 是否批量操作
     * the supplied delivery tag; false to reject just the supplied
     * delivery tag.
     * @param requeue true if the rejected message(s) should be requeued rather    // 是否重新入队
     * than discarded/dead-lettered
     * @throws java.io.IOException if an error is encountered
     */
    void basicNack(long deliveryTag, boolean multiple, boolean requeue)
            throws IOException;

3. 拒收消息（不可批量操作）


    /**
     * Reject a message. Supply the deliveryTag from the {@link com.rabbitmq.client.AMQP.Basic.GetOk}
     * or {@link com.rabbitmq.client.AMQP.Basic.Deliver} method
     * containing the received message being rejected.
     * @see com.rabbitmq.client.AMQP.Basic.Reject
     * @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver}    // 投递标记
     * @param requeue true if the rejected message should be requeued rather than discarded/dead-lettered    // 是否重新入队
     * @throws java.io.IOException if an error is encountered
     */
    void basicReject(long deliveryTag, boolean requeue) throws IOException;

4. 实践

可以通过如下方式，指定消息消费完成后的ack操作：

@RabbitHandler
public void revieveMessage3(Message message, OrderReturnReasonEntity content, Channel channel) {
    byte[] body = message.getBody();
    MessageProperties messageProperties = message.getMessageProperties();
    // channel内按顺序自增，投递标记
    long deliveryTag = messageProperties.getDeliveryTag();
    try {
        if (deliveryTag % 2 == 0) {
            /**
             *  deliveryTag：投递标记，可以指定消费的那条消息
             *  multiple：是否批量操作
             */
            channel.basicAck(deliveryTag, false);
        } else {
            /**
             *  deliveryTag：投递标记，可以指定消费的那条消息
             *  multiple：是否批量操作
             *  requeue：是否重新入队
             */
            channel.basicNack(deliveryTag, false, true);
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
}