rabbitMQ的学习

Posted 2022-02-10 浪上了天

rabbitMQ的使用

MQ模型

简单队列模型

简单队列模型
publisher -----> queue ----> consumer
    
publisher:消息发布者，将消息发送到队列queue
queue：消息队列，负责接收并缓存消息
consumer：订阅队列，处理队列中的消息

发布/订阅模型

发布/订阅模型					

							
					  |-->consumer1
			    |-->queue1--->|
		            |	          |-->consumer2
publisher ---->exchange---->|							
			    |		  |-->consumer1
    			    |-->queue2--->|
                                          |-->consumer2
    
    
publisher:生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给exchange（交换机）
exchange:交换机，一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。
   
    exchange种类：
    	- Fanout：广播，将消息交给所有绑定到交换机的队列
    	- Direct：定向，将消息交给符合routingKey的队列
    	- Topic：主题(使用通配符),将消息交给符合routing pattern(路由模式)的队列
    
consumer:消费者，订阅队列，消费消息
queue：队列，接收消息，缓存消息

使用原生态的rabbitMQ

publisher 实现
    
public class PublisherTest 
    //队列名字
    private final  static  String QUEUE_NAME = "hello";

    public static void main(String[] args) 
        //建立连接
        ConnectionFactory factory = new ConnectionFactory();
        //设置连接参数
        factory.setHost("192.168.10.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("root");
        factory.setPassword("root");
        //建立连接，创建通道
        try(Connection connection = factory.newConnection();
            Channel channel = connection.createChannel())
            //创建队列
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            //发送消息
            String message = "hello world!";
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes(StandardCharsets.UTF_8));
            System.out.println("[x] Sent :" +message);
         catch (IOException | TimeoutException e) 
            e.printStackTrace();
        
    

consumer 实现
    
public class ConsumerTest 

    //队列名字
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws IOException, TimeoutException 
        ConnectionFactory factory = new ConnectionFactory();

        factory.setHost("192.168.10.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("root");
        factory.setPassword("root");

        Connection connection = factory.newConnection();
             Channel channel = connection.createChannel();

            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel) 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                                           AMQP.BasicProperties properties, byte[] body) throws IOException 
                    String message = new String(body);
                    System.out.println("接收到的消息：" + message);
                
            );

    

SpringAMQP的使用

SpringAMQP地址：https://spring.io/projects/spring-amqp

Basic Queue 简单队列模型

1、导入依赖

<!--AMQP依赖，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、配置MQ地址

spring:
  rabbitmq:
    host: 192.168.10.128 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: root # 用户名
    password: root # 密码

3、编写测试类

注：这样写要先在rabbitmq中先创建simple.queue队列   
@Test
public void simplePublisher()
        //消息
        String message = "hello, spring amqp";
        //队列名称
        String queueName = "simple.queue";
        //发送消息
        rabbitTemplate.convertAndSend(queueName,message);
    

4、编写消息接收(与publisher配置一样)

 //监听simple.queue队列
@RabbitListener(queues = "simple.queue")
 public void listenSimple(String msg)
        log.info("接收到的消息：",msg);
    

Work Queue模型

让多个消费者绑定到一个队列，共同消费队列中的消息。

场景：当消息处理比较耗时，生产速度远大于消耗速度，就可以使用此模型

让多个消费者监听同一个队列
    
生产者：
@Test
 public void simplePublisher()
        //消息
        String message = "hello, spring amqp";
        //队列名称
        String queueName = "simple.queue";
        for (int i = 0; i < 50; i++) 
            //发送消息
            rabbitTemplate.convertAndSend(queueName,message);
        
    

消费者：(多个消费者加监听simple.queue队列)
消费者1
@RabbitListener(queues = "simple.queue")
 public void listenSimple1(String msg)
        log.info("接收到的消息：",msg);
    Thread.sleep(1000);
    
消费者2
@RabbitListener(queues = "simple.queue")
 public void listenSimple2(String msg)
        log.info("接收到的消息：",msg);
    
    

