kafka技术总结

妖狐艹你老母 2023-01-02 02:24 100阅读 0赞

# kafka #

在早期的版本中，为了获得极致的性能，在设计方面做了很多的牺牲，比如不保证消息的可靠性，可能会丢失消息，也不支持集群，功能上也比较简陋，这些牺牲对于处理海量日志这个特定的场景都是可以接受的。这个时期的 Kafka 甚至不能称之为一个合格的消息队列。

**但是，请注意，重点一般都在后面。随后的几年 Kafka 逐步补齐了这些短板，你在网上搜到的很多消息队列的对比文章还在说 Kafka 不可靠，其实这种说法早已经过时了。当下的 Kafka 已经发展为一个非常成熟的消息队列产品，无论在数据可靠性、稳定性和功能特性等方面都可以满足绝大多数场景的需求。**

**Kafka 与周边生态系统的兼容性是最好的没有之一，尤其在大数据和流计算领域，几乎所有的相关开源软件系统都会优先支持 Kafka。**

Kafka 使用 Scala 和 Java 语言开发，设计上大量使用了批量和异步的思想，这种设计使得 Kafka 能做到超高的性能。Kafka 的性能，尤其是异步收发的性能，是三者中最好的，但与 RocketMQ 并没有量级上的差异，大约每秒钟可以处理几十万条消息。

但是 Kafka 这种异步批量的设计带来的问题是，它的同步收发消息的响应时延比较高，因为当客户端发送一条消息的时候，Kafka 并不会立即发送出去，而是要等一会儿攒一批再发送，在它的 Broker 中，很多地方都会使用这种“先攒一波再一起处理”的设计。当你的业务场景中，每秒钟消息数量没有那么多的时候，Kafka 的时延反而会比较高。所以，**Kafka 不太适合在线业务场景。**

第一梯队的消息队列：kafka  rocketMq RabbitMQ

第二梯队的消息队列：ActiveMq

Pulsar，很多人可能都没听说过这个产品，Pulsar 是一个新兴的开源消息队列产品，最早是由 Yahoo 开发，目前处于成长期，流行度和成熟度相对没有那么高。与其他消息队列最大的不同是，Pulsar 采用存储和计算分离的设计，我个人非常喜欢这种设计，它有可能会引领未来消息队列的一个发展方向

# zookeepr #

Kafka 在 ZooKeeper 都保存的信息

![f8a33075651262baea83f9dd5e97a7ab.png][]

左侧这棵树保存的是 Kafka 的 Broker 信息，/brokers/ids/\[0…N\]，每个临时节点对应着一个在线的 Broker，Broker 启动后会创建一个临时节点，代表 Broker 已经加入集群可以提供服务了，节点名称就是 BrokerID，节点内保存了包括 Broker 的地址、版本号、启动时间等等一些 Broker 的基本信息。如果 Broker 宕机或者与 ZooKeeper 集群失联了，这个临时节点也会随之消失。

右侧部分的这棵树保存的就是主题和分区的信息。/brokers/topics/ 节点下面的每个子节点都是一个主题，节点的名称就是主题名称。每个主题节点下面都包含一个固定的 partitions 节点，pattitions 节点的子节点就是主题下的所有分区，节点名称就是分区编号。

每个分区节点下面是一个名为 state 的临时节点，节点中保存着分区当前的 leader 和所有的 ISR 的 BrokerID。这个 state 临时节点是由这个分区当前的 Leader Broker 创建的。如果这个分区的 Leader Broker 宕机了，对应的这个 state 临时节点也会消失，直到新的 Leader 被选举出来，再次创建 state 临时节点。

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

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;width:103.15pt;"> 消息中间件的组成 </td> 
 <td style="vertical-align:top;width:313.7pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:103.15pt;"> Broker </td> 
 <td style="vertical-align:top;width:313.7pt;"> 消息服务器，作为server提供消息核心服务 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:103.15pt;"> Producer </td> 
 <td style="vertical-align:top;width:313.7pt;"> 消息生产者，业务的发起方，负责生产消息传输给broker </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:103.15pt;"> consumer </td> 
 <td style="vertical-align:top;width:313.7pt;"> 消息消费者，业务的处理方，负责从broker获取消息并进行业务逻辑处理 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:103.15pt;"> topic </td> 
 <td style="vertical-align:top;width:313.7pt;"> 主题，发布订阅模式下的消息统一汇集地，不同生产者向topic发送消息，由MQ服务器分发到不同的订阅者，实现消息的 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;广播 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:103.15pt;"> queue </td> 
 <td style="vertical-align:top;width:313.7pt;"> 队列，PTP模式下，特定生产者向特定queue发送消息，消费者订阅特定的queue完成指定消息的接收 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:103.15pt;"> message </td> 
 <td style="vertical-align:top;width:313.7pt;"> 消息体，根据不同通信协议定义的固定格式进行编码的数据包，来封装业务数据，实现消息的传输 </td> 
 </tr> 
 </tbody> 
</table>

# 2. **消息中间件的优势** #

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;width:137.5pt;"> 消息中间件的优势 </td> 
 <td style="vertical-align:top;width:288.6pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:137.5pt;"> 系统解耦 </td> 
 <td style="vertical-align:top;width:288.6pt;"> 交互系统之间没有直接的调用关系，只是通过消息传输，故系统侵入性不强，耦合度低。 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:137.5pt;"> 提高系统响应时间 </td> 
 <td style="vertical-align:top;width:288.6pt;"> 例如原来的一套逻辑，完成支付可能涉及先修改订单状态、计算会员积分、通知物流配送几个逻辑才能完成；通过MQ架构设计，就可将紧急重要（需要立刻响应）的业务放到该调用方法中，响应要求不高的使用消息队列，放到MQ队列中，供消费者处理。 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:137.5pt;"> 为大数据处理架构提供服务 </td> 
 <td style="vertical-align:top;width:288.6pt;"> 通过消息作为整合，大数据的背景下，消息队列还与实时处理架构整合，为数据处理提供性能支持。 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:137.5pt;"> Java消息服务——JMS </td> 
 <td style="vertical-align:top;width:288.6pt;"> Java消息服务（Java Message Service，JMS）应用程序接口是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。 JMS中的P2P和Pub/Sub消息模式：点对点（point to point， queue）与发布订阅（publish/subscribe，topic）最初是由JMS定义的。这两种模式主要区别或解决的问题就是发送到队列的消息能否重复消费(多订阅)。 </td> 
 </tr> 
 </tbody> 
