中间件相关问题整理

不念不忘少年蓝@ 2023-07-23 05:54 35阅读 0赞

**目录**

1.用过哪些MQ，怎么用的，和其他mq比较有什么优缺点，MQ的连接是线程安全的吗？

一、常见的消息中间件

1.Kafka主要特性、优缺点

2.RabbitMQ主要特性、优缺点

3.RocketMQ主要特性、优缺点

4.ActiveMQ主要特性、优缺点

二、消息中间件相关知识

1.消息中间件的组成

2.消息中间件模式分类

点对点

3.消息中间件的普遍优势

4.消息中间件常用协议

三、消息队列应用场景

场景一：异步处理

场景二：应用解耦

场景三：流量削锋（一般在秒杀或团抢活动中使用广泛）

场景四：日志处理

场景五：消息通讯

2.MQ系统的数据如何保证不丢失？

一、MQ原则

二、丢失数据场景分析

（一）rabbitmq丢失数据场景分析

1.生产者弄丢了数据

2.rabbitmq自己丢了数据

3.消费端弄丢了数据

（二）kafka丢失数据场景分析

三、如何防止消息丢失

（一）rabbitmq防止消息丢失

1.生产者丢失消息

2.rabbitmq自己弄丢了数据

3.消费者弄丢了数据

（二）kafka防止消息丢失

1.消费端弄丢了数据

2.kafka弄丢了数据

3.生产者弄丢了数据

补充问题：

关于MQ的几件小事（五）如何保证消息按顺序执行

关于MQ的几件小事（三）如何保证消息不重复消费

关于MQ的几件小事（二）如何保证消息队列的高可用

Kafka的文件存储机制

Kafka 如何保证可靠性

Kafka消息是采用Pull模式，还是Push模式

Kafka是如何实现高吞吐率的

Kafka判断一个节点还活着的两个条件

参考书籍、文献和资料

--------------------

备注：针对基本问题做一些基本的总结，不是详细解答！

# 1.用过哪些MQ，怎么用的，和其他mq比较有什么优缺点，MQ的连接是线程安全的吗？ #

## 一、常见的消息中间件 ##

消息队列中间件是分布式系统中重要的组件，**主要解决应用耦合、异步消息、流量削锋等问题，实现高性能、高可用、可伸缩和最终一致性架构。**

当前使用较多的消息队列有RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMQ等，而部分数据库如Redis、MySQL以及phxsql也可实现消息队列的功能。

### 1.Kafka主要特性、优缺点 ###

Apache Kafka是一个分布式消息发布订阅系统。Kafka性能高效、可扩展良好并且可持久化。它的分区特性、可复制和可容错都是其不错的特性。

**主要特性**

*  快速持久化：可以在O(1)的系统开销下进行消息持久化；
 *  高吞吐：在一台普通的服务器上既可以达到10W/s的吞吐速率；
 *  **完全的分布式系统：Broker、Producer和Consumer都原生自动支持分布式，自动实现负载均衡**；
 *  支持同步和异步复制两种高可用机制；
 *  **支持数据批量发送和拉取**；
 *  **零拷贝技术(zero-copy)：减少IO操作步骤，提高系统吞吐量**；
 *  数据迁移、扩容对用户透明；
 *  无需停机即可扩展机器；
 *  **其他特性：丰富的消息拉取模型、高效订阅者水平扩展、实时的消息订阅、亿级的消息堆积能力、定期删除机制**；

**优点**

*  **客户端语言丰富：支持Java、.Net、PHP、Ruby、Python、Go等多种语言**；
 *  高性能：单机写入TPS约在100万条/秒，消息大小10个字节；
 *  提供完全分布式架构，并有replica机制，拥有较高的可用性和可靠性，理论上支持消息无限堆积；
 *  支持批量操作；
 *  消费者采用Pull方式获取消息。消息有序，通过控制能够保证所有消息被消费且仅被消费一次；
 *  **有优秀的第三方KafkaWeb管理界面Kafka-Manager；**
 *  **在日志领域比较成熟，被多家公司和多个开源项目使用**。

**缺点**