上面会有一个问题：消费者2很快就处理了25条信息，但是消费者1还在慢慢处理自己的消息。

出现的原因：消息是平均分配给每个消费者，没有考虑到消费者的处理能力

解决的办法：能者多劳

修改consumer的application.yml配置文件中添加配置

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

发布/订阅模型

Fanout模式（广播模式）

消息发送流程

• 可以有多个队列
• 每个队列绑定到exchange
• 生产者生产消息，发送到交换机上
• 交换机把消息发送到绑定的所有队列
• 订阅队列的消费者，消费消息

1、声明队列和交换机

//声明队列1
    @Bean
    public Queue fanoutQueue1()
        return new Queue("fanout.queue1");
    
    //声明队列2
    @Bean
    public Queue fanoutQueue2()
        return new Queue("fanout.queue2");
    
    //声明交换机
    @Bean
    public FanoutExchange fanoutExchange()
        return new FanoutExchange("amqp.fanout");
    

2、把队列绑定到交换机上

//把队列1和队列2绑定到广播交换机上

    @Bean
    public Binding bindingQueue1()
        return BindingBuilder.bind(fanoutQueue1()).to(fanoutExchange());
    
    @Bean
    public Binding bindingQueue2()
        return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
    

3、发送消息

//生产者将消息发送到交换机上，再由交换机把消息发布到绑定的队列上
    @Test
    public void fanoutPublisher()
        //交换机名称
        String exchange = "amqp.fanout";
        //消息
        String message = "hello world";
        //将消息发送到交换机上
        rabbitTemplate.convertAndSend(exchange,"",message);
    

4、消费消息

//消费者监听队列，并消费队列上的消息

    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg)
        log.info("fanout.queue1接收到的消息：",msg);
    

    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg)
        log.info("fanout.queue2接收到的消息：",msg);
    

direct模式(定向模式)

在fanout模式中，一条消息会被发布到所有订阅的队列，在某些场景下，我们希望不同的消息被不同的队列消费，就可以使用direct模式了

direct模式：

• 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
• publisher 向 Exchange发送消息时，也必须指定消息的 RoutingKey。
• Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

1、使用注解的方式，声明队列和交换机，

//声明队列和交换机，并把队列绑定到交换机上

    //监听routingKey=red 、blue 的消息
    @RabbitListener(bindings = @QueueBinding(
        //队列
        value = @Queue(name = "direct.queue1"),
        //交换机
        exchange = @Exchange(name = "amqp.direct",type = ExchangeTypes.DIRECT),
        //路由key
        key = "red","blue"
    ))
    public void listenDirectQueue1(String msg)
        log.info("direct.queue1接收到的消息：",msg);
    

    //监听routingKey=yellow 、blue 的消息
    @RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue2"),
        exchange = @Exchange(name = "amqp.direct",type = ExchangeTypes.DIRECT),
        key = "yellow","blue"
    ))
    public void listenDirectQueue2(String msg)
        log.info("direct.queue2接收到的消息：",msg);
    

2、声明发布者发送消息

    @Test
    public void directPublisher()
        //交换机名称
        String exchange = "amqp.direct";
        //消息
        String message = "hello world";
        //将消息发送到交换机上
        rabbitTemplate.convertAndSend(exchange,"red",message);  //只有队列1接收到
        rabbitTemplate.convertAndSend(exchange,"yellow",message);  // 只有队列2接收到
        rabbitTemplate.convertAndSend(exchange,"blue",message);   // 队列1和队列2都可以接收到
    

direct交换机和fanout交换机的差异：

• fanout交换机将消息路由给每个与其绑定的队列
• direct交换机根据routingkey判断路由给那个队列
• 如果多个队列具有相同的routingkey，则与fanout功能相似

Topic模式(主题模式)

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

#：匹配一个或多个词

*：匹配不多不少恰好1个词

