史上最透彻的 RabbitMQ 可靠消息传输实战 - 后端 - 掘金

标签: | 发表时间:2019-08-08 10:00 | 作者:
出处:https://juejin.im

缓存架构之史上讲的最明白的RabbitMQ可靠消息传输实战演练

一、背景介绍:消息可靠传递的重要性

比如:某个广告主(如:天猫)想在我们的平台(如:今日头条)投放广告,当通过我们的广告系统新建广告的时候,该消息在同步给redis缓存(es)的时候丢失了,而我们又没有发现,造成该广告无法正常显示出来,那这损失就打了,如果1天都没有该广告的投放记录,那就有可能是上百万的损失了,所以消息的可靠传输多我们的广告系统也是很重要的。 其实,生活中这样的场景很场景,再比如:交易系统、订单系统都必须保证消息的可靠传输,否则,损失是巨大的!!!

二、如何保证消息的可靠传递呢?

1、设置交换机、队列和消息都为持久化

**持久化:**保证在服务器重启的时候可以保持不丢失相关信息,重点解决服务器的异常崩溃而导致的消息丢失问题。但是,将所有的消息都设置为持久化,会严重影响RabbitMQ的性能,写入硬盘的速度比写入内存的速度慢的不只一点点。对于可靠性不是那么高的消息可以不采用持久化处理以提高整体的吞吐率,在选择是否要将消息持久化时,需要在可靠性和吞吐量之间做一个权衡。 处于某种应用场景,如:大流量的订单交易系统,为了不影响性能,我们可以不设置持久化,但是我们会定时扫描数据库中的未发送成功的消息,进行重试发送,实际应用场景,我们其实有很多解决方案,不要故步自封,换个角度多想想,只有经历多了,才能应用的更加得心应手。

1)交换机的持久化

      @Bean
DirectExchange advanceExchange() {
    return new DirectExchange(exchangeName);
}

注释:查看源码,易知,默认是持久化的

2)队列的持久化

      @Bean
public Queue advanceQueue() {
    return new Queue(queueName);
}

注释:查看源码,易知,默认是持久化的

3)消息的持久化

当我们使用RabbitTemplate调用了 convertAndSend(String exchange, String routingKey, final Object object) 方法。默认就是持久化模式

注意:

  • 持久化的消息在到达队列时就被写入到磁盘,并且如果可以,持久化的消息也会在内存中保存一份备份,这样可以提高一定的性能,只有在内存吃紧的时候才会从内存中清楚。
  • 非持久化的消息一般只保存在内存中,在内存吃紧的时候会被换入到磁盘中,以节省内存空间。

2、生产者消息确认机制

当消息发送出去之后,我们如何知道消息有没有正确到达exchange呢?如果在这个过程中,消息丢失了,我们根本不知道发生了什么,也不知道是什么原因导致消息发送失败了 为解决这个问题,主要有如下两种方案:

  • 通过事务机制实现
  • 通过生产者消息确认机制(publisher confirm)实现

但是使用 事务机制实现会严重降低RabbitMQ的消息吞吐量,我们采用一种轻量级的方案—— 生产者消息确认机制

什么是消息确认机制? 简而言之,就是:生产者发送的消息一旦被投递到所有匹配的队列之后,就会发送一个确认消息给生产者,这就使得生产者知晓消息已经正确到达了目的地。 如果消息和队列是持久化存储的,那么确认消息会在消息写入磁盘之后发出。 再补充一个 Mandatory参数:当Mandatory参数设为true时,如果目的不可达,会发送消息给生产者,生产者通过一个回调函数来获取该信息。

3、消费者消息确认机制

为了保证消息从队列可靠地到达消费者,RabbitMQ提供了消费者消息确认机制(message acknowledgement)。采用消息确认机制之后,消费者就有足够的时间来处理消息,不用担心处理消息过程中消费者进程挂掉后消息丢失的问题,因为RabbitMQ会一直等待并持有消息,直到消费者确认了该消息。

4、死信队列

DLX,Dead Letter Exchange 的缩写,又死信邮箱、死信交换机。DLX就是一个普通的交换机,和一般的交换机没有任何区别。 当消息在一个队列中变成死信(dead message)时,通过这个交换机将死信发送到死信队列中(指定好相关参数,rabbitmq会自动发送)。

什么是死信呢?什么样的消息会变成死信呢?

  • 消息被拒绝(basic.reject或basic.nack)并且requeue=false.
  • 消息TTL过期
  • 队列达到最大长度(队列满了,无法再添加数据到mq中)

应用场景分析:在定义业务队列的时候,可以考虑指定一个死信交换机,并绑定一个死信队列,当消息变成死信时,该消息就会被发送到该死信队列上,这样就方便我们查看消息失败的原因了 **如何使用死信交换机呢?

定义业务(普通)队列的时候指定参数:

  • x-dead-letter-exchange: 用来设置死信后发送的交换机
  • x-dead-letter-routing-key:用来设置死信的routingKey
      @Bean