*  Kafka**单机超过64个队列/分区时，Load时会发生明显的飙高现象**。队列越多，负载越高，发送消息响应时间变长；
 *  使用短轮询方式，实时性取决于轮询间隔时间；
 *  **消费失败不支持重试**；
 *  **支持消息顺序，但是一台代理宕机后，就会产生消息乱序**；
 *  社区更新较慢。

### 2.RabbitMQ主要特性、优缺点 ###

RabbitMQ于2007年发布，是一个在**AMQP(高级消息队列协议)**基础上完成的，可复用的企业消息系统，是当前最主流的消息中间件之一。

**主要特性**

*  **可靠性：提供了多种技术可以让你在性能和可靠性之间进行权衡**。这些技术包括持久性机制、投递确认、发布者证实和高可用性机制；
 *  **灵活的路由：消息在到达队列前是通过交换机进行路由的。**RabbitMQ为典型的路由逻辑提供了多种内置交换机类型。如果你有更复杂的路由需求，可以将这些交换机组合起来使用，你甚至可以实现自己的交换机类型，并且当做RabbitMQ的插件来使用；
 *  消息集群：在相同局域网中的多个RabbitMQ服务器可以聚合在一起，作为一个独立的逻辑代理来使用；
 *  队列高可用：队列可以在集群中的机器上进行镜像，以确保在硬件问题下还保证消息安全；
 *  支持多种协议：支持多种消息队列协议；
 *  支持多种语言：用Erlang语言编写，支持只要是你能想到的所有编程语言；
 *  **管理界面：RabbitMQ有一个易用的用户界面，使得用户可以监控和管理消息Broker的许多方面；**
 *  跟踪机制：如果消息异常，RabbitMQ 提供消息跟踪机制，使用者可以找出发生了什么；
 *  插件机制：提供了许多插件，来从多方面进行扩展，也可以编写自己的插件。

**优点**

*  由于Erlang语言的特性，消息队列性能较好，支持高并发；
 *  健壮、稳定、易用、跨平台、支持多种语言、文档齐全；
 *  **有消息确认机制和持久化机制，可靠性高；**
 *  **高度可定制的路由；**
 *  **管理界面较丰富，在互联网公司也有较大规模的应用，社区活跃度高。**

**缺点**

*  尽管结合 Erlang 语言本身的并发优势，性能较好，但是**不利于做二次开发和维护**；
 *  **实现了代理架构，意味着消息在发送到客户端之前可以在中央节点上排队。此特性使得RabbitMQ易于使用和部署，但是使得其运行速度较慢，因为中央节点增加了延迟，消息封装后也比较大；需要学习比较复杂的接口和协议，学习和维护成本较高。**

### 3.RocketMQ主要特性、优缺点 ###

**RocketMQ出自阿里的开源产品，用Java语言实现，在设计时参考了Kafka，并做出了自己的一些改进，消息可靠性上比Kafka更好。**RocketMQ在阿里内部被广泛应用在订单，交易，充值，流计算，消息推送，日志流式处理，binglog分发等场景。

**主要特性**

*  基于**队列模型**：具有高性能、高可靠、高实时、分布式等特点；
 *  **Producer、Consumer、队列都支持分布式**；
 *  **Producer向一些队列轮流发送消息，队列集合称为Topic**。Consumer如果做广播消费，则一个Consumer实例消费这个Topic对应的所有队列；如果做集群消费，则多个Consumer 实例平均消费这个Topic对应的队列集合；
 *  能够**保证严格的消息顺序**；
 *  提供**丰富的消息拉取模式**；
 *  **高效的订阅者水平扩展能力**；
 *  **实时的消息订阅机制**；
 *  **亿级消息堆积能力**；
 *  **较少的外部依赖。**

**优点**

*  **单机支持1万以上持久化队列**；
 *  RocketMQ的**所有消息都是持久化的**，先写入**系统PAGECACHE**，然后**刷盘**，可以保证内存与磁盘都有一份数据，而访问时，直接从内存读取；
 *  **模型简单，接口易用（**JMS的接口很多场合并不太实用）；
 *  性能非常好，可以**允许大量堆积消息在Broker中**；
 *  支持**多种消费模式，包括集群消费、广播消费等**；
 *  **各个环节分布式扩展设计，支持主从和高可用**；
 *  **开发度较活跃，版本更新很快**。

**缺点**