1、使用注解的方式，声明队列和交换机，

    @RabbitListener(bindings = @QueueBinding(
           value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "amqp.topic",type = ExchangeTypes.TOPIC),
            key = "china.#"
    ))
    public void listenTopicQueue1(String msg)
        log.info("topic.queue1接收到的消息：",msg);
    

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name = "amqp.topic",type = ExchangeTypes.TOPIC),
            key = "*.new"
    ))
    public void listenTopicQueue2(String msg)
        log.info("topic.queue2接收到的消息：",msg);
    

2、声明发布者发布消息

    @Test
    public void topicPublisher()
        //交换机名称
        String exchange = "amqp.topic";
        //消息
        String message = "hello world";
        //将消息发送到交换机上
        rabbitTemplate.convertAndSend(exchange,"china.new",message); // 队列1、2接收到
        rabbitTemplate.convertAndSend(exchange,"china.new.now",message); // 队列1接收到
        rabbitTemplate.convertAndSend(exchange,"topic.new",message);   //队列2接收到
    

消息转换器

Spring会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。

默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：

• 数据体积过大
• 有安全漏洞
• 可读性差

解决问题，使用json转换器

配置json转换器

在publisher和consumer中引入依赖

<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId>
    <version>2.9.10</version>
</dependency>

在启动类配置转换bean

@Bean
public MessageConverter jsonMessageConverter()
    return new Jackson2JsonMessageConverter();

在使用SpringAMQP中存在的问题

• 消息可靠性问题 如何确保发送的消息至少被消费一次
• 延迟消息问题 如何实现消息的延迟投递
• 高可用问题 如何避免单点的MQ故障导致的不可用问题
• 消息堆积 如何解决数百万消息堆积，无法及时消费问题

1、消息可靠性

#消息发送的流程

publisher--->exchange-->queue-->consumer

其中每一步都可能导致消息丢失，常见的丢失原因有：

• 发送时丢失
  • 生产者发送消息为送达exchange
  • 消息到达exchange没有到queue
• MQ宕机，queue将消息丢失
• consumer接收到消息后没有消费就宕机了

解决方案：

• 生产者确认机制
• mq持久化
• 消费者确认机制
• 失败重试机制

生产者确认机制

这种机制必须给每个消息指定一个唯一ID。消息发送到MQ以后，会返回一个结果给发送者，表示消息是否处理成功。

返回结果有两种方式：

• publisher-confirm，发送者确认
  • 消息成功投递到交换机，返回ack
  • 消息未投递到交换机，返回nack
• publisher-return，发送者回执
  • 消息投递到交换机，但是没有路由到队列，返回ack及失败的原因

1、导入依赖

<!--AMQP依赖，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、在publisher和consumer服务中添加application.yml配置文件

spring:
  rabbitmq:
    host: 192.168.10.128 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: root # 用户名
    password: root # 密码
    publisher-confirm-type: correlated
    publisher-returns : true
    template:
    	mandatory: true

• publish-confirm-type：开启publisher-confirm，这里支持两种类型：

  • simple: 同步等待confirm结果直到超时
  • correlated：异步回调，定义ConfirmCallback，MQ返回结果时会回调这个ConfirmCallback

• publish-returns：开启publish-return功能，同样是基于callback机制，不过是定义ReturnCallback

• template.mandatory：定义消息路由失败时的策略。true，则调用ReturnCallback；false：则直接丢弃消息

3、定义return回调（ReturnCallback）

每个RabbitTemplate只能配置一个ReturnCallback，因此需要在项目加载时配置：

@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware 

    // 为RabbitTemplate设置路由到队列失败时调用的方法
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException 
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
        rabbitTemplate.setReturnCallback((message, replyCode, replyTest, exchange, routingKey)
                -> log.info("消息发送失败，应答码，原因，交换机，路由键，消息",
                replyCode,replyTest,exchange,routingKey,message));
    

4、定义ConfirmCallback