</table>

# **3.消息中间件应用场景** #

<table style="margin-left:6.75pt;"> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> 消息中间件应用场景 </td> 
 <td style="vertical-align:top;width:213.05pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> 异步通信 </td> 
 <td style="vertical-align:top;width:213.05pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> 削峰(过载保护,流量控制) </td> 
 <td style="vertical-align:top;width:213.05pt;"> 避免过多的请求压垮我们的系统 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> 服务解耦 </td> 
 <td style="vertical-align:top;width:213.05pt;"> &nbsp; </td> 
 </tr> 
 </tbody> 
</table>

# **4.引入消息队列带来的问题** #

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> 引入消息队列带来的问题 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> 引入消息队列带来的延迟问题； </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> 增加了系统的复杂度； </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:213.05pt;"> 可能产生数据不一致的问题。 </td> 
 </tr> 
 </tbody> 
</table>

# 5.丢失数据 #

MQ本身一般数据不会丢失。除非极其罕见的情况，MQ还没来得及持久化自己就挂了，这样可能导致一部分数据丢失。

MQ集群本身收到客户端发送来的消息以后，无论从理论上，还是很多生产系统的实际验证，都不会再丢消息的。

客户端发送数据如果启用了发送接受应答机制也，确保消息不会在传递过程中由于网络或服务器故障丢失

客户端消费数据，每次拉取的数据会存储到消费客户端的本地缓存中，消费者再从客户端的本地缓存读取数据。因为网络原因拉取的数据没传输到客户端，则缓存中就不会有此次拉取的数据还会从新拉取。

相当与消费者客户端本地缓存了一份分区和offset数据，只有在消费者客户端启动的时候才去broker端获取一次消费offset只要消费者被关闭/客户端不宕机，即使不提交offset消费者也能正常消费数据。

**由于业务处理异常导致的数据发送失败不属于mq的数据丢失。**

# 6.如何选择消息中间件 #

选择中间件的考量维度：可靠性，性能，功能，可运维行，可拓展性，是否开源及社区活跃度

1. 开源，社区活跃，生态好(和其他产品有好的集成和兼容)

比如，kafka 开源社区活跃，Kafka 和 Flink 就有比较好的兼容性，Flink 内置了 Kafka 的 Data Source，使用 Kafka 就很容易作为 Flink 的数据源开发流计算应用，如果你用一个比较小众的消息队列产品，在进行流计算的时候，你就不得不自己开发一个 Flink 的 Data Source。

2. 必须具备的特性

*  消息的可靠传递：确保不丢消息；
 *  Cluster：支持集群，确保不会因为某个节点宕机导致服务不可用，当然也不能丢消息；
 *  性能：具备足够好的性能，能满足绝大多数场景的性能要求。

rabbitmq：  
优点：轻量，迅捷，容易部署和使用，拥有灵活的路由配置  
缺点：性能和吞吐量较差，不易进行二次开发  
rocketmq：  
优点：性能好，稳定可靠，有活跃的中文社区，特点响应快  
缺点：兼容性较差，但随意影响力的扩大，该问题会有改善  
kafka：  
优点：拥有强大的性能及吞吐量，兼容性很好  
缺点：由于“攒一波再处理”导致延迟比较高  
pulsar：  
采用存储和计算分离的设计，是消息队里产品中黑马，值得持续关注

RabbitMq

*  如果说，消息队列并不是你将要构建系统的主角之一，你对消息队列功能和性能都没有很高的要求，只需要一个开箱即用易于维护的产品，我建议你使用 RabbitMQ。

RocketMq

*  如果你的系统使用消息队列主要场景是处理在线业务，比如在交易系统中用消息队列传递订单，那 RocketMQ 的低延迟和金融级的稳定性是你需要的。

Kafka

*  如果你需要处理海量的消息，像收集日志、监控信息或是前端的埋点这类数据，或是你的应用场景大量使用了大数据、流计算相关的开源产品，那 Kafka 是最适合你的消息队列。