*  **支持的客户端语言不多，目前是Java及C++**，其中C++还不成熟；
 *  **RocketMQ社区关注度及成熟度也不及前两者**；
 *  没有Web管理界面，提供了一个 CLI (命令行界面) 管理工具带来查询、管理和诊断各种问题；
 *  没有在MQ核心里实现JMS等接口；

### 4.ActiveMQ主要特性、优缺点 ###

ActiveMQ是由Apache出品，ActiveMQ是一个**完全支持JMS1.1和J2EE 1.4规范的JMS Provider实现**。它非常快速，支持多种语言的客户端和协议，而且可以非常容易的嵌入到企业的应用环境中，并有许多高级功能。

**主要特性**

*  服从JMS规范：JMS 规范提供了良好的标准和保证，包括：同步 或 异步 的消息分发，一次和仅一次的消息分发，消息接收和订阅等等。遵从JMS规范的好处在于，不论使用什么JMS实现提供者，这些基础特性都是可用的；
 *  连接灵活性：ActiveMQ提供了广泛的连接协议，支持的协议有：HTTP/S，IP多播，SSL，TCP，UDP等等。对众多协议的支持让ActiveMQ拥有了很好的灵活性；
 *  支持的协议种类多：OpenWire、STOMP、REST、XMPP、AMQP；
 *  持久化插件和安全插件：ActiveMQ提供了多种持久化选择。而且，ActiveMQ的安全性也可以完全依据用户需求进行自定义鉴权和授权；
 *  支持的客户端语言种类多：除了Java之外，还有：C/C++，.NET，Perl，PHP，Python，Ruby；
 *  代理集群：多个ActiveMQ代理可以组成一个集群来提供服务；
 *  异常简单的管理：ActiveMQ是以开发者思维被设计的。所以，它并不需要专门的管理员，因为它提供了简单又使用的管理特性。有很多中方法可以监控ActiveMQ不同层面的数据，包括使用在JConsole或者在ActiveMQ的WebConsole中使用JMX。通过处理JMX的告警消息，通过使用命令行脚本，甚至可以通过监控各种类型的日志。

**优点**

*  跨平台(JAVA编写与平台无关，ActiveMQ几乎可以运行在任何的JVM上)；
 *  可以用JDBC：可以将数据持久化到数据库。虽然使用JDBC会降低ActiveMQ的性能，但是数据库一直都是开发人员最熟悉的存储介质；
 *  支持JMS规范：支持JMS规范提供的统一接口;
 *  支持自动重连和错误重试机制；
 *  有安全机制：支持基于shiro，jaas等多种安全配置机制，可以对Queue/Topic进行认证和授权；
 *  监控完善：拥有完善的监控，包括WebConsole，JMX，Shell命令行，Jolokia的RESTful API；
 *  **界面友善：提供的WebConsole可以满足大部分情况，还有很多第三方的组件可以使用，比如hawtio**；

缺点

*  社区活跃度不及RabbitMQ高；
 *  根据其他用户反馈，会出莫名其妙的问题，会丢失消息；
 *  目前重心放到activemq6.0产品Apollo，对5.x的维护较少；
 *  **不适合用于上千个队列的应用场景**；

## 二、消息中间件相关知识 ##

### 1.消息中间件的组成 ###

*  **Broker**：消息服务器，作为server提供消息核心服务
 *  **Producer**：消息生产者，业务的发起方，负责生产消息传输给broker
 *  **Consumer**：消息消费者，业务的处理方，负责从broker获取消息并进行业务逻辑处理
 *  **Topic**：主题，发布订阅模式下的消息统一汇集地，不同生产者向topic发送消息，由MQ服务器分发到不同的订阅者，实现消息的广播
 *  **Queue**：队列，PTP模式下，特定生产者向特定queue发送消息，消费者订阅特定的queue完成指定消息的接收
 *  **Message**：消息体，根据不同通信协议定义的固定格式进行编码的数据包，来封装业务数据，实现消息的传输

### 2.消息中间件模式分类 ###

*  ### **点对点** ###

PTP点对点:**使用queue作为通信载体**

消息生产者生产消息发送到queue中，然后消息消费者从queue中取出并且消费消息。  
消息被消费以后，queue中不再存储，所以消息消费者不可能消费到已经被消费的消息。 Queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70][]