ConfirmCallback可以在发送消息时指定，因为每个业务处理confirm成功或失败的逻辑不一定相同。

    @Test
    public void testSendMessage()

        //消息内容
        String message = "hello, spring amqp!";
        //设置回调函数策略
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

        correlationData.getFuture().addCallback(confirm -> 
            //连接MQ正常

            if(confirm.isAck())
                //正常ack情况 把消息发送到了交换机
                log.debug("消息发送成功ack，ID",correlationData.getId());
            else
                //nack情况  消息没有发送到交换机
                log.info("消息发送失败-nack，ID",correlationData.getId());
            
        
        //连接MQ异常
        , throwable -> log.error("消息发送失败-连接mq异常，ID",correlationData.getId()));
		//发送消息
        rabbitTemplate.convertAndSend("amq.topic","simple.test",message,correlationData);
    

消息持久化

生产者可以确定的将消息投递到mq中，但是消息发送到mq以后，突然宕机，导致消息丢失，因此想要确保消息在mq中，就需要开启消息持久化机制。

• 交换机持久化
• 队列持久化
• 消息持久化

交换机持久化

@Bean
public DirectExchange simpleExchange()
    // 三个参数：交换机名称、是否持久化、当没有queue与其绑定时是否自动删除
    return new DirectExchange("simple.direct", true, false);

由SpringAMQP声明的交换机都是持久化的。

队列持久化

    @Bean
    public Queue simpleQueue()
        // 使用QueueBuilder构建队列，durable就是持久化的
        return QueueBuilder.durable("simple.queue").build();
    


    @Bean
    public Queue simpleQueue()
        //四个参数分别是：队列名，持久化，是否独有，当没有exchange与其绑定时是否自动删除
        return new Queue("fanout.queue2",true,false,false);
    

由SpringAMQP声明的队列都是持久化的。

消息持久化

利用SpringAMQP发送消息时，可以设置消息的属性（MessageProperties），指定delivery-mode

• 持久化
• 非持久化

        Message message = MessageBuilder
                .withBody("hello, spring amqp!".getBytes(StandardCharsets.UTF_8))
                .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                .build();

SpringAMQP发出的任何消息都是持久化的

消费者确认机制

RabbitMQ是阅后即焚机制，RabbitMQ确认消息被消费者消费后会立刻删除。

而RabbitMQ是通过消费者回执来确认消费者是否成功处理消息的：消费者获取消息后，应该向RabbitMQ发送ACK回执，表明自己已经处理消息。

SpringAMQP则允许配置三种确认模式：

• manual：手动ack，需要在业务代码结束后，调用api发送ack。

• auto：自动ack，由spring监测listener代码是否出现异常，没有异常则返回ack；抛出异常则返回nack

• none：关闭ack，MQ假定消费者获取消息后会成功处理，因此消息投递后立即被删除

修改application.yml配置文件

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: none # 关闭ack  默认的是auto模式

本地失败重试机制

我们可以利用Spring的retry机制，在消费者出现异常时利用本地重试，而不是无限制的requeue到mq队列。

修改application.yml配置文件

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true # 开启消费者失败重试
          initial-interval: 1000ms # 初始的失败等待时长为1秒
          multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
          max-attempts: 3 # 最大重试次数
          stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

失败策略

在之前的测试中，达到最大重试次数后，消息会被丢弃，这是由Spring内部机制决定的。

在开启重试模式后，重试次数耗尽，如果消息依然失败，则需要有MessageRecovery接口来处理，它包含三种不同的实现：

• RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息。默认就是这种方式

• ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队

• RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机

比较优雅的一种处理方案是RepublishMessageRecoverer，失败后将消息投递到一个指定的，专门存放异常消息的队列，后续由人工集中处理。

1、在consumer服务中定义处理失败消息啊的队列和交换机

//错误交换机
@Bean
public DirectExchange errorMessageExchange()
    return new DirectExchange("error.direct");

//储存错误消息的队列
@Bean
public Queue errorQueue()
    return new Queue("error.queue", true);

//把队列绑定到交换机上
@Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange)
    return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");

2、定义RepublishMessageRecoverer，关联队列和交换机

@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate)
    return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");

2、延迟消息问题

死信交换机

什么是死信：

当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）：

• 消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false
• 消息是一个过期消息，超时无人消费
• 要投递的队列消息满了，无法投递

