RabbitMQ面试热点

4种交换器类型
    Direct Exchange: 直连交换器
    小结：直连交换器发送消息会根据路由和交换机的绑定关系发送到队列
         
         如果交换器名称为"",将使用默认交换器
         默认交换器不会绑定任何队列，mq会直接把route_key当做queue名称去查找
​
    Fanout Exchange: 分发交换器（扇出交换器）
    小结: 分发交换器发送消息会分发至所有和其有绑定的队列中，这样消息会被多个消费者         处理
​
    Topic Exchange: 主题交换器
    小结：主题交换器可以让每个队列只接收它关注的信息
​
    Headers Exchange: 头信息交换器（了解）
        头信息交换器可以实现更为复杂的匹配但性能不好，
        不推荐使用，了解即可

5种工作模式
    简单模式：
        消息只有一个消费者
        使用默认交换器即可(direct)
    工作队列模式：
        消息有多个消费者，消息只可以被消费一次
        使用默认交换器即可(direct)
    发布订阅模式:
        消息有多个消费者，而且消息会被多个消费者同时消费
        使用分发交换器即可(fanout)
    路由模式:
        根据路由的key，将消息发送到指定的队列
        使用直接交换器即可(direct)
    通配符（主题）模式：
        根据路由的key,进行通配符匹配,发送到指定的队列(topic)
        使用主题交换器即可

解耦场景:
    1. 商品搜索功能 数据库及索引库的一致性
    2. 商品详情静态化，静态页面的生成
延迟队列:
    3. 利用rabbitmq的死信队列功能实现延时处理
异步采集:
    4. 监控数据的采集可以使用rabbitmq异步采集

transaction机制就是说，发送消息前，开启事务（channel.txSelect()）,然后发送消息，如果发送过程中出现什么异常，事务就会回滚（channel.txRollback()）,如果发送成功则提交事务（channel.txCommit()）。然而，这种方式有个缺点：吞吐量下降。

一旦channel进入confirm模式，所有在该信道上发布的消息都将会被指派一个唯一的ID（从1开始），一旦消息被投递到所有匹配的队列之后，rabbitMQ就会发送一个ACK给生产者（包含消息的唯一ID），这就使得生产者知道消息已经正确到达目的队列了。如果rabbitMQ没能处理该消息，则会发送一个Nack消息给你，你可以进行重试操作。处理Ack和Nack的代码如下所示

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    publisher-confirm-type: correlated  # 开启确认机制回调 必须配置这个才会确认回调
    publisher-returns: true # 开启return机制回调
rabbit:
  queues:
    red: red_queue
    green: green_queue
    yellow: yellow_queue
  exchange: my_topic_exchange

@SpringBootTest
class RabbitmqProducerApplicationTests {
    @Autowired
    RabbitTemplate rabbitTemplate;
    @Test
    void sendMsg(){
        //发送消息确认机制监听
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             * @param correlationData 回调数据
             * @param ack  true: 确认  false: 未确认
             * @param cause 原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("回调数据==>" + correlationData);
                System.out.println("是否确认==>" + ack);
                System.out.println("原因==>" + cause);
            }
        });
        //消费者在消息没有被路由到合适队列情况下会被return监听，而不会自动删除
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("返还消息==>"+message);
                System.out.println("回复编码==>"+replyCode);
                System.out.println("回复消息==>"+replyText);
                System.out.println("交换器信息==>"+exchange);
                System.out.println("路由key==>"+routingKey);
            }
        });
//        rabbitTemplate.convertAndSend("","xx.red.green.xxx","wahaha");
        rabbitTemplate.convertAndSend("my_topic_exchange","xx.red.xx","111");
    }
}

    /**
     * 默认队列是持久化的，非自动删除的
     * The queue is durable, non-exclusive and non auto-delete.
     *
     * @param name the name of the queue.
     */
    public Queue(String name) {
        this(name, true, false, false);
    }

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    listener:
      simple:
        # acknowledge-mode: manual #手动确认
        acknowledge-mode: auto #自动确认    manual #手动确认

