1. 简介与安装

RabbitMQ是基于Erlang语言开发的开源消息通信中间件,支持AMQP,XMPP,SMTP,STOMP协议,消息延迟时微秒级别的。
Ubuntu系统RabbitMQ的安装
在这里插入图片描述

2. 基本概念

  1. Publisher 生产者,发送消息的一方
  2. Consumer 消费者,接收消息的一方
  3. Queue 队列,存储消息
  4. Exchange 交换机,负责消息路由,生产者发送的消息由交换机负责投递到相应的队列。不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失!
  5. VirtualHost 虚拟主机,起到数据隔离的作用,有各自的交换机和队列

3. SpringAMQP

  1. 导入Maven依赖

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    
  2. SpringAMQP提供了RabbitTemplate工具,用于发送消息

  3. yml配置

    spring:
      rabbitmq:
        host: 127.0.0.1 # MQ部署的机器IP
        port: 5672 # 端口
        virtual-host: /test # 虚拟主机
        username: admin # 用户名
        password: admin # 密码
    
  4. RabbitMQ管理系统配置

    • 创建虚拟主机/test
    • 创建交换机test.direct
    • 创建队列test.queue
    • 将队列test.queue绑定到交换机test.direct
  5. 发送消息测试

    class LearnApplicationTests {
    
        @Autowired
        RabbitTemplate rabbitTemplate;
        
        @Test
        void testSend() {
            String exchange = "test.direct";
            String msg = "hello RabbitMQ";
            rabbitTemplate.convertAndSend(exchange, "", msg);
        }
    }
    
  6. 接收消息测试

    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    /**
     * @author zyq
     */
    @Component
    public class SpringRabbitListener {
        @RabbitListener(queues = "test.queue")
        public void listenSimpleQueueMessage(String msg) {
            System.out.println("消费者接收到消息:【" + msg + "】");
        }
    }
    
  7. 一个队列上存在多个监听器时,假定队列test.queue上有两个消费者listener1和listener2,默认情况下队列上的消息会由两个消费者平均分配(第一个消息发给listener1,第二个消息发给listener2,第三个消息发给listener1, …),如果两个消费者的性能存在差异,那么性能好的消费者的资源无法充分利用,可以通过配置prefetch = 1切换到“能者多劳”策略。

    spring:
      rabbitmq:
        host: 127.0.0.1 # MQ部署的机器IP
        port: 5672 # 端口
        virtual-host: /test_host # 虚拟主机
        username: admin # 用户名
        password: admin # 密码
        listener:
          simple:
            prefetch: 1 # 能者多劳,不配置的话是将消息平均发送给消费者
    

4. 交换机类型

  1. Fanout交换机 广播交换机,将消息发送到所有绑定的队列
  2. Direct交换机 根据消息的Routing Key进行判断,只有队列的Routingkey与消息的 Routingkey完全一致,才会接收到消息。
  3. Topic交换机 可以让队列在绑定BindingKey 的时候使用通配符
    • # 匹配一个或多个词
    • * 匹配一个词

5. 消息转换器

5.1 默认转换器

在数据传输时,发送的消息被序列化为字节发送给MQ,接收消息的时候,还会把字节反序列化为Java对象。 只不过,默认情况下Spring采用的序列化方式是JDK序列化。众所周知,JDK序列化存在下列问题:

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

5.2 配置JSON转换器

  1. 引入Maven依赖
    <dependency>
        <groupId>com.fasterxml.jackson.dataformat</groupId>
        <artifactId>jackson-dataformat-xml</artifactId>
    </dependency>
    
  2. 配置Bean
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
    

6 生产者的可靠性

一般情况下,只要生产者与MQ之间的网路连接顺畅,基本不会出现发送消息丢失的情况。少数情况下,可能出现投递的消息没有成功入队。

6.1 生产者超时重连机制

在生产者服务中进行如下配置

spring:
  rabbitmq:
    connection-timeout: 1s # 连接超时时间
    template:
      retry:
        enabled: true # 开启超时重连机制
        initial-interval: 1000ms # 初始等待时间
        multiplier: 1 # 等待时长倍数,下次等待时长 initial-interval * multiplier
        max-attempts: 3 # 重试次数

当网络不稳定时,超时重连机制可以提高消息的发送成功率,但是SpringAMQP提供的重连机制时阻塞式的。不建议开启该功能,若业务需要,需要配置合理的等待时间和重试次数,也可以使用异步线程来执行发送消息的代码。

6.2 生产者确认机制

