每条消息都需要在消息头中指定route-key

Exchange：交换器，每一个交换器都连接一个队列（可以看作局域网中的交换机的端口，每一个端口都连接一台电脑）

Queue队列：用于存储生产者发送的消息

交换器与Queue之间存在绑定关系，一个交换器可以绑定多个队列，存在多种绑定关系
根据消息头中的路由键指定的绑定关系发送到匹配的队列中

网络连接
  生产者与Broker、消费者与Broker通过连接传输消息
  一个客户端只会建立一条连接

一个客户端建立一条连接，一条连接内存储多个通道用于监听不同队列

只需要安装一个rabbitmq，但是可以分离出多个微主机，互相之间配置隔离，使用不同的url访问
1.例如dev、test可以使用不同VHost
2.不同的系统使用不同的VHost

简介：
  1.可靠抵达：消息可靠抵达MQ，消息可靠抵达消费者
  2.事务消息：性能下降250倍，所以采用确认机制代替
文档：https://www.rabbitmq.com/confirms.html#publisher-confirms
Reliable Delivery=》Acknowledgements and Confirms=>Publisher confirms

简介：
  1.生产者发送消息到Queue会经过两个两个过程【确认机制看做一种协议】
    1）消息从publisher到达Broker（到达后会回调confirmCallback，消费者被告知消息是否抵达服务器）
      【集群情况下必须所有的broker接收到才会调用confirmCallback】
    2）消息从Exchange投递到Queue（失败后会回调returnCallback，消费者被告知消息是否抵达Queue）

# 开启发送端确认：
spring.rabbitmq.publisher-returns=true
# 消息在没有被队列接收时是否强行退回
spring.rabbitmq.template.mandatory=true

// 测试步骤，调用单元测试中发送消息的方法，触发回调
@Configuration
public class MyRabbitConfig {
    @Autowired
    RabbitTemplate rabbitTemplate;
    @Bean
    public MessageConverter messageConverter() {
        // 使用json序列化器来序列化消息，发送消息时，消息对象会被序列化成json格式
        return new Jackson2JsonMessageConverter();
    }
    /**
     * 定制RabbitTemplate
     * 1、服务收到消息就会回调
     * 1、spring.rabbitmq.publisher-confirms: true
     * 2、设置确认回调
     * 2、消息正确抵达队列就会进行回调
     * 1、spring.rabbitmq.publisher-returns: true
     * spring.rabbitmq.template.mandatory: true
     * 2、设置确认回调ReturnCallback
     * <p>
     * 3、消费端确认(保证每个消息都被正确消费，此时才可以broker删除这个消息)
     */
    @PostConstruct   // (MyRabbitConfig对象创建完成以后，执行这个方法)
    public void initRabbitTemplate() {
        /**
         * 发送消息触发confirmCallback回调
         * @param correlationData：当前消息的唯一关联数据（如果发送消息时未指定此值，则回调时返回null）
         * @param ack：消息是否成功收到（ack=true，消息抵达Broker）
         * @param cause：失败的原因
         */
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            System.out.println("发送消息触发confirmCallback回调" +
                    "\ncorrelationData ===> " + correlationData +
                    "\nack ===> " + ack + "" +
                    "\ncause ===> " + cause);
        });
    }
}

简介：
  1.生产者发送消息到Queue会经过两个两个过程【确认机制看做一种协议】
    1）消息从publisher到达Broker（到达后会回调confirmCallback，消费者被告知消息是否抵达服务器）
    2）消息从Exchange投递到Queue（失败后会回调returnCallback，消费者被告知消息是否抵达Queue）

# 开启发送端抵达队列确认
spring.rabbitmq.publisher-returns=true