*  **发布/订阅**

Pub/Sub发布订阅（广播）：**使用topic作为通信载体**

消息生产者（发布）将消息发布到topic中，同时有多个消息消费者（订阅）消费该消息。和点对点方式不同，发布到topic的消息会被所有订阅者消费。

queue实现了负载均衡，将producer生产的消息发送到消息队列中，由多个消费者消费。但一个消息只能被一个消费者接受，当没有消费者可用时，这个消息会被保存直到有一个可用的消费者。  
topic实现了发布和订阅，当你发布一个消息，所有订阅这个topic的服务都能得到这个消息，所以**从1到N个订阅者都能得到一个消息的拷贝**。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 1][]

### 3.消息中间件的普遍优势 ###

*  **系统解耦**

交互系统之间没有直接的调用关系，只是通过消息传输，故系统侵入性不强，耦合度低。

*  **提高系统响应时间**

例如原来的一套逻辑，完成支付可能涉及先修改订单状态、计算会员积分、通知物流配送几个逻辑才能完成；**通过MQ架构设计，就可将紧急重要（需要立刻响应）的业务放到该调用方法中，响应要求不高的使用消息队列，放到MQ队列中，供消费者处理**。

*  **为大数据处理架构提供服务**

通过消息作为整合，**大数据的背景下，消息队列还与实时处理架构整合，为数据处理提供性能支持**。

*  **Java消息服务——JMS**

Java消息服务（Java Message Service，JMS）应用程序接口是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。  
JMS中的P2P和Pub/Sub消息模式：点对点（point to point， queue）与发布订阅（publish/subscribe，topic）最初是由JMS定义的。这两种模式主要区别或解决的问题就是发送到队列的消息能否重复消费(多订阅)。

### 4.消息中间件常用协议 ###

*  **AMQP协议**

AMQP即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同开发语言等条件的限制。  
优点：可靠、通用

*  **MQTT协议**

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输）是IBM开发的一个即时通讯协议，有可能成为物联网的重要组成部分。该协议支持所有平台，几乎可以把所有联网物品和外部连接起来，被用来当做传感器和致动器（比如通过Twitter让房屋联网）的通信协议。  
优点：格式简洁、占用带宽小、移动端通信、PUSH、嵌入式系统

*  **STOMP协议**

STOMP（Streaming Text Orientated Message Protocol）是流文本定向消息协议，是一种为MOM(Message Oriented Middleware，面向消息的中间件)设计的简单文本协议。STOMP提供一个可互操作的连接格式，允许客户端与任意STOMP消息代理（Broker）进行交互。  
优点：命令模式（非topic\\queue模式）

*  **XMPP协议**

XMPP（可扩展消息处理现场协议，Extensible Messaging and Presence Protocol）是基于可扩展标记语言（XML）的协议，多用于即时消息（IM）以及在线现场探测。适用于服务器之间的准即时操作。核心是基于XML流传输，这个协议可能最终允许因特网用户向因特网上的其他任何人发送即时消息，即使其操作系统和浏览器不同。  
优点：通用公开、兼容性强、可扩展、安全性高，但XML编码格式占用带宽大

## 三、**消息队列应用场景** ##

在实际应用中常用的使用场景：异步处理、应用解耦、流量削锋和消息通讯四个场景

### 场景一：异步处理 ###

**场景说明：用户注册后，需要发注册邮件和注册短信**。传统的做法有两种 1.串行的方式；2.并行方式

*  串行方式：将注册信息写入数据库成功后，发送注册邮件，再发送注册短信。以上三个任务全部完成后，返回给客户端。
 *  并行方式：将注册信息写入数据库成功后，发送注册邮件的同时，发送注册短信。以上三个任务完成后，返回给客户端。与串行的差别是，并行的方式可以提高处理的时间。

假设三个业务节点每个使用50毫秒钟，不考虑网络等其他开销，则串行方式的时间是150毫秒，并行的时间可能是100毫秒。

因为CPU在单位时间内处理的请求数是一定的，假设CPU1秒内吞吐量是100次。则串行方式1秒内CPU可处理的请求量是7次（1000/150），并行方式处理的请求量是10次（1000/100）

小结：如以上案例描述，传统的方式系统的性能（并发量，吞吐量，响应时间）会有瓶颈。如何解决这个问题呢？