public Queue helloQueue() {
    //将普通队列绑定到私信交换机上
    Map<String, Object> args = new HashMap<>(2);
    args.put(DEAD_LETTER_QUEUE_KEY, deadExchangeName);
    args.put(DEAD_LETTER_ROUTING_KEY, deadRoutingKey);
    Queue queue = new Queue(queueName, true, false, false, args);
    return queue;
}

三、实战演练

项目代码下载地址gitee.com/jikeh/JiKeH… 项目名:spring-boot-rabbitmq-reliability

1、开启生产者消息确认机制

      # 开启发送确认
spring.rabbitmq.publisher-confirms=true
# 开启发送失败退回
spring.rabbitmq.publisher-returns=true

2、开启消费者消息确认机制

      # 开启ACK
spring.rabbitmq.listener.simple.acknowledge-mode=manual

3、基本配置

      @Configuration
public class RabbitConfig {

    public final static String queueName = "hello_queue";

    /**
     * 死信队列:
     */
    public final static String deadQueueName = "dead_queue";
    public final static String deadRoutingKey = "dead_routing_key";
    public final static String deadExchangeName = "dead_exchange";

    /**
     * 死信队列 交换机标识符
     */
    public static final String DEAD_LETTER_QUEUE_KEY = "x-dead-letter-exchange";
    /**
     * 死信队列交换机绑定键标识符
     */
    public static final String DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";

    @Bean
    public Queue helloQueue() {
        //将普通队列绑定到私信交换机上
        Map<String, Object> args = new HashMap<>(2);
        args.put(DEAD_LETTER_QUEUE_KEY, deadExchangeName);
        args.put(DEAD_LETTER_ROUTING_KEY, deadRoutingKey);
        Queue queue = new Queue(queueName, true, false, false, args);
        return queue;
    }

    /**
     * 死信队列:
     */

    @Bean
    public Queue deadQueue() {
        Queue queue = new Queue(deadQueueName, true);
        return queue;
    }

    @Bean
    public DirectExchange deadExchange() {
        return new DirectExchange(deadExchangeName);
    }

    @Bean
    public Binding bindingDeadExchange(Queue deadQueue, DirectExchange deadExchange) {
        return BindingBuilder.bind(deadQueue).to(deadExchange).with(deadRoutingKey);
    }

}

注释:hell_queue就配置了死信交换机、死信队列

4、生产者核心代码

      @Component