**线业务指的是那种服务于web页面或者APP的服务，这种服务都需要很低的延迟，否则APP就会很卡，体验不好。  
流计算一般不会服务在线业务，都是做事后的统计分析。**

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;width:85.2pt;"> 特性 </td> 
 <td style="vertical-align:top;width:85.2pt;"> Kafla </td> 
 <td style="vertical-align:top;width:85.2pt;"> RocketMQ </td> 
 <td style="vertical-align:top;width:85.25pt;"> RabbitMQ </td> 
 <td style="vertical-align:top;width:85.25pt;"> ActiveMQ </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:85.2pt;"> 特点 </td> 
 <td style="vertical-align:top;width:85.2pt;"> 1.快速持久化：通过磁盘顺序读写与零拷贝机制，可以在O(1)的系统开销下进行消息持久化； 2. 高吞吐：在一台普通的服务器上既可以达到几十W/s的吞吐速率； 高堆积：支持topic下消费者较长时间离线，消息堆积量大； 3.完全的分布式系统：Broker、Producer、Consumer都原生自动支持分布式，依赖zookeeper自动实现复杂均衡； 4.支持Hadoop数据并行加载：对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。 </td> 
 <td style="vertical-align:top;width:85.2pt;"> 1.能够保证严格的消息顺序 2.提供针对消息的过滤功能 3.提供丰富的消息拉取模式 4.高效的订阅者水平扩展能力 5.实时的消息订阅机制 6.亿级消息堆积能力 </td> 
 <td style="vertical-align:top;width:85.25pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:85.25pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:85.2pt;"> 单机吞吐量(每秒处理的消息数) </td> 
 <td style="vertical-align:top;width:85.2pt;"> 几十万条消息 </td> 
 <td style="vertical-align:top;width:85.2pt;">十万条消息</td> 
 <td style="vertical-align:top;width:85.25pt;"> 万级 </td> 
 <td style="vertical-align:top;width:85.25pt;"> 万级 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:85.2pt;"> 消息延迟 </td> 
 <td style="vertical-align:top;width:85.2pt;"> 毫秒级 </td> 
 <td style="vertical-align:top;width:85.2pt;"> 毫秒级 </td> 
 <td style="vertical-align:top;width:85.25pt;"> 微妙级 </td> 
 <td style="vertical-align:top;width:85.25pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:85.2pt;"> 消息丢失 </td> 
 <td style="vertical-align:top;width:85.2pt;"> 理论上不丢失 </td> 
 <td style="vertical-align:top;width:85.2pt;"> 理论上不丢失 </td> 
 <td style="vertical-align:top;width:85.25pt;"> 理论上不会 </td> 
 <td style="vertical-align:top;width:85.25pt;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:85.2pt;"> 使用场景 </td> 
 <td style="vertical-align:top;width:85.2pt;"> 流失数据处理，日志采集 </td> 
 <td style="vertical-align:top;width:85.2pt;"> 业务数据处理 </td> 
 <td style="vertical-align:top;width:85.25pt;"> 业务数据处理 </td> 
 <td style="vertical-align:top;width:85.25pt;"> 业务解耦 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:85.2pt;"> 语言 </td> 
 <td style="vertical-align:top;width:85.2pt;"> 服务端：scala+java 客户端：支持多语言，java优先 </td> 
 <td style="vertical-align:top;width:85.2pt;"> 服务端：Java 客户端：只支持java </td> 
 <td style="vertical-align:top;width:85.25pt;"> 服务端：Erlang 客户端：语言无关 </td> 
 <td style="vertical-align:top;width:85.25pt;"> 服务端：Java 客户端：多语言，java优先 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:85.2pt;"> 总结 </td> 
 <td style="vertical-align:top;width:85.2pt;"> &nbsp; 1.Kafka 与周边生态系统的兼容性是最好的没有之一，尤其在大数据和流计算领域，几乎所有的相关开源软件系统都会优先支持 Kafka。 </td> 
 <td style="vertical-align:top;width:85.2pt;"> 
 <ol> 
 <li>模型简单，接口易。</li> 
 <li>阿里大规模使用(支付宝)</li> 
 </ol>1.RocketMQ 对在线业务的响应时延做了很多的优化，大多数情况下可以做到毫秒级的响应，如果你的应用场景很在意响应时延，那应该选择使用 RocketMQ。 2.&nbsp;RocketMQ 的一个劣势是，作为国产的消息队列，相比国外的比较流行的同类产品，在国际上还没有那么流行，与周边生态系统的集成和兼容程度要略逊一筹。 </td> 
 <td style="vertical-align:top;width:85.25pt;"> 
 <ol> 
 <li>Erlang语言难度较大但性能较好</li> 
 <li>不支持动态扩容</li> 
 <li>RabbitMQ 是一个相当轻量级的消息队列，非常容易部署和使用。</li> 
 <li>RabbitMQ 对消息堆积的支持并不好，在它的设计理念里面，消息队列是一个管道，大量的消息积压是一种不正常的情况，应当尽量去避免。当大量消息积压的时候，会导致 RabbitMQ 的性能急剧下降。</li> 
 </ol></td> 
 <td style="vertical-align:top;width:85.25pt;">activeMQ 是最老牌的开源消息队列，是十年前唯一可供选择的开源消息队列，目前已进入老年期，社区不活跃。无论是功能还是性能方面，ActiveMQ 都与现代的消息队列存在明显的差距，它存在的意义仅限于兼容那些还在用的爷爷辈儿的系统。</td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:85.2pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:85.2pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:85.2pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:85.25pt;"> &nbsp; </td> 
 <td style="vertical-align:top;width:85.25pt;"> &nbsp; </td> 
 </tr> 
 </tbody> 
</table>

rocketMQ是怎么做到低延时的？

主要是设计上的选择问题，Kafka中到处都是“批量和异步”设计，它更关注的是整体的吞吐量，而RocketMQ的设计选择更多的是尽量及时处理请求。  
  
比如发消息，同样是用户调用了send()方法，RockMQ它会直接把这个消息发出去，而Kafka会把这个消息放到本地缓存里面，然后择机异步批量发送。  
  
所以，RocketMQ它的时延更小一些，而Kafka的吞吐量更高。

# 7.利用事务消息实现分布式事务 #

大部分传统的单体关系型数据库都完整的实现了 ACID，但是，**对于分布式系统来说，严格的实现 ACID 这四个特性几乎是不可能的**，或者说实现的代价太大，大到我们无法接受。

分布式事务就是要在分布式系统中的实现事务。在分布式系统中，在保证可用性和不严重牺牲性能的前提下，光是要实现数据的一致性就已经非常困难了，所以出现了很多“残血版”的一致性，**比如顺序一致性、最终一致性等等。**

显然实现严格的分布式事务是更加不可能完成的任务。所以，目前大家所说的分布式事务，更多情况下，是在分布式系统中事务的不完整实现。在不同的应用场景中，有不同的实现，目的都是通过一些妥协来解决实际问题。

**事务消息适用的场景主要是那些需要异步更新数据，并且对数据实时性要求不太高的场景。比如我们在开始时提到的那个例子，在创建订单后，如果出现短暂的几秒，购物车里的商品没有被及时清空，也不是完全不可接受的，只要最终购物车的数据和订单数据保持一致就可以了。**

**事务消息需要消息队列提供相应的功能才能实现，Kafka 和 RocketMQ 都提供了事务相关功能。回到订单和购物车这个例子，我们一起来看下如何用消息队列来实现分布式事务。**

![ef965788dd5a47d62a31af914a3777b2.png][]

首先，订单系统在消息队列上开启一个事务。然后订单系统给消息服务器发送一个“半消息”，这个半消息不是说消息内容不完整，它包含的内容就是完整的消息内容，半消息和普通消息的唯一区别是，在事务提交之前，**对于消费者来说，这个消息是不可见的。 **

半消息发送成功后，订单系统就可以执行本地事务了，在订单库中创建一条订单记录，并提交订单库的数据库事务。然后根据本地事务的执行结果决定提交或者回滚事务消息。如果订单创建成功，那就提交事务消息，购物车系统就可以消费到这条消息继续后续的流程。如果订单创建失败，那就回滚事务消息，购物车系统就不会收到这条消息。这样就基本实现了“要么都成功，要么都失败”的一致性要求。

这个实现过程中，有一个问题是没有解决的。**如果在第四步提交事务消息时失败了怎么办？**对于这个问题，Kafka 和 RocketMQ 给出了 2 种不同的解决方案。K**afka 的解决方案比较简单粗暴，直接抛出异常，让用户自行处理**。我们可以在业务代码中反复重试提交，直到提交成功，或者删除之前创建的订单进行补偿。**RocketMQ 则给出了另外一种解决方案。**

