消息中央件 - 4.消息中央件Kafka服务端运行机制

kafka消息传输保障

• at most once(至多一次)：消息大概会丢失，但绝对不会重复传输。
  
• at least once(最少一次)：消息绝不会丢失，但大概会重复传输。
  
• exactly once(精准一次)：每条消息肯定会被传输一次且仅传输一次。
  

kafka客户端的消息传输机制

• 生产者[最少一次(at least once)]
  

生产者发送消息后，若消息被乐成提交到日志文件，由于多副本机制，那么消息就不会丢失。若消息没有被乐成提交到日志文件，如遭遇到了网络颠簸等题目，生产者无法判断消息是否已经提交乐成，此时生产者可以举行重试，这个过程大概造成消息的重复写入，此时生产的消息的传输保障机制为最少一次(at least once)。

• 消耗者[最少一次(at least once)和至多一次(at most once)]
  

消耗者处理处罚消息和提交消耗位移的次序很洪流平上决定了消耗者提供哪种消息传输保障。

• 先处理处罚消息再提交位移： 若消耗者拉取完消息后，应用逻辑先处理处罚消息后提交消耗位移，那么在消息处理处罚之后且位移提交之前消耗者宕机了，等它重新上线后，会从上一次位移提交的位置拉取，造成重复消耗，此时对应最少一次(at least once)。
  
• 先提交位移再处理处罚消息： 若消耗者拉取完消息后，应用逻辑先提交消耗位移后处理处罚消息，那么在位移提交之后且消息处理处罚完成之前消耗者宕机了，等它重新上线后，会从已经提交的位移处拉取，造成消息丢失，此时对应最多一次(at most once)。
  

由此可见，仅从生产者大概消耗者端是无法实现消息的精准一次的特性。以是Kafka的Broker端引入了幂等性的相干概念与客户端的消息传输机制相互共同，进而实现消息精准一次传输的特性。
生产者幂等性(单个分区)

所谓的幂等，简单地说就是对接口的多次调用所产生的效果和调用一次是划一的。生产者在举行重试的时间有大概会重复写入消息，而使用Kafka的幂等性功能之后就可以克制这种环境。开启幂等性的设置参数为enable.idempotence=true，默认未开启。当开启幂等性时，必要与生产者客户端相干设置配套使用，如：retries=0、acks=-1/all、max.in.flight.requests.per.connection<=5等设置。实际被骗开启幂等性后，生产者完全可以不举行设置该参数。
kafka为了实现生产者的幂等性，引入了生产者ID(pid)和序列号(seq)。每个新的生产者实例在初始化的时间都会被分配一个PID，这个PID对用户而言是完全透明的。对于每个PID，消息发送到的每一个分区都有对应的序列号，这些序列号从0开始单调递增。生产者每发送一条消息就会将＜PID，分区＞对应的序列号的值加1。

broker端会在内存中为每一对＜PID，分区＞维护一个序列号。对于收到的每一条消息，只有当它的序列号的值（seq_new）比broker端中维护的对应的序列号的值（seq_old）大1(seq_new=seq_old+1)时，消息才会被吸收。假如seq_new<=seq_old则分析消息重复写入，broker可以直接抛弃。假如seq_new>seq_old+1时，分析有数据尚未写入，存在消息丢失生产者大概会抛出OutOfOrderSequenceException非常。
由于引入序列号来实现幂等也只是针对每一对＜PID，分区＞而言，因此Kafka的幂等只能包管单个生产者会话（session）中单分区的幂等。
Kafka事件实现机制(跨分区)

幂等性并不能跨多个分区运作，而事件可以增补这个缺陷。事件可以包管对生产者多个分区写入利用的原子性。利用的原子性是指多个利用要么全部乐成，要么全部失败，不存在部分乐成、部分失败的大概。Kafka的事件实现告急组件包罗：

• 事件调和器(TransactionCoordinator)： 负责天生事件ID、启动事件和事件提交，同时还可以查抄和回滚事件。每个Kafka Broker都有自己的事件管理器，这些事件管理器之间可以相互通讯，以便调和分布式事件利用。
  