@Configuration
public class MyRabbitConfig {
    @Autowired
    RabbitTemplate rabbitTemplate;
    @Bean
    public MessageConverter messageConverter() {
        // 使用json序列化器来序列化消息，发送消息时，消息对象会被序列化成json格式
        return new Jackson2JsonMessageConverter();
    }
    /**
     * 定制RabbitTemplate
     * 1、服务收到消息就会回调
     * 1、spring.rabbitmq.publisher-confirms: true
     * 2、设置确认回调
     * 2、消息正确抵达队列就会进行回调
     * 1、spring.rabbitmq.publisher-returns: true
     * spring.rabbitmq.template.mandatory: true
     * 2、设置确认回调ReturnCallback
     * <p>
     * 3、消费端确认(保证每个消息都被正确消费，此时才可以broker删除这个消息)
     */
    @PostConstruct   // (MyRabbitConfig对象创建完成以后，执行这个方法)
    public void initRabbitTemplate() {
        /**
         * 消息未到达队列触发returnCallback回调
         * 只要消息没有投递给指定的队列，就触发这个失败回调
         * @param message：投递失败的消息详细信息
         * @param replyCode：回复的状态码
         * @param replyText：回复的文本内容
         * @param exchange：接收消息的交换机
         * @param routingKey：接收消息的路由键
         */
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            // 需要修改数据库 消息的状态【后期定期重发消息】
            System.out.println("消息未到达队列触发returnCallback回调" +
                    "\nmessage ===> " + message +
                    "\nreplyCode ===> " + replyCode + 
                    "\nreplyText ===> " + replyText +
                    "\nexchange ===> " + exchange + 
                    "\nroutingKey ===> " + routingKey);
        });
    }
}

简介：
  1.消费者接收消息也会经过ack消息确认机制，只有消费者成功接收消息，broker才允许删除消息
  2.默认情况下消息抵达客户端后自动确认，服务端消息自动删除
  3.手动确认模式下，消费者接收消息后但是不执行ack/nack进行确认，服务端队列中的消息会从unacked状态变为ready状态等待下一次消费（即使consumer宕机消息也不会丢失）
  4.注意：消息到达消费端，消息会进入unacked状态，如果手动确认模式下，消费者因宕机而未执行ack/nack，消息会一直处于unacked状态不会被删除（持久化了）
  5.签收，ack

# 消费者手动确认模式，关闭自动确认，否则会消息丢失
spring.rabbitmq.listener.simple.acknowledge-mode=manual

@RabbitListener(queues = {"hello-java-queue"})
@Service("orderItemService")
public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {
  @RabbitHandler
    public void revieveMessage(Message message, OrderEntity entity, Channel channel) {
        // 请求体，序列化存储（本例中已使用Jackson2JsonMessageConverter序列化器作JSON序列化存储）
        byte[] body = message.getBody();
        // 请求头
        MessageProperties properties = message.getMessageProperties();
        // channel内按顺序自增的long类型消息标签
        long deliveryTag = properties.getDeliveryTag();
        // JSON反序列得到消息内容对象
        OrderEntity reason = JSONObject.parseObject(body, OrderEntity.class);
        System.out.println("接受到的消息对象" + message);
        System.out.println("接受到的消息内容" + reason);
        System.out.println("接受到的消息内容" + entity);
        try {
            if (deliveryTag == 2) {
                // 手动确认，消息会从unacked中删除，total数量减1
                // boolean multiple：是否批量签收
                channel.basicAck(deliveryTag, false);
            } else {
                // 手动拒签
                // boolean multiple：是否批量拒签
                // boolean requeue：当前拒签消息是否发回服务器重新入队
                channel.basicNack(deliveryTag, false, true);
            }
        } catch (IOException e) {
            // 网络中断
            e.printStackTrace();
        }
    }
}

场景：
  未付款订单，超时后，系统自动取消订单并释放占有物品
常用解决方案：
  Spring的Schedule定时任务轮询数据库
缺点：
  消耗系统内存、增加数据库的压力、存在较大的时间误差
解决：
  在保证可靠消息的前提下，使用延时队列+死信队列，达到最终一致性（柔性事务）

死信Exchange其实就是普通的Exchange
队列设置好自己的Dead Letter Exchange，当此队列中的消息过期后会被转发到这个路由，被称为死信路由

TTL消息 + 死信Exchange
使用一个队列接收死信Exchange中的TTL消息，这样的队列被称为延时队列
注意：存放TTL消息的队列不要让客户端监听（这个队列和延时队列不是同一个，延时队列是存储已经超时的TTL消息）

arguments.put("x-dead-letter-exchange", "order-event-exchange");// 死信路由
arguments.put("x-dead-letter-routing-key", "order.release.order");// 死信路由键
arguments.put("x-message-ttl", 60000); // 消息过期时间 1分钟

场景：
  配合延时队列+死信队列实现最终一致性（柔性事务）