但是，这不代表 RocketMQ 的事务功能比 Kafka 更好，只能说在我们这个例子的场景下，更适合使用 RocketMQ。Kafka 对于事务的定义、实现和适用场景，和 RocketMQ 有比较大的差异，后面的课程中，我们会专门讲到 Kafka 的事务的实现原理。

## **RocketMQ 中的分布式事务实现:** ##

在 RocketMQ 中的事务实现中，增加了事务反查的机制来解决事务消息提交失败的问题。如果 Producer 也就是订单系统，在提交或者回滚事务消息时发生网络异常，RocketMQ 的 Broker 没有收到提交或者回滚的请求，Broker 会定期去 Producer 上反查这个事务对应的本地事务的状态，然后根据反查结果决定提交或者回滚这个事务。

为了支撑这个事务反查机制，我们的业务代码需要实现一个反查本地事务状态的接口，告知 RocketMQ 本地事务是成功还是失败。

在我们这个例子中，反查本地事务的逻辑也很简单，我们只要根据消息中的订单 ID，在订单库中查询这个订单是否存在即可，如果订单存在则返回成功，否则返回失败。RocketMQ 会自动根据事务反查的结果提交或者回滚事务消息。

这个反查本地事务的实现，并不依赖消息的发送方，也就是订单服务的某个实例节点上的任何数据。这种情况下，即使是发送事务消息的那个订单服务节点宕机了，RocketMQ 依然可以通过其他订单服务的节点来执行反查，确保事务的完整性。

综合上面讲的通用事务消息的实现和 RocketMQ 的事务反查机制，使用 RocketMQ 事务消息功能实现分布式事务的流程如下图：

![dfd6a33a7282cd802390116e3714a233.png][]

# 8.如何确保消息不会丢失 #

**检测消息丢失的方法:**

用消息队列最尴尬的情况不是丢消息，而是消息丢了还不知道。

我们可以利用消息队列的有序性来验证是否有消息丢失。原理非常简单，在 Producer 端，我们给每个发出的消息附加一个连续递增的序号，然后在 Consumer 端来检查这个序号的连续性。

如果没有消息丢失，Consumer 收到消息的序号必然是连续递增的，或者说收到的消息，其中的序号必然是上一条消息的序号 +1。如果检测到序号不连续，那就是丢消息了。还可以通过缺失的序号来确定丢失的是哪条消息，方便进一步排查原因。

大多数消息队列的客户端都支持拦截器机制，你可以利用这个拦截器机制，在 Producer 发送消息之前的拦截器中将序号注入到消息中，在 Consumer 收到消息的拦截器中检测序号的连续性，这样实现的好处是消息检测的代码不会侵入到你的业务代码中，待你的系统稳定后，也方便将这部分检测的逻辑关闭或者删除。

**确保消息可靠传递:**

一条消息从生产到消费完成这个过程，可以划分三个阶段:

![c304af0518d1157596f32d483e642297.png][]

1.  生产阶段: 在这个阶段，从消息在 Producer 创建出来，经过网络传输发送到 Broker 端。
2.  存储阶段: 在这个阶段，消息在 Broker 端存储，如果是集群，消息会在这个阶段被复制到其他的副本上。
3.  消费阶段: 在这个阶段，Consumer 从 Broker 上拉取消息，经过网络传输发送到 Consumer 上。

每个阶段都需要正确的编写代码并且设置正确的配置项，才能配合消息队列的可靠性机制，确保消息不会丢失。

*  在生产阶段，你需要捕获消息发送的错误，并重发消息。
 *  在存储阶段，你可以通过配置刷盘和复制相关的参数，让消息写入到多个副本的磁盘上，来确保消息不会因为某个 Broker 宕机或者磁盘损坏而丢失。
 *  在消费阶段，你需要在处理完全部消费业务逻辑之后，再发送消费确认。

**1.在生产阶段**，消息队列通过最常用的请求确认机制，来保证消息的可靠传递。你在编写发送消息代码时，需要注意，正确处理返回值或者捕获异常，就可以保证这个阶段的消息不会丢失。以 Kafka 为例，我们看一下如何可靠地发送消息：

**同步发送时**，只要注意捕获异常即可。

try {
        RecordMetadata metadata = producer.send(record).get();
        System.out.println("消息发送成功。");
    } catch (Throwable e) {
        System.out.println("消息发送失败！");
        System.out.println(e);
    }

**异步发送时**，则需要在回调方法里进行检查。这个地方是需要特别注意的，很多丢消息的原因就是，我们使用了异步发送，却没有在回调中检查发送结果。

producer.send(record, (metadata, exception) -> {
        if (metadata != null) {
            System.out.println("消息发送成功。");
        } else {
            System.out.println("消息发送失败！");
            System.out.println(exception);
        }
    });

**2. 存储阶段，**在存储阶段正常情况下，只要 Broker 在正常运行，就不会出现丢失消息的问题，但是如果 Broker 出现了故障，比如进程死掉了或者服务器宕机了，还是可能会丢失消息的。

避免服务器宕机丢失消息：消息同步刷盘+多副本同步数据

如果对消息的可靠性要求非常高，可以通过配置 Broker 参数来避免因为宕机丢消息。对于单个节点的 Broker，需要配置 Broker 参数，在收到消息后，将消息写入磁盘后再给 Producer 返回确认响应，这样即使发生宕机，由于消息已经被写入磁盘，就不会丢失消息，恢复后还可以继续消费。例如，在 RocketMQ 中，需要将刷盘方式 flushDiskType 配置为 SYNC\_FLUSH 同步刷盘。

如果是 Broker 是由多个节点组成的集群，需要将 Broker 集群配置成：至少将消息发送到 2 个以上的节点，再给客户端回复发送确认响应。这样当某个 Broker 宕机时，其他的 Broker 可以替代宕机的 Broker，也不会发生消息丢失。后面我会专门安排一节课，来讲解在集群模式下，消息队列是如何通过消息复制来确保消息的可靠性的。

**3. 消费阶段，**