// 在监听消息的方法中 加入抛异常的逻辑
​
if(null!=msg || "1".equals(msg)){
     throw new RuntimeException("出错了");
}

 RabbitMQ自动补偿机制触发:(多用于调用第三方接口)
    1. 当我们的消费者在处理我们的消息的时候,程序抛出异常情况下触发自动补偿(默认无限次数重试)
    2. 应该对我们的消息重试设置间隔重试时间,比如消费失败最多只能重试5次,间隔3秒(防止重复消费,幂等问题)
    3. 如果重试后还未消费默认丢弃，如果配置了死信队列，会发送到死信队列中
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    listener:
      simple:
        acknowledge-mode: manual #手动确认
        # acknowledge-mode: auto #自动确认    manual #手动确认
        # 重试策略相关配置
        retry:
          enabled: true # 是否开启重试功能
          max-attempts: 5 # 最大重试次数
          # 时间策略乘数因子   0  1  2  4  8
          # 时间策略乘数因子
          multiplier: 2.0
          initial-interval: 1000ms # 第一次调用后的等待时间
          max-interval: 20000ms # 最大等待的时间值
        default-requeue-rejected: false # 重试后还未消费 是否丢弃 发送到死信对垒
rabbit:
  queues:
    red: red_queue
    green: green_queue
    yellow: yellow_queue

public void greenMsg(String msg){
    System.out.println("green队列收到消息==>" + msg + "当前时间:"+ LocalDateTime.now());
    if(null!=msg || "1".equals(msg)){
        throw new RuntimeException("出错了");
    }
}

    /**
     * 死信队列交换器
     * @return
     */
    @Bean
    public Exchange deadExchange(){
        return new DirectExchange("dead_exchange");
    }
    /**
     * 死信队列
     * @return
     */
    @Bean
    public Queue deadQueue(){
        return new Queue("dead_queue");
    }
    /**
     * 将死信交换器 与 死信队列绑定
     * @return
     */
    @Bean
    public Binding deadBindings(){
        return new Binding("dead_queue", Binding.DestinationType.QUEUE,"dead_exchange","dead",null);
    }

/**
     * 为green队列设置死信队列的交换器和路由
     *
     * 这样 重试失败的消息 或 失效的消息 会被发送到对应死信队列中
     *
     * @return
     */
    @Bean
    public Queue greenQueue(){
        Map<String,Object> args=new HashMap<>();
        // 设置该Queue的死信的信箱
        args.put("x-dead-letter-exchange", "dead_exchange");
        // 设置死信routingKey
        args.put("x-dead-letter-routing-key", "dead");
        return new Queue(GREEN_QUEUE,true,false,false,args);
    }

@Test
    void sendMsg2(){
        // 自己构建消息
        Message message = MessageBuilder.withBody("我是消息内容".getBytes()).build();
        // 设置消息的全局ID
        message.getMessageProperties().setMessageId(UUID.randomUUID().toString()); // 设置自定义的消息ID
        // 发送消息
        rabbitTemplate.send("my_topic_exchange","xx.yellow.xx",message);
    }

Map cacheMap = new HashMap(); // 用map模拟redis中的key:value结构
    @RabbitListener(
            queues = "yellow_queue"
    )
    public void yellowMsg(Message message){
        // 商品数据   商品ID
        byte[] body = message.getBody();
        // 获取消息ID (mq生成的)
        String messageId = message.getMessageProperties().getMessageId();
        System.out.println("消息ID messageId==>"+messageId);
        String msg = new String(body);
        // 查看记录
        Object o = cacheMap.get(messageId);
        if(o!=null){
            System.out.println("已处理");
        }else {
            System.out.println("yellow队列收到消息==>" + msg);
        }
        // 处理该消息将处理记录保存
        cacheMap.put(messageId,msg);
        // 模拟出现异常
        throw new RuntimeException();
    }

以两个节点（rabbit01、rabbit02）为例来进行说明。
rabbit01和rabbit02两个节点仅有相同的元数据，即队列的结构，但消息实体只存在于其中一个节点
rabbit01（或者rabbit02）中。 当消息进入rabbit01节点的Queue后，consumer从rabbit02节点消费
时，RabbitMQ会临时在rabbit01、rabbit02间进行消息传输，把A中的消息实体取出并经过B发送给consumer。
所以consumer应尽量连接每一个节点，从中取消息，即对于同一个逻辑队列，要在多个节点建立物理Queue；否则无论consumer连rabbit01或rabbit02，出口总在rabbit01，会产生瓶颈。
当rabbit01节点故障后，rabbit02节点无法取到rabbit01节点中还未消费的消息实体。如果做了消息持久化，那么得等rabbit01节点恢复，然后才可被消费；如果没有持久化的话，就会产生消息丢失的现象。

在普通集群的基础上，把需要的队列做成镜像队列，消息实体会主动在镜像节点间同步，而不是在客户端取数据时临时拉取，也就是说多少节点消息就会备份多少份。该模式带来的副作用也很明显，除了降低系统性能外，如果镜像队列数量过多，加之大量的消息进入，集群内部的网络带宽将会被这种同步通
讯大大消耗掉。所以在对可靠性要求较高的场合中适用
由于镜像队列之间消息自动同步，且内部有选举master机制，即使master节点宕机也不会影响整个集群的使用，达到去中心化的目的，从而有效的防止消息丢失及服务不可用等问题。