• 事件性生产者(TxProducer)： Kafka的事件性生产者答应应用步伐在事件范围内发送消息。事件性生产者产生的消息会被添加到一个本地事件中。当事件性生产者提交事件时，它将发送 Prepare Commit 哀求到事件管理器，然后期待事件管理器的相应。假如相应乐成，事件性生产者会将消息发送到Kafka集群中。假如相应失败，事件性生产者会取消利用并回滚本地事件。
  
• 事件性消耗者(TxComsumer)： Kafka的事件性消耗者可以读取事件性主题中的消息，并使用事件管理器来调和消耗过程。当事件性消耗者开始一个事件时，它会读取主题中的消息，并将其添加到事件中。假如事件乐成提交，消息将被消耗。假如事件回滚，则消息再被消耗时自动忽略。
  
• 事件日志(ControlBatch)： Kafka使用事件日志来记录事件利用的状态信息。每个事件包罗一组 Prepare Commit 记录和一个 Commit 记录。Prepare Commit 记任命于添加事件记录，并在提交事件之前向事件管理器发送 Prepare Commit 哀求。Commit 记任命于提交事件，并关照事件管理器事件已经乐成完成。
  
  
  
  

transactionalId与PID逐一对应，两者之间所差别的是transactionalId由用户显式设置，而PID是由Kafka内部分配的。而且当新的生产者使用已有的transactionalId启动时，会通过事件调和器获取当前该事件的状态，而使用transactionalId的旧生产者则会立即失效(通过producer epoch实现)。
生产者端

生产者代码相干

1.         public interface Producer<K, V> extends Closeable {
2.             // 初始化事务，包括申请 producer id
3.             void initTransactions();
4.             // 开始事务，这里会更改事务的本地状态
5.             void beginTransaction() throws ProducerFencedException;
6.             // 提交消费位置，offsets表示每个分区的消费位置，consumerGroupId表示消费组的名称
7.             void sendOffsetsToTransaction(Map<TopicPartition, ffsetAndMetadata> offsets, String consumerGroupId) throws ProducerFencedException;
8.             // 发送事务提交请求
9.             void commitTransaction() throws ProducerFencedException;
10.             // 发送事务回滚请求
11.             void abortTransaction() throws ProducerFencedException;
12.         }

复制代码

生产者设置相干

• transactional.id：事件ID，不能为空
  
• enable.idempotence：幂等性，必须开启。设置为true。
  
• transaction.timeout.ms：事件超时时间，默以为10秒，最长为15分钟。
  
• acks：要求此设置项必须设置为all。[开启幂等性后可不消设置]
  
• retries：重试次数，必须大于0。[开启幂等性后可不消设置]
  
• max.in.flight.requests.per.connection：单个毗连并行发送消息的哀求数目，必须小于或即是5。[开启幂等性后可不消设置]
  

消耗者端

消耗事件级别（isolation.level）

• read_uncommitted：消耗端应用可以看到（消耗到）未提交的事件，固然对于已提交的事件也是可见的。
  
• read_committed：消耗端应用不可以看到尚未提交的事件内的消息
  

当生产者开启事件时，KafkaConsumer在拉取到未提交事件的消息后会判断当前的事件级别，假如是read_uncommitted读未提交，则将该消息推送给消耗者线程举行消耗。假如是read_committed读已提交，则将该消息缓存在内部，直到生产者提交事件后才会将该消息推送给消耗者线程。固然假如生产者放弃了事件，KafkaConsumer也会将这些缓存的消息抛弃而不推送给消耗者线程。
LastStableOffset

由此可见，Kafka中的事件与我们所明确的事件实在并不是一回事。Kafka中的事件本质上照旧为相识决一组消息在差别分区下的幂等性保障题目，从而实现消息传输中精准一次的特性。再次夸大消息的精准一次语义是必要开启事件生产者、事件消耗者另有Broker端一同协作从而终极实现该特性。
   Kafka 事件的实现原理
Kafka事件「原理分析」
  消息的精准一次传输特性

从上述几个点形貌来看，要实现精准一次的语义就必要开启 Kafka 事件的功能而且使用公道的设置。别的，个人以为在实际开辟的过程中仍然必要有我们自己的机制去完成消息的幂等消耗题目，比如消息的ID。