消费阶段采用和生产阶段类似的确认机制来保证消息的可靠传递，客户端从 Broker 拉取消息后，执行用户的消费业务逻辑，成功后，才会给 Broker 发送消费确认响应。如果 Broker 没有收到消费确认响应，下次拉消息的时候还会返回同一条消息，确保消息不会在网络传输过程中丢失，也不会因为客户端在执行消费逻辑中出错导致丢失。

你在编写消费代码时需要注意的是，**不要在收到消息后就立即发送消费确认，而是应该在执行完所有消费业务逻辑之后，再发送消费确认。**

# 9.消息积压了该如何处理 #

**1. 发送端性能优化**

发送端业务代码的处理性能，实际上和消息队列的关系不大，因为一般发送端都是先执行自己的业务逻辑，最后再发送消息。如果说，你的代码发送消息的性能上不去，你需要优先检查一下，是不是发消息之前的业务逻辑耗时太多导致的。对于发送消息的业务逻辑，只需要注意设置合适的并发和批量大小，就可以达到很好的发送性能。

**2. 消费端性能优化**

使用消息队列的时候，大部分的性能问题都出现在消费端，如果消费的速度跟不上发送端生产消息的速度，就会造成消息积压。如果这种性能倒挂的问题只是暂时的，那问题不大，只要消费端的性能恢复之后，超过发送端的性能，那积压的消息是可以逐渐被消化掉的。

要是消费速度一直比生产速度慢，时间长了，整个系统就会出现问题，要么，消息队列的存储被填满无法提供服务，要么消息丢失，这对于整个系统来说都是严重故障。

所以，我们在设计系统的时候，一定要保证消费端的消费性能要高于生产端的发送性能，这样的系统才能健康的持续运行。

消费端的性能优化除了优化消费业务逻辑以外，也可以通过水平扩容，增加消费端的并发数来提升总体的消费性能。特别需要注意的一点是，在扩容 Consumer 的实例数量的同时，必须同步扩容主题中的分区（也叫队列）数量，确保 Consumer 的实例数和分区数量是相等的。如果 Consumer 的实例数量超过分区数量，这样的扩容实际上是没有效果的。原因我们之前讲过，因为对于消费者来说，在每个分区上实际上只能支持单线程消费。

**消息积压了该如何处理？**

能导致积压突然增加，最粗粒度的原因，只有两种：要么是发送变快了，要么是消费变慢了。

对于系统发生消息积压的情况，需要先解决积压，再分析原因，毕竟保证系统的可用性是首先要解决的问题。快速解决积压的方法就是通过水平扩容增加 Consumer 的实例数量。

大部分消息队列都内置了监控的功能，只要通过监控数据，很容易确定是哪种原因。如果是单位时间发送的消息增多，比如说是赶上大促或者抢购，短时间内不太可能优化消费端的代码来提升消费性能，唯一的方法是通过扩容消费端的实例数来提升总体的消费能力。

如果短时间内没有足够的服务器资源进行扩容，没办法的办法是，将系统降级，通过关闭一些不重要的业务，减少发送方发送的数据量，最低限度让系统还能正常运转，服务一些重要业务。

还有一种不太常见的情况，你通过监控发现，无论是发送消息的速度还是消费消息的速度和原来都没什么变化，这时候你需要检查一下你的消费端，是不是消费失败导致的一条消息反复消费这种情况比较多，这种情况也会拖慢整个系统的消费速度。如果监控到消费变慢了，你需要检查你的消费实例，分析一下是什么原因导致消费变慢。优先检查一下日志是否有大量的消费错误，如果没有错误的话，可以通过打印堆栈信息，看一下你的消费线程是不是卡在什么地方不动了，比如触发了死锁或者卡在等待某些资源上了。

# 10.Kafka如何实现高性能IO #

Kafka 的高性能设计中的几个关键的技术点：

*  使用批量处理的方式来提升系统吞吐能力。
 *  基于磁盘文件高性能顺序读写的特性来设计的存储结构。
 *  利用操作系统的 PageCache 来缓存数据，减少 IO 并提升读性能。
 *  使用零拷贝技术加速消费流程。

**使用批量消息提升服务端处理能力:**

当你调用 send() 方法发送一条消息之后，无论你是同步发送还是异步发送，我们可以设置Kafka 都不立即就把这条消息发送出去。它会先把这条消息，存放在内存中缓存起来，然后选择合适的时机把缓存中的所有消息组成一批，一次性发给 Broker。简单地说，就是攒一波一起发。**只有相同分区的消息才能组成同一个批消息。**

在服务端，Kafka 不会把一批消息再还原成多条消息，再一条一条地处理，这样太慢了。Kafka 这块儿处理的非常聪明，每批消息都会被当做一个“批消息”来处理。也就是说，在 Broker 整个处理流程中，无论是写入磁盘、从磁盘读出来、还是复制到其他副本这些流程中**，批消息都不会被解开，一直是作为一条“批消息”来进行处理的。**

在消费时，消息同样是以批为单位进行传递的，Consumer 从 Broker 拉到一批消息后，在客户端把批消息解开，再一条一条交给用户代码处理。

比如说，你在客户端发送 30 条消息，在业务程序看来，是发送了 30 条消息，而对于 Kafka 的 Broker 来说，它其实就是处理了 1 条包含 30 条消息的“批消息”而已。显然处理 1 次请求要比处理 30 次请求要快得多。构建批消息和解开批消息分别在发送端和消费端的客户端完成，不仅减轻了 Broker 的压力，最重要的是减少了 Broker 处理请求的次数，提升了总体的处理能力。

**使用顺序读写提升磁盘 IO 性能:**

相比于网络传输和内存，磁盘 IO 的速度是比较慢的。对于消息队列的**服务端**来说，性能的瓶颈主要在**磁盘 IO** 这一块。

对于磁盘来说，它有一个特性，就是顺序读写的性能要远远好于随机读写。在 SSD（固态硬盘）上，顺序读写的性能要比随机读写快几倍，如果是机械硬盘，这个差距会达到几十倍。

操作系统每次从磁盘读写数据的时候，需要先寻址，也就是先要找到数据在磁盘上的物理位置，然后再进行数据读写。如果是机械硬盘，这个寻址需要比较长的时间，因为它要移动磁头，这是个机械运动，机械硬盘工作的时候会发出咔咔的声音，就是移动磁头发出的声音。