public class HelloSender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void send(String exchange, String routingKey) {
        String context = "你好现在是 " + new Date();
        System.out.println("send content = " + context);
        this.rabbitTemplate.setMandatory(true);
        this.rabbitTemplate.setConfirmCallback(this);
        this.rabbitTemplate.setReturnCallback(this);
        this.rabbitTemplate.convertAndSend(exchange, routingKey, context);
    }

    /**
     * 确认后回调:
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (!ack) {
            System.out.println("send ack fail, cause = " + cause);
        } else {
            System.out.println("send ack success");
        }
    }

    /**
     * 失败后return回调:
     *
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("send fail return-message = " + new String(message.getBody()) + ", replyCode: " + replyCode + ", replyText: " + replyText + ", exchange: " + exchange + ", routingKey: " + routingKey);
    }
}

5、消费者核心代码

      @Component
@RabbitListener(queues = RabbitConfig.queueName)
public class HelloReceiver {

    @RabbitHandler
    public void process(String hello, Channel channel, Message message) throws IOException {
        try {
            Thread.sleep(2000);
            System.out.println("睡眠2s");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        try {
            //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉;否则消息服务器以为这条消息没处理掉 后续还会在发
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            System.out.println("receiver success = " + hello);
        } catch (Exception e) {
            e.printStackTrace();
            //丢弃这条消息
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
            System.out.println("receiver fail");
        }

    }

}

6、测试生产者消息确认功能:分为4种场景来测试

      //1、exchange, queue 都正确, confirm被回调, ack=true
@RequestMapping("/send1")
@ResponseBody
public String send1() {
    helloSender.send(null, RabbitConfig.queueName);
    return "success";
}

//2、exchange 错误, queue 正确, confirm被回调, ack=false
@RequestMapping("/send2")
@ResponseBody
public String send2() {
    helloSender.send("fail-exchange", RabbitConfig.queueName);
    return "success";
}

//3、exchange 正确, queue 错误, confirm被回调, ack=true; return被回调 replyText:NO_ROUTE
@RequestMapping("/send3")
@ResponseBody
public String send3() {
    helloSender.send(null, "fail-queue");
    return "success";
}

//4、exchange 错误, queue 错误, confirm被回调, ack=false
@RequestMapping("/send4")
@ResponseBody
public String send4() {
    helloSender.send("fail-exchange", "fail-queue");
    return "success";
}

7、测试消费者消息确认功能

  • 1)当添加这行代码的时候: channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);测试结果:消息被正常消费,消息从队列中删除
  • 2)当注释掉这行代码的时候: channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);测试结果:消息会被重复消费,一直保留在队列当中

8、测试死信队列

当执行这行代码的时候: channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);消息会被加入到死信队列中:

四、拓展

除了我们上面讲的基本可靠性保证外,其实还有很多性能优化方案、可靠性保证方案:集群监控、流控、镜像队列、HAProxy+Keeplived高可靠负载均衡 我们后续会继续分享上述内容,欢迎持续关注…… 下节课,我们将会将该功能应用到缓存架构上了


相关 [rabbitmq 可靠 消息] 推荐:

RabbitMQ高级之如何保证消息可靠性?

- - SegmentFault 最新的文章
人生终将是场单人旅途,孤独之前是迷茫,孤独过后是成长. 本篇是消息队列 RabbitMQ的第四弹. RabbitMQ我已经写了三篇了,基础的收发消息和基础的概念我都已经写了,学任何东西都是这样,先基础的上手能用,然后遇到问题再去解决,无法理解就去深入源码,随着时间的积累对这一门技术的理解也会随之提高.

史上最透彻的 RabbitMQ 可靠消息传输实战 - 后端 - 掘金

- -
缓存架构之史上讲的最明白的RabbitMQ可靠消息传输实战演练. 一、背景介绍:消息可靠传递的重要性. 比如:某个广告主(如:天猫)想在我们的平台(如:今日头条)投放广告,当通过我们的广告系统新建广告的时候,该消息在同步给redis缓存(es)的时候丢失了,而我们又没有发现,造成该广告无法正常显示出来,那这损失就打了,如果1天都没有该广告的投放记录,那就有可能是上百万的损失了,所以消息的可靠传输多我们的广告系统也是很重要的.

消息中间件的技术选型心得-RabbitMQ、ActiveMQ和ZeroMQ

- - haohtml's blog
RabbitMQ、ActiveMQ和ZeroMQ都是极好的消息中间件,但是我们在项目中该选择哪个更适合呢. 下面我会对这三个消息中间件做一个比较,看了后你们就心中有数了. RabbitMQ是AMQP协议领先的一个实现,它实现了代理(Broker)架构,意味着消息在发送到客户端之前可以在中央节点上排队.

rabbitmq消息去重及防丢失解决方案

- - fatboa
个人笔记,如有描述不当,欢迎留言指出~. 我们知道一个电商项目里时刻都有海量的消息通知,比如顾客注册通知、签到通知、下单通知等等,而我们公司的电商项目更加复杂,包含了客户端、门店端以及供应商端三端,各种各样的消息通知游走在各个服务模块间. 如果每个模块都要实现一套消息通知的功能,那无疑是多余的. 所以我把各模块的消息功能提取出来独立成一个服务模块,就像一个快递员,把各模块的消息准确投递至各端.

消息中间件选型分析——从Kafka与RabbitMQ的对比来看全局

- - 程序猿DD
有很多网友留言:公司要做消息中间件选型,该如何选. 消息选型的确是一个大论题,实则说来话长的事情又如何长话短说. 对此笔者专门撰稿一篇内功心法: 如何看待消息中间件的选型,不过这篇只表其意未表其行,为了弥补这种缺陷,笔者最近特意重新撰稿一篇,以供参考. 温馨提示:本文一万多字,建议先马(关注)后看.

【架构】关于RabbitMQ

- - 学习笔记
1      什么是RabbitMQ. RabbitMQ是实现AMQP(高级消息队列协议)的消息中间件的一种,最初起源于金融系统,用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗. 消息中间件主要用于组件之间的解耦,消息的发送者无需知道消息使用者的存在,反之亦然:. 例如一个日志系统,很容易使用RabbitMQ简化工作量,一个Consumer可以进行消息的正常处理,另一个Consumer负责对消息进行日志记录,只要在程序中指定两个Consumer所监听的queue以相同的方式绑定到同一个exchange即可,剩下的消息分发工作由RabbitMQ完成.

Kafka 的消息可靠传递

- - IT瘾-dev
Kafka提供的基础保障可以用来构建可靠的系统, 却无法保证完全可靠. 需要在可靠性和吞吐之间做取舍.. Kafka在分区上提供了消息的顺序保证.. 生产的消息在写入到所有的同步分区上后被认为是. 生产者可以选择在消息提交完成后接收broker的确认, 是写入leader之后, 或者所有的副本. 只要有一个副本存在, 提交的消息就不会丢失.

RabbitMQ (三) 发布/订阅

- - CSDN博客架构设计推荐文章
转发请标明出处: http://blog.csdn.net/lmj623565791/article/details/37657225. 本系列教程主要来自于官网入门教程的翻译,然后自己进行了部分的修改与实验,内容仅供参考. 上一篇博客中,我们实现了工作队列,并且我们的工作队列中的一个任务只会发给一个工作者,除非某个工作者未完成任务意外被杀死,会转发给另外的工作者,如果你还不了解: RabbitMQ (二)工作队列.

rabbitmq java client api详解

- - 五四陈科学院
以下内容由 [五四陈科学院]提供. AMQP协议是一个高级抽象层消息通信协议,RabbitMQ是AMQP协议的实现. 每个rabbitmq-server叫做一个Broker,等着tcp连接进入. 在rabbitmq-server进程内有Exchange,定义了这个消息的发送类型. Queue是进程内的逻辑队列,有多个,有名字.