死信交换机（Dead Letter Exchange，DLX）如果这个包含死信的队列配置了dead-letter-exchange属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中

//死信交换机的流程


	普通的交换机	      reject |
publisher--->exchange--->queue------>|consumer
                           /         |
                          /
                         /
		    exchange----->queue
		      dead交换机      储存死信的队列

1、在consumer服务中定义死信交换机、死信队列

//声明一个普通队列关联死信交换机
    @Bean
    public Queue simpleQueue()
        return QueueBuilder.durable("simple.queue")
                .deadLetterExchange("dead.direct")
                .deadLetterRoutingKey("simple")
                .build();
    
//声明死信交换机
    @Bean
    public DirectExchange dlExchange()
        return new DirectExchange("dead.direct",true,false);

    
//声明死信队列
    @Bean
    public Queue dlQueue()
        return new Queue("dead.queue",true);
    
//将死信队列关联到死信交换机上
    @Bean
    public Binding binding()
        return BindingBuilder.bind(dlQueue()).to(dlExchange()).with("simple");
    

# 消费者确认模式: auto
	acknowledge-mode: auto
    特征: 当消息不能被消费时,会重新入队,再次投递给消费者进行被消费
	default-requeue-rejected: false # 拒绝消息重新入队,如果队列绑定了死信交换机则消息会投递到死信交换机并路由到死信队列
# 本地重试:
	当本地重试次数耗尽时,如果当前队列没有绑定死信交换机或错误队列,则消息丢弃
    如果提供了错误队列,则消息投递到错误队列
    如果队列绑定了死信交换机,则消息以死信的形式存放到死信队列

TTL

一个队列中的消息如果超时未消费，则会变为死信，超时分为两种情况：

• 消息所在的队列设置了超时时间
• 消息本身设置了超时时间

ttl=5000	ttl.direct   ttl.queue
publisher--->exchange--->queue
    			/
    		       /
    		      /
		exchange---->queue
	dl.ttl.exchange         dl.ttl.queue

队列设置超时时间

1、声明 死信交换机、死信队列：

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "dl.ttl.queue", durable = "true"),
    exchange = @Exchange(name = "dl.ttl.direct"),
    key = "ttl"
))
public void listenDlQueue(String msg)
    log.info("接收到 dl.ttl.queue的延迟消息：", msg);

2、声明一个队列，指定TTL

@Bean
public Queue ttlQueue()
    return QueueBuilder.durable("ttl.queue") // 指定队列名称，并持久化
        .ttl(10000) // 设置队列的超时时间，10秒
        .deadLetterExchange("dl.ttl.direct") // 指定死信交换机
        .deadLetterRoutingKey("ttl")
        .build();

3、声明交换机，将ttl.queue队列关联上

@Bean
public DirectExchange ttlExchange()
    return new DirectExchange("ttl.direct");

@Bean
public Binding ttlBinding()
    return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");

4、发送消息

@Test
public void testTTLQueue() 
    // 创建消息
    String message = "hello, ttl queue";
    // 消息ID，需要封装到CorrelationData中
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    // 发送消息
    rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);
    // 记录日志
    log.debug("发送消息成功");

消息设置超时时间

@Test
public void testSendTTLMessage() throws InterruptedException 
        // 1.消息体
//        String msg = "超时消息...";
        Message msg = MessageBuilder
                .withBody("hello, ttl message".getBytes(StandardCharsets.UTF_8))
                .setExpiration("5000")
                .build();
        // 2.发送消息
        rabbitTemplate.convertAndSend("ttl.direct","ttl", msg);
        log.info("发送消息成功...");
    

延迟队列

利用TTL结合死信交换机，我们实现了消息发出后，消费者延迟收到消息的效果。这种消息模式就称为延迟队列（Delay Queue）模式。

延迟队列的使用场景包括：

• 延迟发送短信
• 用户下单，如果用户在15 分钟内未支付，则自动取消
• 预约工作会议，20分钟后自动通知所有参会人员

安装DelayExchange插件https://www.rabbitmq.com/community-plugins.html