顺序读写相比随机读写省去了大部分的寻址时间，它只要寻址一次，就可以连续地读写下去，所以说，性能要比随机读写要好很多。Kafka 就是充分利用了磁盘的这个特性。它的存储设计非常简单，对于每个分区，它把从 Producer 收到的消息，顺序地写入对应的 log 文件中，一个文件写满了，就开启一个新的文件这样顺序写下去。消费的时候，也是从某个全局的位置开始，也就是某一个 log 文件中的某个位置开始，顺序地把消息读出来。这样一个简单的设计，充分利用了顺序读写这个特性，极大提升了 Kafka 在使用磁盘时的 IO 性能。

**利用缓存(PageCache)加速消息读写:**

在 Kafka 中，它会利用 PageCache 加速消息读写。

**[pagecache kafka][]**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70]

producer生产消息时，会使用pwrite()系统调用【对应到Java NIO中是FileChannel.write() API】按偏移量写入数据，并且都会先写入page cache里。consumer消费消息时，会使用sendfile()系统调用【对应FileChannel.transferTo() API】，零拷贝地将数据从page cache传输到broker的Socket buffer，再通过网络传输。

**如果Kafka producer的生产速率与consumer的消费速率相差不大，那么就能几乎只靠对broker page cache的读写完成整个生产-消费过程**，磁盘访问非常少。这个结论俗称为“读写空中接力”。

作者：LittleMagic  
链接：https://www.jianshu.com/p/92f33aa0ff52  
来源：简书  
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

**ZeroCopy：零拷贝技术：**

我们知道，在服务端，处理消费的大致逻辑是这样的：

*  首先，从文件中找到消息数据，读到内存中；
 *  然后，把消息通过网络发给客户端。

这个过程中，数据实际上做了 2 次或者 3 次复制：

1.  从文件复制数据到 PageCache 中，如果命中 PageCache，这一步可以省掉；
2.  从 PageCache 复制到**应用程序的内存空间中**，也就是我们可以操作的对象所在的内存；
3.  从**应用程序的内存空间**复制到 Socket 的缓冲区，这个过程就是我们调用网络应用框架的 API 发送数据的过程。

kafka 使用零拷贝技术可以把这个复制次数减少一次，**上面的 2、3 步骤两次复制合并成一次复制。直接从 PageCache 中把数据复制到 Socket 缓冲区中，**这样不仅减少一次数据复制，更重要的是，由于不用把数据复制到用户内存空间，DMA 控制器可以直接完成数据复制，不需要 CPU 参与，速度更快。

下面是这个零拷贝对应的系统调用：

#include <sys/socket.h>
 ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count);

# 11. 数据压缩 #

在一台配置比较高的服务器上，对 Kafka 做过一个极限的性能压测，想验证一下 Kafka 到底有多快。我使用的种子消息大小为 1KB，只要是并发数量足够多，不开启压缩时，可以打满万兆网卡的全部带宽，TPS 接近 100 万。开启压缩时，TPS 可以达到 2000 万左右，吞吐量提升了大约 20 倍！

测试服务器的网卡就是普通的万兆网卡，极限带宽也就是 10Gb/s，压测时候的实际网络流量大概在 7Gb/s 左右。

**什么情况适合使用数据压缩？**

**压缩它的本质是资源的置换，是一个时间换空间，或者说是 CPU 资源换存储资源的游戏。**

**就像木桶的那个短板一样，每一个系统它都有一个性能瓶颈资源，可能是磁盘 IO，网络带宽，也可能是 CPU。如果使用压缩，能用长板来换一些短板，那总体上就能提升性能，这样就是划算的。如果用了压缩之后，短板更短了，那就不划算了，不如不用。**

在使用压缩之前，首先你需要考虑，当前这个场景是不是真的适合使用数据压缩。

比如，进程之间通过网络传输数据，这个数据是不是需要压缩呢？我和你一起来对比一下：

*  不压缩直接传输需要的时间是： 传输未压缩数据的耗时。
 *  使用数据压缩需要的时间是： 压缩耗时 + 传输压缩数据耗时 + 解压耗时。

到底是压缩快，还是不压缩快呢？其实不好说。影响的因素非常多，比如数据的压缩率、网络带宽、收发两端服务器的繁忙程度等等。

**业余本身是cpu密集性**(业务逻辑处理本身占用cpu较高)不适合压缩：

压缩和解压的操作都是计算密集型的操作，非常耗费 CPU 资源。如果你的应用处理业务逻辑就需要耗费大量的 CPU 资源，就不太适合再进行压缩和解压。

**io密集型适合压缩：**

又比如说，如果你的系统的瓶颈是磁盘的 IO 性能，CPU 资源又很闲，这种情况就非常适合在把数据写入磁盘前先进行压缩。

但是，如果你的系统读写比严重不均衡，你还要考虑，每读一次数据就要解压一次是不是划算。

**应该选择什么压缩算法？**

压缩算法可以分为有损压缩和无损压缩。有损压缩主要是用来压缩音视频，它压缩之后是会丢失信息的。我们这里讨论的全都是无损压缩，也就是说，数据经过压缩和解压过程之后，与压缩之前相比，是 100% 相同的。

目前常用的压缩算法包括：ZIP，GZIP，SNAPPY，LZ4 等等。

选择压缩算法的时候，主要需要考虑数据的压缩率和压缩耗时。一般来说，压缩率越高的算法，压缩耗时也越高。如果是对性能要求高的系统，可以选择压缩速度快的算法，比如 LZ4；如果需要更高的压缩比，可以考虑 GZIP 或者压缩率更高的 XZ 等算法。

**Kafka 是如何处理消息压缩的？**

Kafka 是否开启压缩，这是可以配置，它也支持配置使用哪一种压缩算法。原因我们在上面说过，不同的业务场景是否需要开启压缩，选择哪种压缩算法是不能一概而论的。所以，Kafka 的设计者把这个选择权交给使用者。

在开启压缩时，Kafka 选择一批消息一起压缩，每一个批消息就是一个压缩分段。使用者也可以通过参数来控制每批消息的大小。

**在 Kafka 中，生产者生成一个批消息发给服务端，在服务端中是不会拆分批消息的。那按照批来压缩，意味着，在服务端也不用对这批消息进行解压，可以整批直接存储，然后整批发送给消费者。最后，批消息由消费者进行解压。**

**在服务端不用解压，就不会耗费服务端宝贵的 CPU 资源，同时还能获得压缩后，占用传输带宽小，占用存储空间小的这些好处，这是一个非常聪明的设计。**