消息追踪之rabbitmq_tracing插件
作用: 将发消息 消费消息记录到日志中
https://www.cnblogs.com/Harriet/p/10149144.html

spring.rabbitmq.address 
客户端连接的地址，有多个的时候使用逗号分隔，该地址可以是IP与Port的结合   
​
spring.rabbitmq.cache.channel.checkout-timeout
当缓存已满时，获取Channel的等待时间，单位为毫秒  
​
spring.rabbitmq.cache.channel.size  
缓存中保持的Channel数量  
​
spring.rabbitmq.cache.connection.mode   
连接缓存的模式 CHANNEL
​
spring.rabbitmq.cache.connection.size   
缓存的连接数   
​
spring.rabbitmq.connnection-timeout 
连接超时参数单位为毫秒：设置为“0”代表无穷大  
​
spring.rabbitmq.dynamic 
默认创建一个AmqpAdmin的Bean    true
​
spring.rabbitmq.host    
RabbitMQ的主机地址   localhost
​
spring.rabbitmq.listener.acknowledge-mode   
容器的acknowledge模式     
​
spring.rabbitmq.listener.auto-startup   
启动时自动启动容器   true
​
spring.rabbitmq.listener.concurrency    
消费者的最小数量     
​
spring.rabbitmq.listener.default-requeue-rejected   
投递失败时是否重新排队 true
​
spring.rabbitmq.listener.max-concurrency    
消费者的最大数量     
​
spring.rabbitmq.listener.prefetch   
在单个请求中处理的消息个数，他应该大于等于事务数量    
​
spring.rabbitmq.listener.retry.enabled  
不论是不是重试的发布  false
​
spring.rabbitmq.listener.retry.initial-interval 
第一次与第二次投递尝试的时间间隔    1000
​
spring.rabbitmq.listener.retry.max-attempts 
尝试投递消息的最大数量 3
​
spring.rabbitmq.listener.retry.max-interval 
两次尝试的最大时间间隔 10000
​
spring.rabbitmq.listener.retry.multiplier   
上一次尝试时间间隔的乘数    1.0
​
spring.rabbitmq.listener.retry.stateless    
不论重试是有状态的还是无状态的 true
​
spring.rabbitmq.listener.transaction-size   
在一个事务中处理的消息数量。为了获得最佳效果，该值应设置为小于等于每个请求中处理的消息个数，即spring.rabbitmq.listener.prefetch的值   
​
spring.rabbitmq.password    
登录到RabbitMQ的密码   
​
spring.rabbitmq.port    
RabbitMQ的端口号    5672
​
spring.rabbitmq.publisher-confirms  
开启Publisher Confirm机制   false
​
spring.rabbitmq.publisher-returns   
开启publisher Return机制    false
​
spring.rabbitmq.requested-heartbeat 
请求心跳超时时间，单位为秒    
​
spring.rabbitmq.ssl.enabled 
启用SSL支持 false
​
spring.rabbitmq.ssl.key-store   
保存SSL证书的地址   
​
spring.rabbitmq.ssl.key-store-password  
访问SSL证书的地址使用的密码  
​
spring.rabbitmq.ssl.trust-store 
SSL的可信地址     
​
spring.rabbitmq.ssl.trust-store-password    
访问SSL的可信地址的密码    
​
spring.rabbitmq.ssl.algorithm   
SSL算法，默认使用Rabbit的客户端算法库  
​
spring.rabbitmq.template.mandatory  
启用强制信息  false
​
spring.rabbitmq.template.receive-timeout    
receive()方法的超时时间    0
​
spring.rabbitmq.template.reply-timeout  
sendAndReceive()方法的超时时间 5000
​
spring.rabbitmq.template.retry.enabled  
设置为true的时候RabbitTemplate能够实现重试  false
​
spring.rabbitmq.template.retry.initial-interval 
第一次与第二次发布消息的时间间隔    1000
​
spring.rabbitmq.template.retry.max-attempts 
尝试发布消息的最大数量 3
​
spring.rabbitmq.template.retry.max-interval 
尝试发布消息的最大时间间隔   10000
​
spring.rabbitmq.template.retry.multiplier   
上一次尝试时间间隔的乘数    1.0
​
spring.rabbitmq.username    
登录到RabbitMQ的用户名  
​
spring.rabbitmq.virtual-host    
连接到RabbitMQ的虚拟主机