安装DelayExchange

1、下载插件

2、我们是基于docker安装的rabbitmq，所以要把下载的文件挂载到数据卷中

// 查看数据卷的位置
docker volume inspect mq-plugins   

3、把下载的文件放在数据卷中

/var/lib/docker/volumes/mq-plugins/_data

4、安装插件

进入容器

docker exec -it mq bash

开启插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

安装完成

使用DelayExchange

1、声明DelayExchange交换机

   //延迟队列
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "delay.queue",durable = "true"),
            exchange = @Exchange(name = "delay.direct",delayed = "true"),
            key = "delay"
    ))
    public void listenDelayedQueue(String msg)
        log.info("接收到delay.queue的消息：",msg);
    

2、发送消息

发送消息时，一定要携带x-delay属性，指定延迟时间

    @Test
    public void testDelayMsg()
        Message message = MessageBuilder
                .withBody("hello delay message".getBytes(StandardCharsets.UTF_8))
                .setHeader("x-delay",1000)
                .build();

        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend("delay.direct","delay",message,correlationData);
        log.info("发送消息成功");
    

3、消息堆积问题

惰性队列

消息堆积问题：

当生产者发送消息的速度超过了消费者处理消息的速度，就会导致队列中的消息堆积，直到队列存储消息达到上限。之后发送的消息就会成为死信，可能会被丢弃

解决消息堆积问题的思路：

• 增加更多消费者，提高消费速度。也就是我们之前说的work queue模式
• 扩大队列容积，提高堆积上限

惰性队列

特征：

• 接收到消息后直接存入磁盘而非内存
• 消费者要消费消息时才会从磁盘中读取并加载到内存
• 支持数百万条的消息存储

设置惰性队列

1、使用命令行设置lazy-queue

rabbitmqctl set_policy Lazy "^lazy-queue$" \'"queue-mode":"lazy"\' --apply-to queues  

• rabbitmqctl ：RabbitMQ的命令行工具
• set_policy ：添加一个策略
• Lazy ：策略名称，可以自定义
• "^lazy-queue$" ：用正则表达式匹配队列的名字
• \'"queue-mode":"lazy"\' ：设置队列模式为lazy模式
• --apply-to queues ：策略的作用对象，是所有的队列

2、基于@Bean声明lazy-queue

    @Bean
    public Queue lazyQueue()
        return QueueBuilder.durable("lazy-queue")
                .lazy()   //开启x-queue-mode 为lazy
                .build();
    

3、基于@RabbitListener声明lazy-queue

    @RabbitListener(queuesToDeclare = @Queue(
            name = "lazy-queue",
            durable = "true",
            arguments = @Argument(name = "x-queue-mode",value = "lazy")
    ))
    public void listenLazyQueue(String msg)
        log.info("接收到消息：",msg);
    

惰性队列的优点

• 基于磁盘存储，消息上限高
• 没有间歇性的page-out，性能比较稳定

惰性队列的缺点

• 基于磁盘存储，消息时效性会降低
• 性能受限于磁盘的IO

4、高可用问题

搭建集群

集群的分类

• 普通集群：是一种分布式集群，将队列分散到集群的各个节点，从而提高整个集群的并发能力。

• 镜像集群：是一种主从集群，普通集群的基础上，添加了主从备份功能，提高集群的数据可用性。

普通集群

普通集群，或者叫标准集群（classic cluster），具备下列特征：

• 会在集群的各个节点间共享部分数据，包括：交换机、队列元信息。不包含队列中的消息。
• 当访问集群某节点时，如果队列不在该节点，会从数据所在节点传递到当前节点并返回
• 队列所在节点宕机，队列中的消息就会丢失

普通模式：普通模式集群不进行数据同步，每个MQ都有自己的队列、数据信息（其它元数据信息如交换机等会同步）。例如我们有2个MQ：mq1，和mq2，如果你的消息在mq1，而你连接到了mq2，那么mq2会去mq1拉取消息，然后返回给你。如果mq1宕机，消息就会丢失。