**在使用 Kafka 时，如果生产者和消费者的 CPU 资源不是特别吃紧，开启压缩后，可以节省网络带宽和服务端的存储空间，提升总体的吞吐量，一般都是个不错的选择。**

# 12.Kafka的协调服务ZooKeeper #

Kafka 主要使用 ZooKeeper 来保存它的元数据、监控 Broker 和分区的存活状态，并利用 ZooKeeper 来进行选举。

Kafka 在 ZooKeeper 中保存的元数据，主要就是 Broker 的列表和主题分区信息两棵树。这份元数据同时也被缓存到每一个 Broker 中。客户端并不直接和 ZooKeeper 来通信，而是在需要的时候，通过 RPC 请求去 Broker 上拉取它关心的主题的元数据，然后保存到客户端的元数据缓存中，以便支撑客户端生产和消费。

目前 Kafka 的这种设计，集群的可用性是严重依赖 ZooKeeper 的，也就是说，如果 ZooKeeper 集群不能提供服务，那整个 Kafka 集群也就不能提供服务了，这其实是一个不太好的设计。

如果你需要要部署大规模的 Kafka 集群，建议的方式是，拆分成多个互相独立的小集群部署，每个小集群都使用一组独立的 ZooKeeper 提供服务。这样，每个 ZooKeeper 中存储的数据相对比较少，并且如果某个 ZooKeeper 集群故障，只会影响到一个小的 Kafka 集群，故障的影响面相对小一些。

**ZooKeeper 的作用：**

**ZooKeeper 作为一个分布式的协调服务框架，主要用来解决分布式集群中，应用系统需要面对的各种通用的一致性问题。**ZooKeeper 本身可以部署为一个集群，集群的各个节点之间可以通过选举来产生一个 Leader，选举遵循半数以上的原则，所以一般集群需要部署奇数个节点。

**ZooKeeper 最核心的功能是，它提供了一个分布式的存储系统，数据的组织方式类似于 UNIX 文件系统的树形结构。**由于这是一个可以保证一致性的存储系统，所以你可以放心地在你的应用集群中读写 ZooKeeper 的数据，而不用担心数据一致性的问题。分布式系统中一些需要整个集群所有节点都访问的元数据，比如集群节点信息、公共配置信息等，特别适合保存在 ZooKeeper 中。

利用 ZooKeeper 临时节点和 Watcher 机制，我们很容易随时来获取业务集群中每个节点的存活状态，并且可以监控业务集群的节点变化情况，当有节点上下线时，都可以收到来自 ZooKeeper 的通知。

Kafka 在 ZooKeeper 都保存的信息

![f8a33075651262baea83f9dd5e97a7ab.png][]

我们从左往右来看，左侧这棵树保存的是 Kafka 的 Broker 信息，/brokers/ids/\[0…N\]，每个临时节点对应着一个在线的 Broker，Broker 启动后会创建一个临时节点，代表 Broker 已经加入集群可以提供服务了，节点名称就是 BrokerID，节点内保存了包括 Broker 的地址、版本号、启动时间等等一些 Broker 的基本信息。如果 Broker 宕机或者与 ZooKeeper 集群失联了，这个临时节点也会随之消失。

**Kafka 客户端如何找到对应的 Broker？**

那 Kafka 客户端如何找到主题、队列对应的 Broker 呢？其实，通过上面 ZooKeeper 中的数据结构，你应该已经可以猜的八九不离十了。是的，先根据主题和队列，在右边的树中找到分区对应的 state 临时节点，我们刚刚说过，state 节点中保存了这个分区 Leader 的 BrokerID。拿到这个 Leader 的 BrokerID 后，再去左侧的树中，找到 BrokerID 对应的临时节点，就可以获取到 Broker 真正的访问地址了。

Kafka 在每个 Broker 中都维护了一份和 ZooKeeper 中一样的元数据缓存，并不是每次客户端请求元数据就去读一次 ZooKeeper。由于 ZooKeeper 提供了 Watcher 这种监控机制，Kafka 可以感知到 ZooKeeper 中的元数据变化，从而及时更新 Broker 中的元数据缓存。

ZooKeeper 也并不是完美的，在使用的时候你需要注意几个问题：

1.  不要往 ZooKeeper 里面写入大量数据，它不是一个真正意义上的存储系统，只适合存放少量的数据。依据服务器配置的不同，ZooKeeper 在写入超过几百 MB 数据之后，性能和稳定性都会严重下降。
2.  不要让业务集群的可用性依赖于 ZooKeeper 的可用性，什么意思呢？你的系统可以使用 Zookeeper，但你要留一手，要考虑如果 Zookeeper 集群宕机了，你的业务集群最好还能提供服务。因为 ZooKeeper 的选举过程是比较慢的，而它对网络的抖动又比较敏感，一旦触发选举，这段时间内的 ZooKeeper 是不能提供任何服务的。

# 13.Kafka中如何实现事务 #

**Kafka 的事务和 Exactly Once 可以解决什么问题**

RocketMQ 中的事务，它解决的问题是，确保执行本地事务和发消息这两个操作，要么都成功，要么都失败。并且，RocketMQ 增加了一个事务反查的机制，来尽量提高事务执行的成功率和数据一致性。

Kafka 中的事务，它解决的问题是，确保在一个事务中发送的多条消息，要么都成功，要么都失败。注意，这里面的多条消息不一定要在同一个主题和分区中，可以是发往多个主题和分区的消息。当然，你可以在 Kafka 的事务执行过程中，加入本地事务，来实现和 RocketMQ 中事务类似的效果，但是 Kafka 是没有事务反查机制的。

Kafka 的这种事务机制，单独来使用的场景不多。更多的情况下被用来配合 Kafka 的幂等机制来实现 Kafka 的 Exactly Once 语义。我在之前的课程中也强调过，这里面的 Exactly Once，和我们通常理解的消息队列的服务水平中的 Exactly Once 是不一样的。

我们通常理解消息队列的服务水平中的 Exactly Once，它指的是，消息从生产者发送到 Broker，然后消费者再从 Broker 拉取消息，然后进行消费。这个过程中，确保每一条消息恰好传输一次，不重不丢。我们之前说过，包括 Kafka 在内的几个常见的开源消息队列，都只能做到 At Least Once，也就是至少一次，保证消息不丢，但有可能会重复。做不到 Exactly Once。

