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

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

2018-09-26 594

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

比如：某个广告主（如：天猫）想在我们的平台（如：今日头条）投放广告，当通过我们的广告系统新建广告的时候，该消息在同步给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;
}

三、实战演练

项目代码下载地址：https://gitee.com/jikeh/JiKeHCN-RELEASE.git 项目名：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高可靠负载均衡我们后续会继续分享上述内容，欢迎持续关注…… 下节课，我们将会将该功能应用到缓存架构上了

本文来自云栖社区合作伙伴“开源中国”

本文作者：王练

原文链接

微信关注我们

原文链接：https://yq.aliyun.com/articles/645790

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

Kubeless —— Kubernetes 原生 Serverless 框架

Kubeless详细介绍 Kubeless 是一个基于 Kubernetes 的 Serverless 框架，允许您部署少量代码，而无需担心底层基础架构管道。它利用 Kubernetes 资源提供自动扩展、API 路由、监控、故障排除等功能。 Kuberless 包含以下部分：持 Python、Node.js、Ruby、PHP、Golang、.NET、Ballerina 和自定义运行时 CLI 兼容 AWS Lambda CLI 事件触发器使用 Kafka 消息传递系统和 HTTP 事件 Prometheus 默认监控函数调用和函数延迟 ServerlessFramework 插件

2018-09-27

782

一、前言民生银行 IT 运维工作经历了多年实践，已经建设了CMDB、IT运维管理系统（流程平台）、集中监控系统、交易性能监控系统、自动化运维系统、日志管理平台等管理工具，并在实际工作中不断深入的优化，在近年还打造了运维大数据平台，用以支撑 IT 运维管理工作。在日常工作中，监（各类监控）、管（流程）、控（自动化）和CMDB系统均建立了映射关系，用以打通各系统的数据消费场景。但实际工作中依然面临着工具分散，依靠运维人员经验和频繁切换各专业分析工具，以实现故障定位、影响分析等操作，运维数据消费效率存在进步空间。二、建设思路与成果基于上述背景，民生银行尝试借助架构管理可视化工具，将配置数据（CMDB）、监控数据（集中监控告警、交易性能监控）、自动化运维管理工具、IT运维管理系统的变更数据整合到 IT 运维架构图上，打造统一的运维数据消费

2018-09-27

758

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。