引入消息队列，**将不是必须的业务逻辑，异步处理**。改造后的架构如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 2][]

按照以上约定，用户的响应时间相当于是注册信息写入数据库的时间，也就是50毫秒。注册邮件，发送短信写入消息队列后，直接返回，因此写入消息队列的速度很快，基本可以忽略，因此用户的响应时间可能是50毫秒。因此**架构改变后，系统的吞吐量提高到每秒20 QPS。比串行提高了3倍，比并行提高了两倍。**

### **场景二：应用解耦** ###

场景说明：用户下单后，订单系统需要通知库存系统。传统的做法是，订单系统调用库存系统的接口。假如库存系统无法访问，则订单减库存将失败，从而导致订单失败，缺点在于订单系统与库存系统耦合。

如何解决以上问题呢？引入应用消息队列后的方案，如下图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 3][]

*  订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功
 *  库存系统：**订阅下单的消息，采用拉/推的方式，获取下单信息，库存系统根据下单信息，进行库存操作**

假如：在下单时库存系统不能正常使用，也不影响正常下单，因为下单后，订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦。

### 场景三：流量削锋（一般在秒杀或团抢活动中使用广泛） ###

应用场景：秒杀活动，一般会因为流量过大，导致流量暴增，应用挂掉。为解决这个问题，一般需要在应用前端加入消息队列：可以控制活动的人数、可以缓解短时间内高流量压垮应用。**用户的请求，服务器接收后，首先写入消息队列。假如消息队列长度超过最大数量，则直接抛弃用户请求或跳转到错误页面**，**秒杀业务根据消息队列中的请求信息，再做后续处理。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 4][]

### **场景四：日志处理** ###

日志处理是指将消息队列用在日志处理中，比如Kafka的应用，解决大量日志传输的问题。架构简化如下，

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 5][]

*  日志采集客户端：负责日志数据采集，定时写受写入Kafka队列
 *  Kafka消息队列：负责日志数据的接收，存储和转发
 *  日志处理应用：订阅并消费kafka队列中的日志数据

查看新浪kafka日志处理应用案例：转自（http://cloud.51cto.com/art/201507/484338.htm）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 6][]

*  (1)Kafka：接收用户日志的消息队列
 *  (2)Logstash：做日志解析，统一成JSON输出给Elasticsearch
 *  (3)Elasticsearch：实时日志分析服务的核心技术，一个schemaless，实时的数据存储服务，通过index组织数据，兼具强大的搜索和统计功能
 *  (4)Kibana：基于Elasticsearch的数据可视化组件，超强的数据可视化能力是众多公司选择ELK stack的重要原因

### **场景五：消息通讯** ###

消息通讯是指，消息队列一般都内置了高效的通信机制，因此也可以用在纯的消息通讯。比如实现点对点消息队列，或者聊天室等

*  点对点通讯：客户端A和客户端B使用同一队列，进行消息通讯。

![20181219192323196][]

*  聊天室通讯：客户端A，客户端B，客户端N订阅同一主题，进行消息发布和接收。实现类似聊天室效果。

![20181219192323212][]

以上实际是消息队列的两种消息模式，点对点或发布订阅模式。模型为示意图，供参考。

# 2.MQ系统的数据如何保证不丢失？ #

## 一、MQ原则 ##

数据不能多，也不能少，不能多是说消息不能重复消费。不能少，就是说不能丢失数据。如果mq传递的是非常核心的消息，支撑核心的业务，那么这种场景是一定不能丢失数据的。

## 二、丢失数据场景分析 ##

丢数据一般分为两种，一种是mq把消息丢了，一种就是消费时将消息丢了。

下面从rabbitmq和kafka分别说一下，丢失数据的场景：

### （一）rabbitmq丢失数据场景分析 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 7][]

### **1.生产者弄丢了数据** ###

生产者将数据发送到rabbitmq的时候，可能在传输过程中因为网络等问题而将数据弄丢了。

### **2.rabbitmq自己丢了数据** ###

如果没有开启rabbitmq的持久化，那么rabbitmq一旦重启，那么数据就丢了。所以**必须开启持久化将消息持久化到磁盘，这样就算rabbitmq挂了，恢复之后会自动读取之前存储的数据，一般数据不会丢失**。除非极其罕见的情况，rabbitmq还没来得及持久化自己就挂了，这样可能导致一部分数据丢失。