![79d48d68807438e7a131ce8361efc5f3.png][]

**那 Kafka 中的 Exactly Once 又是解决的什么问题呢？它解决的是，在流计算中，用 Kafka 作为数据源，并且将计算结果保存到 Kafka 这种场景下，数据从 Kafka 的某个主题中消费，在计算集群中计算，再把计算结果保存在 Kafka 的其他主题中。这样的过程中，保证每条消息都被恰好计算一次，确保计算结果正确。**

![9004b43ace507f3dde4b8c5370d26577.png][]

举个例子，比如，我们把所有订单消息保存在一个 Kafka 的主题 Order 中，在 Flink 集群中运行一个计算任务，统计每分钟的订单收入，然后把结果保存在另一个 Kafka 的主题 Income 里面。要保证计算结果准确，就要确保，无论是 Kafka 集群还是 Flink 集群中任何节点发生故障，每条消息都只能被计算一次，不能重复计算，否则计算结果就错了。这里面有一个很重要的限制条件，就是数据必须来自 Kafka 并且计算结果都必须保存到 Kafka 中，才可以享受到 Kafka 的 Excactly Once 机制。

可以看到，Kafka 的 Exactly Once 机制，是为了解决在“读数据 - 计算 - 保存结果”这样的计算过程中数据不重不丢，而不是我们通常理解的使用消息队列进行消息生产消费过程中的 Exactly Once。

**Kafka 的事务是如何实现的？**

事务的实现流程:

![4620244fa177159299951fda740c8ca6.png][]

总结一下 Kafka 这个两阶段的流程，准备阶段，生产者发消息给协调者开启事务，然后消息发送到每个分区上。提交阶段，生产者发消息给协调者提交事务，协调者给每个分区发一条“事务结束”的消息，完成分布式事务提交。

不同之处在于对处于事务中的消息的处理方式，RocketMQ 是把这些消息暂存在一个特殊的队列中，待事务提交后再移动到业务队列中；而 Kafka 直接把消息放到对应的业务分区中，配合客户端过滤来暂时屏蔽进行中的事务消息。同时你需要了解，RocketMQ 和 Kafka 的事务，它们的适用场景是不一样的，RocketMQ 的事务适用于解决本地事务和发消息的数据一致性问题，**而 Kafka 的事务则是用于实现它的 Exactly Once 机制，应用于实时计算的场景中**。

# 14.Kafka 如何通过选举来产生新的 Leader #

**分区leader的选举：**

Kafka 的选举，因为 Kafka 的选举实现比较简单。严格地说，Kafka 分区的 Leader 并不是选举出来的，而是 Controller 指定的。Kafka 使用 ZooKeeper 来监控每个分区的多个副本，如果发现某个分区的主节点宕机了，Controller 会收到 ZooKeeper 的通知，这个时候，Controller 会从 ISR 节点中选择一个节点，指定为新的 Leader。

controller的选举：

在 Kafka 中 Controller 本身也是通过 ZooKeeper 选举产生的。接下来我要讲的，Kafka Controller 利用 ZooKeeper 选举的方法，你一定要记住并学会，因为这种方法非常简单实用，并且适用性非常广泛，在设计很多分布式系统中都可以用到。

这种选举方法严格来说也不是真正的“选举”，而是一种抢占模式。实现也很简单，每个 Broker 在启动后，都会尝试在 ZooKeeper 中创建同一个临时节点：/controller，并把自身的信息写入到这个节点中。由于 ZooKeeper 它是一个可以保证数据一致性的分布式存储，所以，集群中只会有一个 Broker 抢到这个临时节点，那它就是 Leader 节点。其他没抢到 Leader 的节点，会 Watch 这个临时节点，如果当前的 Leader 节点宕机，所有其他节点都会收到通知，它们会开始新一轮的抢 Leader 游戏。

# 15.为什么说 Pulsar 存储计算分离的架构是未来消息队列的发展方向 #

早期的消息队列，主要被用来在系统之间异步交换数据，大部分消息队列的存储能力都比较弱，不支持消息持久化，不提倡在消息队列中堆积大量的消息，这个时期的消息队列，本质上是一个数据的管道。

现有的流计算平台，包括 Storm、Flink 和 Spark，它们的节点都是无状态的纯计算节点，是没有数据存储能力的。所以，现在的流计算平台，它很难做大量数据的聚合，并且在数据可靠性保证、数据一致性、故障恢复等方面，也做得不太好。

而消息队列正好相反，它很好地保证了数据的可靠性、一致性，但是 Broker 只具备存储能力，没有计算的功能，数据流进去什么样，流出来还是什么样。同样是处理实时数据流的系统，一个只能计算不能存储，一个只能存储不能计算，那未来如果出现一个新的系统，**既能计算也能存储，如果还能有不错的性能，是不是就会把现在的消息队列和流计算平台都给替代了？这是很有可能的。**

**对于一个“带计算功能的消息队列”来说，采用存储计算分离的设计，计算节点负责流计算，存储节点负责存储消息，这个设计就非常和谐了。**

Kafka 在做 Kafka Streams，Pulsar 在做 Pulsar Functions，其实大家都在不约而同的做同一件事儿，就是流计算。

[f8a33075651262baea83f9dd5e97a7ab.png]: /images/20221120/cf50e2e3ec5d4138bc61f195615ca00f.png
[ef965788dd5a47d62a31af914a3777b2.png]: /images/20221120/287304df641c44f29acaa46e47105310.png
[dfd6a33a7282cd802390116e3714a233.png]: /images/20221120/cc0f20b99d3243e5a9ea89524cf41af3.png
[c304af0518d1157596f32d483e642297.png]: /images/20221120/af1cff928b3b4955abaab81daf9832ab.png
[pagecache kafka]: https://blog.csdn.net/yangshengwei230612/article/details/112463026
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70]: /images/20221120/bbdcf69c40f341308756d41dac28940f.png
[79d48d68807438e7a131ce8361efc5f3.png]: /images/20221120/056a0336720c461383d9a33e0b6e7dd3.png
[9004b43ace507f3dde4b8c5370d26577.png]: /images/20221120/47c2def32ab54ff1a1e2322fb6ea1b8f.png
[4620244fa177159299951fda740c8ca6.png]: /images/20221120/9cb369ce4f84418eb2499d6c5e12fc80.png