• 使用事件：Kafka提供了事件功能，通过将消息的消耗和位移提交利用放在同一个事件中，可以实现准确一次消耗。消耗者可以使用手动位移提交，在处理处罚完消息后再提交位移，确保只有当消息处理处罚乐成后才会更新位移。
  
• 消耗者组与消耗位移：Kafka中的消耗者组（Consumer Group）可以协同工作来实现准确一次消耗。消耗者在处理处罚消息后，提交位移，以确保其他消耗者组中的消耗者不会重复消耗雷同的消息。
  
• 幂等性和去重：消耗者处理处罚消息的时间实现幂等，即无论消息处理处罚多少次，效果都是划一的。通过在消耗者端实现幂等逻辑，可以克制重复处理处罚消息。别的，可以使用唯一标识符或消息ID来举行消息去重，不处理处罚已经处理处罚过的消息。
  

控制器

在 Kafka 集群中会有一个或多个 broker，此中有一个 broker 会被推举为控制器（Kafka Controller）。在恣意时间，集群中有且仅有一个控制器。它负责管理整个集群中全部分区和副本的状态。当某个分区的leader副本出现故障时，由控制器负责为该分区推举新的leader副本。当检测到某个分区的ISR聚集发生厘革时，由控制器负责关照全部broker更新其元数据信息。当增长分区的数目时，同样照旧由控制器负责分区的重新分配。
控制器的推举

kafka控制器发生推举的机会：1.集群启动；2.控制器节点故障；3.控制器节点失联；4.控制器故障规复。

• 每个Broker启动时会在zookeeper中注册自己的信息到/brokers/ids/${broker_id}，表明自己是集群中的一员。
  
• broker会争先在zookeeper中创建/controller一个临时节点。谁先创建谁就成为了Controller。只有/controller中broker的值正常时才答应创建，非常时可以被修改。
  
• 当broker成为了Controller之后，会在zookeeper中的永世节点/controller_epoch修改当前controller的变动版本(自增1)。非Controller节点在收到Controller节点后起首会判断该哀求携带的版本是否是最新的，克制旧Controller的无效哀求。
  
• 当控制器乐成推举，而且已经完成初始化后，它会将自己的元数据信息（包罗分区和副本的状态信息等）生存到Zookeeper中，供其他Broker使用，同时还会定期发送心跳消息，以告知其他Broker自己的状态信息。
  

必要注意的是，Kafka控制器推举过程中，必须包管Zookeeper的正常运行，否则大概会导致推举失败大概数据不划一等题目。该kafka3.0以上的版本中，kafka已抛弃zk，而是将该部分信息注册到了内部。
集群脑裂题目的处理处罚

当控制器地点的节点由于某些缘故起因(网络不可达、GC等)产生故障不可对外相应到达肯定的时间后，集群中的其他节点就会以为该节点大概已经存在故障，从而触发控制器的推举。当新的控制器被推举之后，原故障的节点已经规复，此时集群中存在两个控制器，一个是新推举出来的控制器，一个是从故障中规复的控制器，这就是所谓的脑裂。
Kafka是通过使用controller_epoch来处理处罚，controller_epoch只是一个单调递增的数。第一次选择控制器时，controller_epoch值为1。假如再次选择新控制器，controller_epoch为2，依次单调递增。每个新的Controller被当选后都会向zookeeper的永世节点/controller_epoch举行自增运算。当其他Broker知道当前的controller_epoch时，假如他们从Controller收到包罗旧（较小）controller_epoch的消息，则它们将被忽略。即Broker根据最大的controller_epoch来区分最新的Controller。
   Kafka Controller推举过程
  ZK节点数据

消息的可靠性

固然Kafka提供了许多特性，但实际上假如对某些设置不认识或不能很好的明确，就大概会粉碎已有的功能特性。尤其是在可靠性这一方面是必要去认真研究的点。在设置这一方面而言，RocketMQ 确实要比 Kafka 友爱太多。
消息可靠性的机制