主机名	控制台端口	amqp通信端口
mq1	8081 ---> 15672	8071 ---> 5672
mq2	8082 ---> 15672	8072 ---> 5672
mq3	8083 ---> 15672	8073 ---> 5672

1、获取cookie

要使两个节点能够通信，它们必须具有相同的共享秘密，称为Erlang cookie。cookie 只是一串最多 255 个字符的字母数字字符。

每个集群节点必须具有相同的 cookie。实例之间也需要它来相互通信。

进入容器获取集群

docker exec -it mq cat /var/lib/rabbitmq/.erlang.cookie

得到的cookie值如下

NOMLMQCBYBBLHWWTFMZO

2、在/tmp目录新建一个配置文件rabbitmq.conf

cd /tmp
# 创建文件
touch rabbitmq.conf

文件内容

loopback_users.guest = false
listeners.tcp.default = 5672
cluster_formation.peer_discovery_backend = rabbit_peer_discovery_classic_config
cluster_formation.classic_config.nodes.1 = rabbit@mq1
cluster_formation.classic_config.nodes.2 = rabbit@mq2
cluster_formation.classic_config.nodes.3 = rabbit@mq3

在创建一个文件，记录cookie

cd /tmp
# 创建cookie文件
touch .erlang.cookie
# 写入cookie
echo "NOMLMQCBYBBLHWWTFMZO" > .erlang.cookie
# 修改cookie文件的权限
chmod 600 .erlang.cookie

3、创建3个目录

cd /tmp
# 创建目录
mkdir mq1 mq2 mq3
    
然后拷贝rabbitmq.conf、cookie文件到mq1、mq2、mq3：
    
# 拷贝
cp rabbitmq.conf mq1
cp rabbitmq.conf mq2
cp rabbitmq.conf mq3
cp .erlang.cookie mq1
cp .erlang.cookie mq2
cp .erlang.cookie mq3   

4、启动集群

创建一个网络

docker network create mq-net

运行命令

第一个mq
    
docker run -d --net mq-net \\
-v $PWD/mq1/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \\
-v $PWD/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \\
-e RABBITMQ_DEFAULT_USER=root \\
-e RABBITMQ_DEFAULT_PASS=root \\
--name mq1 \\
--hostname mq1 \\
-p 8071:5672 \\
-p 8081:15672 \\
rabbitmq:3.8-management
    
第二个mq
    
docker run -d --net mq-net \\
-v $PWD/mq2/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \\
-v $PWD/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \\
-e RABBITMQ_DEFAULT_USER=root \\
-e RABBITMQ_DEFAULT_PASS=root \\
--name mq2 \\
--hostname mq2 \\
-p 8072:5672 \\
-p 8082:15672 \\
rabbitmq:3.8-management
    
第三个mq
    
docker run -d --net mq-net \\
-v $PWD/mq3/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \\
-v $PWD/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \\
-e RABBITMQ_DEFAULT_USER=itcast \\
-e RABBITMQ_DEFAULT_PASS=123321 \\
--name mq3 \\
--hostname mq3 \\
-p 8073:5672 \\
-p 8083:15672 \\
rabbitmq:3.8-management

5、访问web控制台，验证是否搭建成功

镜像集群

镜像模式特征：

• 镜像队列结构是一主多从（从就是镜像）
• 所有操作都是主节点完成，然后同步给镜像节点
• 主宕机后，镜像节点会替代成新的主（如果在主从同步完成前，主就已经宕机，可能出现数据丢失）
• 不具备负载均衡功能，因为所有操作都会有主节点完成（但是不同队列，其主节点可以不同，可以利用这个提高吞吐量）

镜像模式配置

镜像模式有三种模式、