### **3.消费端弄丢了数据** ###

主要是**因为消费者消费时，刚消费到，还没有处理，结果消费者就挂了，这样你重启之后，rabbitmq就认为你已经消费过了，然后就丢了数据。**

### （二）kafka丢失数据场景分析 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 8][]

**1.生产者弄丢了数据**

**生产者没有设置相应的策略，发送过程中丢失数据**。

**2.kafka弄丢了数据**

**比较常见的一个场景，就是kafka的某个broker宕机了，然后重新选举partition的leader时**。如果此时follower还没来得及同步数据，leader就挂了，然后某个follower成为了leader，他就少了一部分数据。

**3.消费者弄丢了数据**

**消费者消费到了这个数据，然后消费之自动提交了offset，让kafka知道你已经消费了这个消息，当你准备处理这个消息时，自己挂掉了，那么这条消息就丢了**。

## 三、如何防止消息丢失 ##

### （一）rabbitmq防止消息丢失 ###

### **1.生产者丢失消息** ###

*  **可以选择使用rabbitmq提供是事物功能**，

就是生产者在发送数据之前开启事物，然后发送消息，如果消息没有成功被rabbitmq接收到，那么生产者会受到异常报错，这时就可以回滚事物，然后尝试重新发送；如果收到了消息，那么就可以提交事物。

channel.txSelect();//开启事物
      try{
          //发送消息
      }catch(Exection e){
          channel.txRollback()；//回滚事物
          //重新提交
      }

**缺点：**rabbitmq事物已开启，就会变为同步阻塞操作，生产者会阻塞等待是否发送成功，太耗性能会造成吞吐量的下降。

*  可以开启confirm模式。

在生产者哪里设置开启了confirm模式之后，每次写的消息都会分配一个唯一的id，然后如何写入了rabbitmq之中，rabbitmq会给你回传一个ack消息，告诉你这个消息发送OK了；如果rabbitmq没能处理这个消息，会回调你一个nack接口，告诉你这个消息失败了，你可以进行重试。而且你可以结合这个机制知道自己在内存里维护每个消息的id，如果超过一定时间还没接收到这个消息的回调，那么你可以进行重发。

//开启confirm
        channel.confirm();
        //发送成功回调
        public void ack(String messageId){
          
        }
    
        // 发送失败回调
        public void nack(String messageId){
            //重发该消息
        }

*  **二者不同**

事务机制是同步的，你提交了一个事物之后会阻塞住，但是confirm机制是异步的，发送消息之后可以接着发送下一个消息，然后rabbitmq会回调告知成功与否。  
**一般在生产者这块避免丢失，都是用confirm机制。**

### **2.rabbitmq自己弄丢了数据** ###

**设置消息持久化到磁盘**。设置持久化有两个步骤：

*  ①创建queue的时候将其设置为持久化的，这样就可以保证rabbitmq持久化queue的元数据，但是不会持久化queue里面的数据。
 *  ②发送消息的时候讲消息的deliveryMode设置为2，这样消息就会被设为持久化方式，此时rabbitmq就会将消息持久化到磁盘上。
 *  必须要同时开启这两个才可以。

而且持久化可以跟生产的confirm机制配合起来，只有消息持久化到了磁盘之后，才会通知生产者ack，这样就算是在持久化之前rabbitmq挂了，数据丢了，生产者收不到ack回调也会进行消息重发。

### **3.消费者弄丢了数据** ###

**使用rabbitmq提供的ack机制**，首先关闭rabbitmq的自动ack，然后每次在确保处理完这个消息之后，在代码里手动调用ack。这样就可以避免消息还没有处理完就ack。

### （二）kafka防止消息丢失 ###

### **1.消费端弄丢了数据** ###

**关闭自动提交offset，在自己处理完毕之后手动提交offset，这样就不会丢失数据**。

### **2.kafka弄丢了数据** ###

**一般要求设置4个参数来保证消息不丢失**：