CREATE TABLE `mq_message` (
  `message_id` char(32) NOT NULL,
  `content` json,
  `to_exchane` varchar(255) DEFAULT NULL,
  `routing_key` varchar(255) DEFAULT NULL,
  `class_type` varchar(255) DEFAULT NULL,
  `message_status` int(1) DEFAULT '0' COMMENT '0-新建 1-已发送 2-错误抵达 3-已抵达',
  `create_time` datetime DEFAULT NULL,
  `update_time` datetime DEFAULT NULL,
  PRIMARY KEY (`message_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

前提：
    使用定时器扫描mq_message定时重发
情况1：网络连接失败，消息未抵达Broker
    解决：发送消息时同时将消息持久化到MQ中并设定状态为已抵达
         当出现异常时在catch处修改消息状态为错误抵达
情况2：消息抵达Broker，但为抵达queue，消息会丢失（只有抵达了queue消息才会持久化）
  解决：开启生产者确认机制，将触发returnCallback的returnedMessage的消息状态修改为错误抵达
情况3：消费者未ack时宕机，导致消息丢失
  解决：开启消费者手动ack

情况1：业务逻辑已经执行，但是ack时宕机，消息由unack变为ready，消息重新入队
  解决：将接口设计成幂等性，例如库存解锁时判断工作单的状态，已解锁则无操作
  解决2：防重表

情况1：生产者流量太大
  解决：减慢发送消息速率（验证码、防刷、重定向、削峰）
  
情况2：消费者能力不足或宕机
  解决：上线更多消费者
  解决2：上线专门的队列消费服务，批量取出消息入库，离线处理业务慢慢处理

可以添加一个消息服务，各模块调用发送消息API即可
实现消息存库+异常修改状态
思考：如果feign调用失败没有问题，做好本地事务，feign调用失败回滚即可

场景：
  高并发秒杀模块，将秒杀成功创建订单的消息发送给MQ，订单模块按照自己的能力消费生成订单
  这个过程就是队列削峰（不走购物车逻辑，否则秒杀的高并发流量会带给订单模块）

1.订单服务中引入依赖，场景启动器，引入后RabbitAutoConfiguration自动生效
<!--rabbitmq-->
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2.RabbitAutoConfiguration生效后自动注入多个容器：
  CachingConnectionFactory：
  RabbitTemplate：
  AmqpAdmin：
  RabbitMessagingTemplate：
3.RabbitProperties：配置类

用于创建Exchange、Queue、Binding

启动类标注：（只有需要用到监听消息时才需要该注解，开启后可以使用@RabbitListener）
@EnableRabbit

spring:
  rabbitmq:
    host: 192.168.56.10
    port: 5672
    # 虚拟主机
    virtual-host: /
    # 开启发送端抵达队列确认【发送端确认机制+本地事务表】
    publisher-returns: true
    # 开启发送确认【发送端确认机制+本地事务表】
    publisher-confirm-type: correlated
    # 只要抵达队列，优先回调return confirm
    template:
      mandatory: true
    # 使用手动确认模式，关闭自动确认【消息丢失】
    listener:
      simple:
        acknowledge-mode: manual

@Configuration
public class MyRabbitConfig {
    @Bean
    public MessageConverter messageConverter() {
        // 使用json序列化器来序列化消息，发送消息时，消息对象会被序列化成json格式
        return new Jackson2JsonMessageConverter();
    }
}

@Autowired
AmqpAdmin amqpAdmin;
@Test
void createExchange() {
    // 创建交换机
    // String name, boolean durable, boolean autoDelete
    DirectExchange exchange = new DirectExchange("hello-java-exchange", true, false);
    amqpAdmin.declareExchange(exchange);
    log.info("Exchange创建[{}]成功", "hello-java-exchange");
}

@Autowired
AmqpAdmin amqpAdmin;
@Test
void createQueue() {
    // 创建队列
    // String name, boolean durable, boolean exclusive, boolean autoDelete
    // exclusive：是否排他，true：只有一个连接可以使用此队列，其他连接无法连上此队列
    Queue queue = new Queue("hello-java-queue", true, false, false);
    amqpAdmin.declareQueue(queue);
    log.info("Queue创建[{}]成功", "hello-java-queue");
}

@Autowired
AmqpAdmin amqpAdmin;
@Test
void createBinding() {
    // 创建绑定，交换机绑定目的地
    // String destination：目的地name
    // DestinationType destinationType：目的地类型【queue或exchange（路由）】
    // String exchange：待绑定交换机
    // String routingKey：路由键
    Binding bind = new Binding("hello-java-queue", Binding.DestinationType.QUEUE,"hello-java-exchange", "hello.java", null)
    amqpAdmin.declareBinding(bind);
    log.info("Binding创建[{}]成功", "hello-java-binding");
}

@Autowired
RabbitTemplate rabbitTemplate;
@Test
void sendMsg() {
    // 如果发送的消息是个对象，会使用序列化机制，将对象写出去。对象类必须实现 serializable
    // 如果使用JSON序列化器，则不需要类实现Serializable
    // String exchange：交换机
    // String routingKey：路由键
    // final Object object：消息
    // CorrelationData correlationData：可指定消息ID
    
    // 消息对象，可以是任意类型，类必须实现serializable，消息会以序列化的方式写入流中
    OrderReturnReasonEntity message = new OrderReturnReasonEntity();// 退货原因
    message.setId(1L);
    message.setCreateTime(new Date());
    message.setName("哈哈");
    // 消息ID
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java", message, correlationData);
}

简介：
  1.用于标注在监听类或监听方法上，接收消息，需要指定监听的队列（数组）
  2.使用该注解之前，需要在启动类加上该注解：@EnableRabbit
  3.@RabbitListener即可以标注在方法上又可以标注在类上
    标注在类上：表示该类是监听类，使得@RabbitHandler注解生效
    标注在方法上：表示该方法时监听方法，会监听指定队列获得消息
  4.一般只标注在方法上，并配合@RabbitHandler使用，重载的方式接收不同消息对象

import com.rabbitmq.client.Channel;
@Service("orderItemService")
public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {
    /**
     * queues：声明需要监听的队列
     * channel：当前传输数据的通道
     * 获取实际消息内容有两种方式：
     *  方式一：在方法参数列表中直接声明出来
     *  方式二：从请求体中取出消息的二进制形式，然后通过JSON反序列化即可
     */
    @RabbitListener(queues = {"hello-java-queue"})
    public void revieveMessage(Message message, OrderReturnReasonEntity entity, Channel channel) {
        // 请求体，序列化存储（本例中已使用Jackson2JsonMessageConverter序列化器作JSON序列化存储）
        byte[] body = message.getBody();
        // 请求头
        MessageProperties messageProperties = message.getMessageProperties();
        // JSON反序列得到消息内容对象
        OrderReturnReasonEntity reason = JSONObject.parseObject(body, OrderReturnReasonEntity.class);
        System.out.println("接受到的消息对象" + message);
        System.out.println("接受到的消息内容" + reason);
        System.out.println("接受到的消息内容" + entity);
    }
}

简介：
  复制一份订单配置9011
  -Xmx100m
  --server.port=9011
  同时运行多个订单模块，然后测试发送多条消息
  
结论：
  1.多个客户端可以共同监听同一队列
  2.一条消息同时只能被一个客户端接收
  3.同一个客户端接收消息是串行的，revieveMessage方法执行完后才会继续接收下一条消息

作用：
  配合@RabbitListener，使用方法重载的方法接收不同的消息类型
简介：
  1.用于标注在监听方法上，接收消息，不需要指定监听的队列
  2.使用该注解之前，需要在启动类加上该注解：@EnableRabbit
  3.@RabbitListener只可以标注在方法，重载的方式接收不同消息对象

@Slf4j
@SpringBootTest
class GulimallOrderApplicationTests {
    @Autowired
  RabbitTemplate rabbitTemplate;
    @Test
    void sendMsg2() {
        // 该测试案例向同一队列发送了两个不同类型的消息对象
        // 消息一：
        OrderReturnReasonEntity message1 = new OrderReturnReasonEntity();// 退货原因
        message1.setId(1L);
        message1.setCreateTime(new Date());
        message1.setName("哈哈");
        CorrelationData correlationData1 = new CorrelationData(UUID.randomUUID().toString());// 消息ID
        rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java", message1, correlationData1);
        // 消息二：
        OrderEntity message2 = new OrderEntity();// 退货原因
        message2.setId(1L);
        message2.setCreateTime(new Date());
        message2.setOrderSn("哈哈");
        CorrelationData correlationData2 = new CorrelationData(UUID.randomUUID().toString());// 消息ID
        rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java", message2, correlationData2);
    }
}

@RabbitListener(queues = {"hello-java-queue"})
@Service("orderItemService")
public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {
    /**
     * 方法重载1
     */
    @RabbitHandler
    public void revieveMessage(Message message, OrderReturnReasonEntity entity, Channel channel) {
        // 请求体，序列化存储（本例中已使用Jackson2JsonMessageConverter序列化器作JSON序列化存储）
        byte[] body = message.getBody();
        // 请求头
        MessageProperties messageProperties = message.getMessageProperties();
        // JSON反序列得到消息内容对象
        OrderReturnReasonEntity reason = JSONObject.parseObject(body, OrderReturnReasonEntity.class);
        System.out.println("接受到的消息对象" + message);
        System.out.println("接受到的消息内容" + reason);
        System.out.println("接受到的消息内容" + entity);
    }
    /**
     * 方法重载2
     */
    @RabbitHandler
    public void revieveMessage(Message message, OrderEntity entity, Channel channel) {
        // 请求体，序列化存储（本例中已使用Jackson2JsonMessageConverter序列化器作JSON序列化存储）
        byte[] body = message.getBody();
        // 请求头
        MessageProperties messageProperties = message.getMessageProperties();
        // JSON反序列得到消息内容对象
        OrderEntity reason = JSONObject.parseObject(body, OrderEntity.class);
        System.out.println("接受到的消息对象" + message);
        System.out.println("接受到的消息内容" + reason);
        System.out.println("接受到的消息内容" + entity);
    }
}

spring:
  rabbitmq:
    host: 192.168.56.10
    port: 5672
    # 虚拟主机
    virtual-host: /
    # 开启发送端发送确认，无论是否到达broker都会触发回调【发送端确认机制+本地事务表】
    publisher-confirm-type: correlated
    # 开启发送端抵达队列确认，消息未被队列接收时触发回调【发送端确认机制+本地事务表】
    publisher-returns: true
    # 消息在没有被队列接收时是否强行退回
    template:
      mandatory: true
    # 消费者手动确认模式，关闭自动确认，否则会消息丢失
    listener:
      simple:
        acknowledge-mode: manual

描述：
  可以共用一个exchange，指定不同的路由分别绑定延时队列和死信队列

/**
 * 创建队列，交换机，延时队列，绑定关系 的configuration
 * 1.Broker中的Queue、Exchange、Binding不存在的情况下，会自动创建（在RabbitMQ），不会重复创建覆盖
 * 2.懒加载，只有第一次使用的时候才会创建（例如监听队列）
 */
@Configuration
public class MyRabbitMQConfig {
    /**
     * 延时队列
     */
    @Bean
    public Queue orderDelayQueue() {
        /**
         * Queue(String name,  队列名字
         *       boolean durable,  是否持久化
         *       boolean exclusive,  是否排他
         *       boolean autoDelete, 是否自动删除
         *       Map<String, Object> arguments) 属性【TTL、死信路由、死信路由键】
         */
        HashMap<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", "order-event-exchange");// 死信路由
        arguments.put("x-dead-letter-routing-key", "order.release.order");// 死信路由键
        arguments.put("x-message-ttl", 60000); // 消息过期时间 1分钟
        return new Queue("order.delay.queue", true, false, false, arguments);
    }
    /**
     * 交换机（死信路由）
     */
    @Bean
    public Exchange orderEventExchange() {
        return new TopicExchange("order-event-exchange", true, false);
    }
    /**
     * 死信队列
     */
    @Bean
    public Queue orderReleaseQueue() {
        return new Queue("order.release.order.queue", true, false, false);
    }
    /**
     * 绑定：交换机与订单解锁延迟队列
     */
    @Bean
    public Binding orderCreateBinding() {
        /**
         * String destination, 目的地（队列名或者交换机名字）
         * DestinationType destinationType, 目的地类型（Queue、Exhcange）
         * String exchange,
         * String routingKey,
         * Map<String, Object> arguments
         **/
        return new Binding("order.delay.queue",
                Binding.DestinationType.QUEUE,
                "order-event-exchange",
                "order.create.order",
                null);
    }
    /**
     * 绑定：交换机与订单解锁死信队列
     */
    @Bean
    public Binding orderReleaseBinding() {
        return new Binding("order.release.order.queue",
                Binding.DestinationType.QUEUE,
                "order-event-exchange",
                "order.release.order",
                null);
    }
    /**
     * 绑定：交换机与库存解锁
     */
    @Bean
    public Binding orderReleaseOtherBinding() {
        return new Binding("stock.release.stock.queue",
                Binding.DestinationType.QUEUE,
                "order-event-exchange",
                "order.release.other.#",