ha-mode	ha-params	效果
准确模式exactly	队列的副本量count	集群中队列副本（主服务器和镜像服务器之和）的数量。count如果为1意味着单个副本：即队列主节点。count值为2表示2个副本：1个队列主和1个队列镜像。换句话说：count = 镜像数量 + 1。如果群集中的节点数少于count，则该队列将镜像到所有节点。如果有集群总数大于count+1，并且包含镜像的节点出现故障，则将在另一个节点上创建一个新的镜像。
all	(none)	队列在群集中的所有节点之间进行镜像。队列将镜像到任何新加入的节点。镜像到所有节点将对所有群集节点施加额外的压力，包括网络I / O，磁盘I / O和磁盘空间使用情况。推荐使用exactly，设置副本数为（N / 2 +1）。
nodes	node names	指定队列创建到哪些节点，如果指定的节点全部不存在，则会出现异常。如果指定的节点在集群中存在，但是暂时不可用，会创建节点到当前客户端连接到的节点。

exactly模式

rabbitmqctl set_policy ha-two "^two\\." \'"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"\'

• rabbitmqctl set_policy：固定写法
• ha-two：策略名称，自定义
• "^two\\."：匹配队列的正则表达式，符合命名规则的队列才生效，这里是任何以two.开头的队列名称
• \'"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"\': 策略内容
  • "ha-mode":"exactly"：策略模式，此处是exactly模式，指定副本数量
  • "ha-params":2：策略参数，这里是2，就是副本数量为2，1主1镜像
  • "ha-sync-mode":"automatic"：同步策略，默认是manual，即新加入的镜像节点不会同步旧的消息。如果设置为automatic，则新加入的镜像节点会把主节点中所有消息都同步，会带来额外的网络开销

all模式

rabbitmqctl set_policy ha-all "^all\\." \'"ha-mode":"all"\'

• ha-all：策略名称，自定义
• "^all\\."：匹配所有以all.开头的队列名
• \'"ha-mode":"all"\'：策略内容
  • "ha-mode":"all"：策略模式，此处是all模式，即所有节点都会称为镜像节点

nodes模式

rabbitmqctl set_policy ha-nodes "^nodes\\." \'"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]\'

• rabbitmqctl set_policy：固定写法
• ha-nodes：策略名称，自定义
• "^nodes\\."：匹配队列的正则表达式，符合命名规则的队列才生效，这里是任何以nodes.开头的队列名称
• \'"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]\': 策略内容
  • "ha-mode":"nodes"：策略模式，此处是nodes模式
  • "ha-params":["rabbit@mq1", "rabbit@mq2"]：策略参数，这里指定副本所在节点名称

仲裁队列

仲裁队列：仲裁队列是3.8版本以后才有的新功能，用来替代镜像队列，具备下列特征：

• 与镜像队列一样，都是主从模式，支持主从数据同步
• 使用非常简单，没有复杂的配置
• 主从同步基于Raft协议，强一致

在任意控制台添加一个队列，一定要选择队列类型为Quorum类型。

使用java代码创建

@Bean
public Queue quorumQueue() 
    return QueueBuilder
        .durable("quorum.queue") // 持久化
        .quorum() // 仲裁队列
        .build();

集群扩容

1、新建一个集群

docker run -d --net mq-net \\
-v $PWD/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \\
-e RABBITMQ_DEFAULT_USER=root \\
-e RABBITMQ_DEFAULT_PASS=root \\
--name mq4 \\
--hostname mq4 \\
-p 8074:15672 \\
-p 8084:15672 \\
rabbitmq:3.8-management

2、进入容器

 docker exec -it mq4 bash

3、停止mq进程

rabbitmqctl stop_app

4、重置mq中的数据

rabbitmqctl reset

5、加入集群

rabbitmqctl join_cluster rabbit@mq1  // rabbit@mq1随便选择集群中的服务都可以，相当于介绍人

6、启动mq

rabbitmqctl start_app

增加仲裁队列副本

1、先查看quorum.queue这个队列目前的副本情况，进入mq1容器：

docker exec -it mq1 bash
    
rabbitmq-queues quorum_status "haha.queue"  //"haha.queue" 仲裁队列的名字

把mq4也加入进来

rabbitmq-queues add_member "haha.queue" "rabbit@mq4"  // "haha.queue" 仲裁队列的名字   "rabbit@mq4" 服务的节点

我们是最棒的