实在研究Kafka中消息可靠性传输无非就是研究消息传输都有哪些到场者，而且各个到场者之间是怎样调和的。消息传输的到场者无非就是：1.生产者发送消息、2.服务器端同步消息并完发展期化、3.消耗者完成消息的消耗。而相干的调和无非就是依赖：1.ack机制、2.消息的发送或吸收方式、3.位移的管理，包罗Broker端位移的机制以及消耗者端位移的提交方式。

• 生产者端：通过消息发送方以同步或异步的方式获取发送的消息的相应效果，失败做重试或其他处理处罚计谋。
  
• 服务端ACK：指定分区中有N个副本收到这条消息，生产者才以为这条消息是写入乐成的。
  
  
  
  • acks = 1，默以为1。生产者发送消息，只要 leader 副本乐成写入消息，就代表乐成。这种方案的题目在于，当返回乐成后，假如 leader 副本和 follower 副本还没有来得及同步，leader 就瓦解了，那么在推举后新的 leader 就没有这条消息，也就丢失了。
    
  • acks = 0。生产者发送消息后直接算写入乐成，不必要期待相应。这个方案的题目很显着，只要服务端写消息时出现任何题目，都会导致消息丢失。
    
  • acks = -1/all。生产者发送消息后，必要期待 ISR 中的全部副本都乐成写入消息后才华收到服务端的相应。毫无疑问这种方案的可靠性是最高的，但是假如 ISR 中只有leader 副本，那么就和 acks = 1 毫无差别了。
    
  
  
• 服务端分区副本可以通过HW位移体现消息是否可以被消耗。
  
• 消耗端手动提交位移：默认环境下，当消耗者消耗到消息后，就会自动提交位移。但是假如消耗者消耗堕落，没有进入真正的业务处理处罚，那么就大概会导致这条消息消耗失败，从而丢失。开启手动提交位移后，期待业务正常处理处罚完成后，再提交offset，从而包管消息已乐成处理处罚。
  

从这个机制上看，假如要包管消息的可靠性就必要保障每个传输环节都是可靠的。但实际上并不是每个环境都能得到包管。比如：

• 生产者发送消息到Broker：由服务端的ACK机制和发送端获取相应效果并举行处理处罚来共同包管。当 acks 为 0，只要服务端写消息时出现任何题目，都会导致消息丢失；当 acks 设置为 1 时，生产者发送消息，只要 leader 副本乐成写入消息，就代表乐成。这种方案的题目在于，当返回乐成后，假如 leader 副本和 follower 副本还没有来得及同步，leader 就瓦解了，那么在推举后新的 leader 就没有这条消息，也就丢失了。
  
• Broker 存储数据：broker存储消息起首是先写入到利用体系的Page Cache中。什么时间将缓存的数据写入文件中是由利用体系自行决定。以是假云云机会器忽然挂了，而且消息尚未被同步到follower副本中，消息也是会丢失的。
  
• 消耗者消耗数据：在开启自动提交 offset 时，只要消耗者消耗到消息，那么就会自动提交偏移量，假如业务还没有来得及处理处罚，那么消息就会丢失。
  

单个Partition内的消息次序性

Kafka 可以包管同一个分区中的消息是有序的。假如生产者按照肯定的次序发送消息，那么这些消息也会次序地写入分区，进而消耗者也可以按照同样的次序消耗它们。
对于某些应用来说，次序性非常告急，比如 MySQL 的 binlog 传输，假如出现错误就会造成非常严峻的结果。假如将 acks 参数设置为非零值，而且max.in.flight.requests.per.connection参数设置为大于1的值，那么就会出现错序的征象：假如第一批次消息写入失败，而第二批次消息写入乐成，那么生产者会重试发送第一批次的消息，此时假如第一批次的消息写入乐成，那么这两个批次的消息就出现了错序。
一样平常而言，在必要包管消息次序的场所发起把参数max.in.flight.requests.per.connection设置为1，而不是把acks设置为0，不外如许也会影响团体的吞吐。
以是单个Partition内的消息的次序确实是可以包管的，但是这个次序并不是生产者发送消息的次序，而是Kafka将消息写入到分区内的次序。