配置文件配置选项

spring:
  rabbitmq:
    publisher-confirm-type: correlated # 开启publisher confirm机制,并设置confirm类型
    # none:关闭confirm机制; simple:同步阻塞等待MQ的回执; correlated:MQ异步回调返回回执(推荐)
    publisher-returns: true # 开启publisher return机制
  1. Publisher Return 消息成功到达交换机,但是路由失败时会触发ReturnCallback,往往时编程导致的,可以避免

    @Configuration
    public class MqConfig implements ApplicationContextAware {
        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
            RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
            rabbitTemplate.setReturnCallback((message, code, text, exchange, key) -> {
                // 实现return callback
                System.err.println("【Return Call】 message: " + message + ", replyText: " + text);
            });
        }
    }
    
  2. Publisher Confirm

    • 消息投递到交换机,但是路由失败,触发ReturnCallback,返回ACK
    • 临时消息(不需要持久化)投递到交换机并入队成功,返回ACK
    • 持久化消息投递到交换机,入队成功并完成持久化,返回ACK
    • 其他情况返回NACK,标识投递失败
    @Test
    void contextLoads() {
        //  new CorrelationData(UUID.randomUUID().toString());
        CorrelationData cd = new CorrelationData();
        cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
            @Override
            public void onFailure(Throwable throwable) {
                // Future本身发生错误,一般不需要处理
            }
    
            @Override
            public void onSuccess(CorrelationData.Confirm confirm) {
                // Future处理成功
                if (confirm.isAck()) {
                    // 消息发送成功 ACK
                } else {
                    // 消息发送失败 NACK
                    // 执行消息发送失败的业务逻辑
                }
            }
        });
        String exchange = "";
        String routingKey = "";
        String msg = "";
        rabbitTemplate.convertAndSend(exchange, routingKey, msg, cd);
    }
    
  3. 总结
    生产者确认机制比较耗费资源,一般不开启,不开启确认每秒钟可以投递数万的消息,而开启后只能投递数千。若业务需要高可靠性,只需要开启Publisher Confirm处理NACK的情况即可。

6. MQ的可靠性

消息到达MQ以后,如果MQ不能及时保存,也会导致消息丢失。
- MQ宕机;
- 内存空间不足,引发MQ阻塞执行持久化;

6.1 数据持久化

  1. 交换机持久化(默认开启);
  2. 队列持久化(默认开启);
  3. 消息持久化(Delivery-mode需要指定为2,也就是持久化)
    - 若不开启消息持久化,在内存不足时,会发生MQ阻塞写磁盘PageOut;
    - 若开启消息持久化,会同步将消息写到磁盘,MQ不会出现阻塞的现象,速度稍微慢一点点。

6.2 惰性队列 Lazy Queue

  1. 从3.6.0开始支持,从3.12开始默认使用该策略
  2. 接收到消息后直接写入磁盘(内存默认只保留2048条),消息消费时才加载到内存,支持百万消息存储

7. 消费者的可靠性

当RabbitMQ向消费者投递消息以后,需要知道消费者的处理状态如何。因为消费者消费消息可能出现故障,比如:

  • 消息投递的过程中出现了网络故障
  • 消费者接收到消息后突然宕机
  • 消费者接收到消息后,因处理不当导致异常

7.1 消费者确认机制

RabbitMQ提供了消费者确认机制(Consumer Acknowledgement),当消费者处理消息后,向RabbitMQ发送一个回执,告知RabbitMQ自己消息处理状态。回执有三种可选值:

  • ack:成功处理消息,RabbitMQ从队列中删除该消息
  • nack:消息处理失败,RabbitMQ需要再次投递消息
  • reject:消息处理失败并拒绝该消息,RabbitMQ从队列中删除该消息

消息确认机制的实现方式

  • none:不处理。即消息投递给消费者后立刻ack,消息会立刻从MQ删除。非常不安全,不建议使用
  • manual:手动模式。需要自己在业务代码中调用api,发送ack或reject,存在业务入侵,但更灵活
  • auto:自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强,当业务正常执行时则自动返回ack. 当业务出现异常时,根据异常判断返回不同结果:
    • 如果是业务异常,会自动返回nack;
    • 如果是消息处理或校验异常,自动返回reject,比如发生MessageConversionException

7.2 失败重试机制

开启消费者确认机制后,如果消息处理一直返回NACK,那么消息会反复进行入队和处理,会导致MQ压力飙升。
而开启失败重试机制后,消息会在本地重试,而不是重新入队,本地重试达到最大次数后,默认会返回reject丢弃消息。
在消费者服务的配置文件中进行配置

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