*  ①给topic设置 **replication.factor**参数：这个值必须大于1，表示要求每个partition必须至少有2个副本。
 *  ②在kafka服务端设置**min.isync.replicas**参数：这个值必须大于1，表示 要求一个leader至少感知到有至少一个follower在跟自己保持联系正常同步数据，这样才能保证leader挂了之后还有一个follower。
 *  ③在生产者端设置**acks=all**：表示 要求每条每条数据，必须是写入所有replica副本之后，才能认为是写入成功了
 *  ④在生产者端设置**retries=MAX**(很大的一个值，表示无限重试)：表示 这个是要求一旦写入事变，就无限重试

### **3.生产者弄丢了数据** ###

如果按照上面设置了ack=all，则一定不会丢失数据，要求是，你的leader接收到消息，所有的follower都同步到了消息之后，才认为本次写成功了。**如果没满足这个条件，生产者会自动不断的重试，重试无限次**。

## 补充问题： ##

### 关于MQ的几件小事（五）如何保证消息按顺序执行 ###

[https://www.jianshu.com/p/02fdcb9e8784][https_www.jianshu.com_p_02fdcb9e8784]

### 关于MQ的几件小事（三）如何保证消息不重复消费 ###

[https://www.jianshu.com/p/172295e2e978][https_www.jianshu.com_p_172295e2e978]

### 关于MQ的几件小事（二）如何保证消息队列的高可用 ###

[https://www.jianshu.com/p/ab64681beb17][https_www.jianshu.com_p_ab64681beb17]

### Kafka的文件存储机制 ###

Kafka中消息是以topic进行分类的，生产者通过topic向Kafka broker发送消息，消费者通过topic读取数据。

然而topic在物理层面又能以partition为分组，一个topic可以分成若干个partition。partition还可以细分为segment，一个partition物理上由多个segment组成，segment文件由两部分组成，分别为“.index”文件和“.log”文件，分别表示为segment索引文件和数据文件。这两个文件的命令规则为：partition全局的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset值。

### Kafka 如何保证可靠性 ###

如果我们要往 Kafka 对应的主题发送消息，我们需要通过 Producer 完成。前面我们讲过 Kafka 主题对应了多个分区，每个分区下面又对应了多个副本；为了让用户设置数据可靠性， **Kafka 在 Producer 里面提供了消息确认机制**。也就是说我们可以通过配置来决定消息发送到对应分区的几个副本才算消息发送成功。可以在定义 Producer 时通过 acks 参数指定。这个参数支持以下三种值：

*  acks = 0：意味着如果生产者能够通过网络把消息发送出去，那么就认为消息已成功写入 Kafka 。在这种情况下还是有可能发生错误，比如发送的对象无能被序列化或者网卡发生故障，但如果是分区离线或整个集群长时间不可用，那就不会收到任何错误。在 acks=0 模式下的运行速度是非常快的（这就是为什么很多基准测试都是基于这个模式），你可以得到惊人的吞吐量和带宽利用率，不过如果选择了这种模式， 一定会丢失一些消息。
 *  acks = 1：意味若 Leader 在收到消息并把它写入到分区数据文件（不一定同步到磁盘上）时会返回确认或错误响应。在这个模式下，如果发生正常的 Leader 选举，生产者会在选举时收到一个 LeaderNotAvailableException 异常，如果生产者能恰当地处理这个错误，它会重试发送悄息，最终消息会安全到达新的 Leader 那里。不过在这个模式下仍然有可能丢失数据，比如消息已经成功写入 Leader，但在消息被复制到 follower 副本之前 Leader发生崩溃。
 *  acks = all（这个和 request.required.acks = -1 含义一样）：意味着 Leader 在返回确认或错误响应之前，会等待所有同步副本都收到悄息。如果和min.insync.replicas 参数结合起来，就可以决定在返回确认前至少有多少个副本能够收到悄息，生产者会一直重试直到消息被成功提交。不过这也是最慢的做法，因为生产者在继续发送其他消息之前需要等待所有副本都收到当前的消息。

### Kafka消息是采用Pull模式，还是Push模式 ###

Kafka最初考虑的问题是，customer应该从brokes拉取消息还是brokers将消息推送到consumer，也就是pull还push。在这方面，Kafka遵循了一种大部分消息系统共同的传统的**设计：producer将消息推送到broker，consumer从broker拉取消息。**

push模式下，当broker推送的速率远大于consumer消费的速率时，consumer恐怕就要崩溃了。

**最终Kafka还是选取了传统的pull模式。**Pull模式的另外一个好处是consumer可以自主决定是否批量的从broker拉取数据。Pull有个缺点是，如果broker没有可供消费的消息，将导致consumer不断在循环中轮询，直到新消息到t达。为了避免这点，Kafka有个参数可以让consumer阻塞知道新消息到达。

### Kafka是如何实现高吞吐率的 ###

1.  **顺序读写**：kafka的消息是不断追加到文件中的，这个特性使kafka可以**充分利用磁盘的顺序读写性能**
2.  **零拷贝**：跳过“用户缓冲区”的拷贝，建立一个**磁盘空间和内存的直接映射**，数据不再复制到“用户态缓冲区”
3.  **文件分段**：kafka的队列topic被分为了多个区partition，每个partition又分为多个段segment，所以一个队列中的消息实际上是保存在N多个片段文件中
4.  **批量发送**：Kafka允许进行批量发送消息，先将消息缓存在内存中，然后一次请求批量发送出去
5.  **数据压缩**：Kafka还**支持对消息集合进行压缩**，Producer可以通过GZIP或Snappy格式对消息集合进行压缩

### Kafka判断一个节点还活着的两个条件 ###

1.  **节点必须可以维护和 ZooKeeper 的连接**，Zookeeper 通过心跳机制检查每个节点的连接
2.  **如果节点是个 follower,他必须能及时的同步 leader 的写操作，延时不能太久**

# 参考书籍、文献和资料 #

1.[https://www.cnblogs.com/Vito-Yan/p/10319826.html][https_www.cnblogs.com_Vito-Yan_p_10319826.html]

2.[https://blog.csdn.net/wqc19920906/article/details/82193316][https_blog.csdn.net_wqc19920906_article_details_82193316]

3.[https://www.jianshu.com/p/70c5b8d51c75][https_www.jianshu.com_p_70c5b8d51c75]

4.[http://cloud.51cto.com/art/201507/484338.htm][http_cloud.51cto.com_art_201507_484338.htm]

5.[https://www.jianshu.com/p/8ed16edc73e4][https_www.jianshu.com_p_8ed16edc73e4]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70]: /images/20230528/1f75c176c2b54556b45405dcad2d0d00.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 1]: /images/20230528/8c974d10118a4f1e911c15cac69c301a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 2]: /images/20230528/b8187ae0bc4b44498f8ae120bbb34862.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 3]: /images/20230528/db21753a81dd4d15b3aec0759fb53ef6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 4]: /images/20230528/43908d3f28eb4476a69a1f6828bc8ea7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 5]: /images/20230528/d4bfc9b03d394550bc46655f8f1178ec.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 6]: /images/20230528/4b187ebcd07844fd94f12b2d7b477546.png
[20181219192323196]: /images/20230528/24cb1df1306a413b8f8165a782d90555.png
[20181219192323212]: /images/20230528/c4e386456e2641b79937f6d6d732ada4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 7]: /images/20230528/22bb15a2ef65429b92639aedda36b5fc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9mZW5nMTAzMzAxMTE_size_16_color_FFFFFF_t_70 8]: /images/20230528/31af493d00d044dc8e719b222d4738f5.png
[https_www.jianshu.com_p_02fdcb9e8784]: https://www.jianshu.com/p/02fdcb9e8784
[https_www.jianshu.com_p_172295e2e978]: https://www.jianshu.com/p/172295e2e978
[https_www.jianshu.com_p_ab64681beb17]: https://www.jianshu.com/p/ab64681beb17
[https_www.cnblogs.com_Vito-Yan_p_10319826.html]: https://www.cnblogs.com/Vito-Yan/p/10319826.html
[https_blog.csdn.net_wqc19920906_article_details_82193316]: https://blog.csdn.net/wqc19920906/article/details/82193316/
[https_www.jianshu.com_p_70c5b8d51c75]: https://www.jianshu.com/p/70c5b8d51c75
[http_cloud.51cto.com_art_201507_484338.htm]: http://cloud.51cto.com/art/201507/484338.htm
[https_www.jianshu.com_p_8ed16edc73e4]: https://www.jianshu.com/p/8ed16edc73e4