7.3 失败处理策略

本地重试达到最大次数后,默认会返回reject丢弃消息,而有些业务显然无法接受消息的丢失。MQ支持之定义重试次数耗尽后的处理策略

  • RejectAndDontRequeueRecoverer:重试耗尽后,直接reject,丢弃消息,默认方式
  • ImmediateRequeueMessageRecoverer:重试耗尽后,返回nack,消息重新入队
  • RepublishMessageRecoverer:重试耗尽后,将失败消息投递到指定的交换机。(后续可进行人工处理)
    需要定义如下配置类
@Configuration
public class MqErrorConfig {
    private final static String ERROR_EXCHANGE = "error.direct";
    private final static String ERROR_QUEUE = "error.queue";
    private final static String ERROR_ROTING_KEY = "error";

    /**
     * 创建处理失败消息的交换机
     * @return
     */
    @Bean
    public DirectExchange errorExchange() {
        return new DirectExchange(ERROR_EXCHANGE);
    }

    /**
     * 创建存放失败消息的队列
     * @return
     */
    @Bean
    public Queue errorQueue() {
        return new Queue(ERROR_QUEUE);
    }

    /**
     * 交换机与队列绑定
     * @param errorQueue
     * @param errorExchange
     * @return
     */
    @Bean
    public Binding errorBinding(Queue errorQueue, DirectExchange errorExchange) {
        return BindingBuilder.bind(errorQueue).to(errorExchange).with(ERROR_ROTING_KEY);
    }

    /**
     * 注册处理失败消息处理策略
     * @param rabbitTemplate
     * @return
     */
    @Bean
    public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {
        return new RepublishMessageRecoverer(rabbitTemplate, ERROR_EXCHANGE, ERROR_ROTING_KEY);
    }
}

7.4 业务幂等性方案

在程序开发中,则是指同一个业务,执行一次或多次对业务状态的影响是一致的。

7.4.1 唯一消息ID

  • 每一条消息都生成一个唯一的id,与消息一起投递给消费者。
  • 消费者接收到消息后处理自己的业务,业务处理成功后将消息ID保存到数据库
  • 如果下次又收到相同消息,去数据库查询判断是否存在,存在则为重复消息放弃处理。
    进行如下配置,SpringAMQP会在消息头部自动添加唯一ID
@Bean
public MessageConverter messageConverter(){
    // 1.定义消息转换器
    Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
    // 2.配置自动创建消息id,用于识别不同消息,也可以在业务中基于ID判断是否是重复消息
    jjmc.setCreateMessageIds(true);
    return jjmc;
}

7.4.2 业务判断

非幂等性业务,会对数据进行更改,那么我们在执行业务逻辑前,可先判断数据记录是否处于未处理状态,比如可以根据订单的状态。

7.5 兜底策略

开启定时任务主动去查询数据库,判断数据有需要处理的数据。

8. 延迟消息

8.1 死信交换机

设计两个队列两个交换机,当消息过期时,消息会被投递到死信队列,只需监听死信队列即可。通过设置队列dead-letter-exchange指定过期的消息投递的交换机,也就是死信交换机。对于消息,通过expration指定过期时间。
然而,RabbitMQ的消息过期是基于追溯方式来实现的,也就是说当一个消息的TTL到期以后不一定会被移除或投递到死信交换机,而是在消息恰好处于队首时才会被处理。 当队列中消息堆积很多的时候,过期消息可能不会被按时处理,因此你设置的TTL时间不一定准确。

8.2 DelayExchange插件

开启队列的delayed配置,并且在投递消息时设置delay时长。
延迟消息插件内部会维护一个本地数据库表,同时使用Elang Timers功能实现计时。如果消息的延迟时间设置较长,可能会导致堆积的延迟消息非常多,会带来较大的CPU开销,同时延迟消息的时间会存在误差。 因此,不建议设置延迟时间过长的延迟消息。
改进策略,将消息的delay时长分段,比如将延迟时间切割成10s 10s 10s 15s 15s …,大部分消息在前30s内就已经可以被消费,不需要等到30分钟,可以有效防止消息堆积。

参考资料:https://www.bilibili.com/video/BV1mN4y1Z7t9/

点赞(0) 打赏

评论列表 共有 0 条评论

暂无评论

微信公众账号

微信扫一扫加关注

发表
评论
返回
顶部