Kafka

典型回答 当我们使用Kafka发送消息时，一般有两种方式，分别是同步发送（producer.send(msg).get() ）及异步发送（producer.send(msg, callback)）。 同步发送的时候，可以在发送消息后，通过get方法等待消息结果：producer.send(record).get(); ，这种情况能够准确的拿到消息最终的发送结果，要么是成功，要么是失败。 而异步发送，是采用了callback的方式进行回调的，可以大大的提升消息的吞吐量，也可以根据回调来判断消息是否发送成功。 不管是同步发送还是异步发送，最终都需要在Producer端把消息发送到Broker中，那么这个过程大致如下： Kafka 的 Producer 在发送消息时通常涉及两个线程，主线程（Main）和发送线程（Sender）和一个消息累加器（RecordAccumulator） Main线程是 Producer 的入口，负责初始化 Producer 的配置、创建 KafkaProducer 实例并执行发送逻辑。它会按照用户定义的发送方式（同步或异步）发送消息，然后等待消息发送完成。 一条消息的发送，在调用send方法后，会经过拦截器、序列化器及分区器。 拦截器主要用于在消息发送之前和之后对消息进行定制化的处理，如对消息进行修改、记录日志、统计信息等。 序列化器负责将消息的键和值对象转换为字节数组，以便在网络上传输。 分区器决定了一条消息被发送到哪个 Partition 中。它根据消息的键（如果有）或者特定的分区策略，选择出一个目标 Partition。 RecordAccumulator在 Kafka Producer 中起到了消息积累和批量发送的作用，当 Producer 发送消息时，不会立即将每条消息发送到 Broker，而是将消息添加到 RecordAccumulator 维护的内部缓冲区中，RecordAccumulator 会根据配置的条件（如batch.size、linger.ms）对待发送的消息进行批量处理。 当满足指定条件时，RecordAccumulator 将缓冲区中的消息组织成一个批次（batch），然后一次性发送给 Broker。如果发送失败或发生错误，RecordAccumulator 可以将消息重新分配到新的批次中进行重试。这样可以确保消息不会丢失，同时提高消息的可靠性。 Send线程是负责实际的消息发送和处理的。发送线程会定期从待发送队列中取出消息，并将其发送到对应的 Partition 的 Leader Broker 上。它主要负责网络通信操作，并处理发送请求的结果，包括确认的接收、错误处理等。 **NetworkClient 和 Selector **是两个重要的组件，分别负责网络通信和 I/O 多路复用。 发送线程会把消息发送到Kafka集群中对应的Partition的Partition Leader，Partition Leader 接收到消息后，会对消息进行一系列的处理。它会将消息写入本地的日志文件（Log） 为了保证数据的可靠性和高可用性，Kafka 使用了消息复制机制。Leader Broker 接收到消息后，会将消息复制到其他副本（Partition Follower）。副本是通过网络复制数据的，它们会定期从 Leader Broker 同步消息。 每一个Partition Follower在写入本地log之后，会向Leader发送一个ACK。 但是我们的Producer其实也是需要依赖ACK才能知道消息有没有投递成功的，而这个ACK是何时发送的，Producer又要不要关心呢？这就涉及到了kafka的ack机制，生产者会根据设置的 request.required.acks 参数不同，选择等待或或直接发送下一条消息： ...

典型回答 高水位（HW，High Watermark）是Kafka中的一个重要的概念，主要是用于管理消费者的进度和保证数据的可靠性的。 高水位标识了一个特定的消息偏移量（offset），即一个分区中已提交消息的最高偏移量（offset），消费者只能拉取到这个 offset 之前的消息。消费者可以通过跟踪高水位来确定自己消费的位置。 这里的已提交指的是ISRs中的所有副本都记录了这条消息 在Kafka中，HW主要有两个作用： 消费进度管理：消费者可以通过记录上一次消费的偏移量，然后将其与分区的高水位进行比较，来确定自己的消费进度。消费者可以在和高水位对比之后继续消费新的消息，确保不会错过任何已提交的消息。这样，消费者可以按照自己的节奏进行消费，不受其他消费者的影响。 数据的可靠性：高水位还用于确保数据的可靠性。在Kafka中，只有消息被写入主副本（Leader Replica）并被所有的同步副本（In-Sync Replicas，ISR）确认后，才被认为是已提交的消息。高水位表示已经被提交的消息的边界。只有高水位之前的消息才能被认为是已经被确认的，其他的消息可能会因为副本故障或其他原因而丢失。 还有一个概念，叫做LEO，即 Log End Offset，他是日志最后消息的偏移量。 它标识当前日志文件中下一条待写入消息的 offset。 当消费者消费消息时，它可以使用高水位作为参考点，只消费高水位之前的消息，以确保消费的是已经被确认的消息，从而保证数据的可靠性。如上图，只消费offet为6之前的消息。 我们都知道，在Kafka中，每个分区都有一个Leader副本和多个Follower副本。 当Leader副本发生故障时，Kafka会选择一个新的Leader副本。这个切换过程中，需要保证数据的一致性，即新的Leader副本必须具有和旧Leader副本一样的消息顺序。 为了实现这个目标，Kafka引入了Leader Epoch的概念。Leader Epoch是一个递增的整数，每次副本切换时都会增加。它用于标识每个Leader副本的任期。 每个副本都会维护自己的Leader Epoch记录。它记录了副本所属的分区在不同Leader副本之间切换时的任期。 在副本切换过程中，新的Leader会检查旧Leader副本的Leader Epoch和高水位。只有当旧Leader副本的Leader Epoch小于等于新Leader副本的Leader Epoch，并且旧Leader副本的高水位小于等于新Leader副本的高水位时，新Leader副本才会接受旧Leader副本的数据。 通过使用Leader Epoch和高水位的验证，Kafka可以避免新的Leader副本接受旧Leader副本之后的消息，从而避免数据回滚。只有那些在旧Leader副本的Leader Epoch和高水位之前的消息才会被新Leader副本接受。 扩展知识 Leader Epoch的过程 每个分区都有一个初始的Leader Epoch，通常为0。 当Leader副本发生故障或需要进行切换时，Kafka会触发副本切换过程。 副本切换过程中，Kafka会从ISR（In-Sync Replicas，同步副本）中选择一个新的Follower副本作为新的Leader副本。 新的Leader副本会增加自己的Leader Epoch，使其大于之前的Leader Epoch。这表示进入了一个新的任期。 新的Leader副本会验证旧Leader副本的状态以确保数据的一致性。它会检查旧Leader副本的Leader Epoch和高水位。 如果旧Leader副本的Leader Epoch小于等于新Leader副本的Leader Epoch，并且旧Leader副本的高水位小于等于新Leader副本的高水位，则验证通过。 一旦验证通过，新的Leader副本会开始从ISR中的一部分副本复制数据，以确保新Leader上的数据与旧Leader一致。 一旦新的Leader副本复制了旧Leader副本的所有数据，并达到了与旧Leader副本相同的高水位，副本切换过程就完成了。

典型回答 ✅Kafka的架构是怎么样的？ 在上面的文章中我们介绍过Kafka的架构，这里面有个很关键的组件，是Zookeeper。（但是后来被移除了，详见后文） ZooKeeper是Kafka集群中使用的分布式协调服务，用于维护Kafka集群的状态和元数据信息，例如主题和分区的分配信息、消费者组和消费者偏移量等。 展开说就是有以下这些功能： 集群管理 Zookeeper 负责管理 Kafka 集群中的所有 Broker，当有 Broker 加入或退出时，Zookeeper 负责通知其他组件，让 Kafka 及时更新集群信息。 实现原理：当 Kafka 的 Broker 启动时，会在 Zookeeper 创建一个临时节点。如果 Broker 崩溃或断开连接，临时节点会自动删除，其他组件能感知到这个 Broker 不再可用。 1 /brokers/ids/{broker_id} 该节点会存储 Broker 的 host、port 信息，这样其他 Broker 或客户端可以从 Zookeeper 获取 Broker 列表。 分区和副本管理 kafka的数据存储是基于分区和副本的，zk则负责存储这些分区的元数据，包括： 该分区有哪些副本（Replicas） 哪个副本是 Leader，哪些是 Follower 副本的 ISR（In-Sync Replica，即同步副本集合）列表 ✅介绍一下Kafka的ISR机制？ 实现原理： Kafka 在 /brokers/topics/{topic_name}/partitions/{partition_id}/state 目录下存储每个分区的 Leader、副本、ISR（In-Sync Replica）列表。 1 /brokers/topics/topic1/partitions/0/state 存储内容大致为： 1 2 3 4 5 { "leader": 1001, "replicas": [1001, 1002, 1003], "isr": [1001, 1002] } Broker 会监听监听这个节点，当 Leader 发生变化时，Kafka 通过 Zookeeper 通知所有相关 Broker 更新 Leader 信息。 ...

典型回答 很多人说kafka不支持事务消息，但其实，Kafka支持事务消息，只不过此事务消息非彼事务消息。 在Kafka中，事务消息可以确保一组生产或消费操作要么全部成功，要么全部失败，以保证消息处理的原子性。也就是说，他的作用是保证一组消息要么都成功，要么都失败。 没错，这就是事务中原子性的定义。 只不过，通常在分布式系统中，我们通常说的事务消息，如RocketMQ的事务消息保证的时候本地事务和发MQ能作为一个原子性，即要么一起成功，要么一起失败。 所以，Kafka的事务消息只保证他自己的消息发送的原子性。而RocketMQ的事务消息是保证本地事务和发消息的原子性。 那么，继续说Kafka的事务消息。 Kafka的事务机制允许一个生产者向多个主题和分区发送消息，并确保这些消息要么全部成功提交，要么全部回滚。事务由生产者负责开启、提交和回滚。Kafka通过引入事务日志和协调器来实现事务消息的管理。 Kafka为了实现事务，有几个关键组件： Transaction Coordinator（事务协调器） 事务协调器是Kafka的一个特殊组件，负责管理生产者的事务状态。它会分配一个唯一的Transaction ID给每个生产者，并维护一个Transaction Log来记录事务的开始、提交或中止等状态信息。 Producer ID (PID) 和 Epoch Kafka为每个生产者（Producer）分配一个唯一的PID（Producer ID）和Epoch（版本号）。PID用于标识生产者，Epoch用于标识该生产者的事务版本。通过这两个字段，可以防止因重启等导致的重复消息和过期消息。 Transaction Log（事务日志） 事务日志是Kafka的一个特殊内部主题，用于记录事务的开始、提交、回滚等操作。事务协调器通过该日志确保事务的原子性和一致性。 他的实现过程大致如下： 开启事务 生产者在初始化时需要配置transactional.id来启用事务，在每次事务开始时，生产者会给协调者发请求来开启事务，协调者在事务日志中记录下事务 ID。 发送消息 生产者发消息之前，先发送请求事务协调器，让他记录消息发送的主题和分区。接下来开始向Broker发消息， 提交事务 当所有消息发送完毕，生产者调用commitTransaction()来提交事务。此时，事务协调器会将所有消息的状态改为已提交，并通知消费者可以读取这些消息。 回滚事务 如果在事务过程中出现异常，生产者可以调用abortTransaction()回滚事务。事务协调器会将所有相关消息标记为中止状态，确保消费者无法读取未完成的消息。 这里的事务提交和回滚，其实也是遵循了一个2阶段提交的。协调者在接收到提交或者回滚请求时，第一阶段，他会把事务的状态设置为“预提交”，并写入事务日志。接下来第二阶段，协调者在事务相关的所有分区中，都会写一条“事务结束”的特殊消息，当 Kafka 的消费者，也就是客户端，读到这个事务结束的特殊消息之后，它就可以把之前暂时过滤的那些未提交的事务消息，放行给业务代码进行消费了。最后，协调者记录最后一条事务日志，标识这个事务已经结束了。 以下是我在极客时间的《消息队列高手课》中看到过的作者画的一张图，供大家参考： 扩展知识 如何发送事务消息 消息的生产者可以通过如下方式发送一个事务消息： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("transactional.id", "my-transactional-id"); // 配置事务ID KafkaProducer<String, String> producer = new KafkaProducer<>(props); producer.initTransactions(); // 初始化事务 try { producer.beginTransaction(); // 开启事务 producer.send(new ProducerRecord<>("topic1", "key1", "value1")); producer.send(new ProducerRecord<>("topic2", "key2", "value2")); producer.commitTransaction(); // 提交事务 } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) { producer.close(); // 无法恢复的异常，需关闭生产者 } catch (KafkaException e) { producer.abortTransaction(); // 其他异常，回滚事务 } 消费者在读取事务消息时，需要将隔离级别设置为read_committed： ...

典型回答 Kafka 的整体架构比较简单，是显式分布式架构，主要由 Producer（生产者）、broker（Kafka集群）和 consumer（消费者） 组成。 生产者（Producer）：生产者负责将消息发布到Kafka集群中的一个或多个主题(Topic)，每个Topic包含一个或多个分区（Partition）。 主题：Topic。主题是承载消息的逻辑容器，在实际使用中多用来区分具体的业务。 分区：Partition。一个有序不变的消息序列。每个主题下可以有多个分区。 消费者（Consumer）：消费者负责从Kafka集群中的一个或多个主题消费消息，并将消费的偏移量（Offset）提交回Kafka以保证消息的顺序性和一致性。 偏移量：Offset。表示分区中每条消息的位置信息，是一个单调递增且不变的值。 Kafka集群：Kafka集群是由多个Kafka节点（Broker）组成的分布式系统。每个节点都可以存储一个或多个主题（topic）的分区（partition）副本，以提供高可用性和容错能力。 如下图中，包含了 Broker1、Broker2和 Broker3组成了一个集群。用来提升高可用性。 在集群中，每个分区（partition）都可以有多个副本。这些副本中包含了一个 Leader （也可以叫做Leader Partition 或者 Leader Replication） 和多个 Follower （也可以叫做Follower Partition 或者 Follower Replication），只有 Leader 才能处理生产者和消费者的请求，而 Follower 只是 Leader 的备份，用于提供数据的冗余备份和容错能力。如果 Leader 发生故障，Kafka 集群会自动将 Follower 提升为新的 Leader ，从而实现高可用性和容错能力。 ZooKeeper：ZooKeeper是Kafka集群中使用的分布式协调服务，用于维护Kafka集群的状态和元数据信息，例如主题和分区的分配信息、消费者组和消费者偏移量等。（消费者组的偏移量从kafka 0.9版本开始就不再存放在zk中，而是存放在内部的topic中）

典型回答 重平衡的问题大家都知道，他会导致STW，所以大家都不希望他发生，或者希望他能把影响降到最低。 ✅什么是Kafka的重平衡机制？ 然后有一种优化方案是渐进式重平衡，目的是在进行消费者组重平衡时，尽可能减少数据中断和不必要的分区变更。它由 Cooperative Sticky Assignor提供支持，并在 Kafka 2.4.0 引入。 开启方式： 1 2 props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"); 开启了CooperativeStickyAssignor之后，Kafka 通过 两阶段分配 机制实现渐进式重平衡，下面举个例子，假设有一个topic，共6个partition，一个消费者，其中有两个消费者再消费数据： 这时候有一个新的消费者加入，就需要重平衡。 如果是传统重平衡方式，Kafka 会： 让 ConsumerA 和 ConsumerB 释放所有分区（完全停止消费）。 Kafka 重新分配 P0 ~ P5 给 ConsumerA、ConsumerB、ConsumerC。 消费者重新开始消费（期间完全中断） 而如果是渐进式重平衡，则采用2阶段方式。 第一阶段：部分撤销 ConsumerA 和 ConsumerB 不会释放所有分区，而是部分释放。 Kafka 计算 最小变更 方案，并决定 ConsumerA、ConsumerB 各释放一个分区，给 ConsumerC 用。即可能是ConsumerA 释放 Partition-2；ConsumerB 释放 Partition-5 第二阶段：重新分配 Kafka 重新分配被释放的分区，则 ConsumerC 消费 Partition-2 和 Partition-5 这么样的重平衡过程，就可以减少所有分区都停止消费的情况，而只有其中部分分区需要重新分配而已。以上使用增加消费者的方式距离的，如果是减少消费者其实也一样。只有移除的消费者释放自己的分区，再重新分配给其他分区就好了，其他分区自己的那部分不用改变。

典型回答 Kafka 的重平衡机制是指在消费者组中新增或删除消费者时，Kafka 集群会重新分配主题分区给各个消费者，以保证每个消费者消费的分区数量尽可能均衡。 重平衡机制的目的是实现消费者的负载均衡和高可用性，以确保每个消费者都能够按照预期的方式消费到消息。 重平衡的 3 个触发条件： 消费者组成员数量发生变化。（新消费者的加入或者退出） 订阅主题（Topic）数量发生变化。 订阅主题的分区（Partition）数发生变化。 还有两种异常情况： 组协调器（Group Coordinator） 是 Kafka 负责管理消费者组的 Broker 节点。如果它崩溃或者发生故障，Kafka 需要重新选举新的 Group Coordinator，并进行重平衡。 当消费者组中的 Leader 消费者崩溃或退出。Kafka 需要选举新的 Leader，重新进行重平衡。 当Kafka 集群要触发重平衡机制时，大致的步骤如下： 暂停消费：在重平衡开始之前，Kafka 会暂停所有消费者的拉取操作，以确保不会出现重平衡期间的消息丢失或重复消费。 计算分区分配方案：Kafka 集群会根据当前消费者组的消费者数量和主题分区数量，计算出每个消费者应该分配的分区列表，以实现分区的负载均衡。 通知消费者：一旦分区分配方案确定，Kafka 集群会将分配方案发送给每个消费者，告诉它们需要消费的分区列表，并请求它们重新加入消费者组。 重新分配分区：在消费者重新加入消费者组后，Kafka 集群会将分区分配方案应用到实际的分区分配中，重新分配主题分区给各个消费者。 恢复消费：最后，Kafka 会恢复所有消费者的拉取操作，允许它们消费分配给自己的分区。 Kafka 的重平衡机制能够有效地实现消费者的负载均衡和高可用性，提高消息的处理能力和可靠性。但是，由于重平衡会带来一定的性能开销和不确定性，因此在设计应用时需要考虑到重平衡的影响，并采取一些措施来降低重平衡的频率和影响。 在重平衡过程中，所有 Consumer 实例都会停止消费，等待重平衡完成。但是目前并没有什么好的办法来解决重平衡带来的STW，只能尽量避免它的发生。 扩展知识 消费者的五种状态 Kafka的Consumer实例五种状态，分别是： 状态 描述 Empty 组内没有任何成员，但是消费者可能存在已提交的位移数据，而且这些位移尚未过期 Dead 同样是组内没有任何成员，但是组的元数据信息已经被协调者端移除，协调者保存着当前向他注册过的所有组信息 PreparingRebalance 消费者组准备开启重平衡，此时所有成员都需要重新加入消费者组 CompletingRebalance 消费者组下所有成员已经加入，各个成员中等待分配方案 Stable 消费者组的稳定状态，该状态表明重平衡已经完成，组内成员能够正常消费数据 状态的流转过程： 优化重平衡 重平衡会导致STW，应尽量减少问题发生，可以有以下几种优化方式， 默认情况下，消费者离开后会导致重平衡。但如果开启静态成员，Kafka 不会立即移除该消费者，而是等待一段时间（group.instance.id）。 这样，如果消费者重启，Kafka 仍然保持它的分区分配，不触发重平衡。 还有就是，Kafka 提供了多种分区分配策略，选择合适的策略可以减少重平衡的影响： ...

典型回答 Kafka 的存储理念非常简洁：它将所有收到的消息简单地以顺序追加（Append-Only）的方式写入磁盘文件。 这种利用顺序磁盘 I/O 的方式，这也是kafka性能好的重要原因之一。 ✅Kafka 为什么这么快？ Kafka 的存储结构是一个从逻辑概念到物理文件的层级映射关系： 逻辑概念：**<font style="color:rgb(64, 64, 64);background-color:rgb(236, 236, 236);">Topic</font>** -> **<font style="color:rgb(64, 64, 64);background-color:rgb(236, 236, 236);">Partition</font>** 物理文件：**<font style="color:rgb(64, 64, 64);background-color:rgb(236, 236, 236);">Partition</font>** -> **<font style="color:rgb(64, 64, 64);background-color:rgb(236, 236, 236);">Log Segment</font>** ****文件 ...

典型回答 使用消息队列的主要目的主要记住这几个关键词：解耦、异步、削峰填谷 解耦：在一个复杂的系统中，不同的模块或服务之间可能需要相互依赖，如果直接使用函数调用或者 API 调用的方式，会造成模块之间的耦合，当其中一个模块发生改变时，需要同时修改调用方和被调用方的代码。而使用消息队列作为中间件，不同的模块可以将消息发送到消息队列中，不需要知道具体的接收方是谁，接收方可以独立地消费消息，实现了模块之间的解耦。 异步：有些操作比较耗时，例如发送邮件、生成报表等，如果使用同步的方式处理，会阻塞主线程或者进程，导致系统的性能下降。而使用消息队列，可以将这些操作封装成消息，放入消息队列中，异步地处理这些操作，不影响主流程的执行，提高了系统的性能和响应速度。 削峰填谷：削峰填谷是一种在高并发场景下平衡系统压力的技术，通常用于平衡系统在高峰期和低谷期的资源利用率，提高系统的吞吐量和响应速度。在削峰填谷的过程中，通常使用消息队列作为缓冲区，将请求放入消息队列中，然后在系统负载低的时候进行处理。这种方式可以将系统的峰值压力分散到较长的时间段内，减少瞬时压力对系统的影响，从而提高系统的稳定性和可靠性。 另外消息队列还有以下优点： 可靠性高：消息队列通常具有高可靠性，可以实现消息的持久化存储、消息的备份和故障恢复等功能，保证消息不会丢失。 扩展性好：通过增加消息队列实例或者添加消费者实例，可以实现消息队列的水平扩展，提高系统的处理能力。 灵活性高：消息队列通常支持多种消息传递模式，如点对点模式和发布/订阅模式，可以根据不同的业务场景选择不同的模式。 扩展知识 消息队列实现 市面上有很多成熟的消息队列中间件可以供我们使用，其中比较常用的有kafka、activeMQ、RabbitMQ和RocketMQ等。 kafka、activeMQ、RabbitMQ和RocketMQ都有哪些区别？

kafka是一个成熟的消息队列，一直以性能高著称，它之所以能够实现高吞吐量和低延迟，主要是由于以下几个方面的优化，我试着从发送端，存储端以及消费端分别介绍一下。 消息发送 批量发送：Kafka 通过将多个消息打包成一个批次，减少了网络传输和磁盘写入的次数，从而提高了消息的吞吐量和传输效率。 异步发送：生产者可以异步发送消息，不必等待每个消息的确认，这大大提高了消息发送的效率。 消息压缩：支持对消息进行压缩，减少网络传输的数据量。 并行发送：通过将数据分布在不同的分区（Partitions）中，生产者可以并行发送消息，从而提高了吞吐量。 消息存储 零拷贝技术：Kafka 使用零拷贝技术来避免了数据的拷贝操作，降低了内存和 CPU 的使用率，提高了系统的性能。 ✅什么是零拷贝？ 磁盘顺序写入：Kafka把消息存储在磁盘上，且以顺序的方式写入数据。顺序写入比随机写入速度快很多，因为它减少了磁头寻道时间。避免了随机读写带来的性能损耗，提高了磁盘的使用效率。 页缓存：Kafka 将其数据存储在磁盘中，但在访问数据时，它会先将数据加载到操作系统的页缓存中，并在页缓存中保留一份副本，从而实现快速的数据访问。 稀疏索引：Kafka 存储消息是通过分段的日志文件，每个分段都有自己的索引文件。这些索引文件中的条目不是对分段中的每条消息都建立索引，而是每隔一定数量的消息建立一个索引点，这就构成了稀疏索引。稀疏索引减少了索引大小，使得加载到内存中的索引更小，提高了查找特定消息的效率。 分区和副本：Kafka 采用分区和副本的机制，可以将数据分散到多个节点上进行处理，从而实现了分布式的高可用性和负载均衡。 ✅介绍一下Kafka的ISR机制？ 消息消费 消费者群组：通过消费者群组可以实现消息的负载均衡和容错处理。 并行消费：不同的消费者可以独立地消费不同的分区，实现消费的并行处理。 批量拉取：Kafka支持批量拉取消息，可以一次性拉取多个消息进行消费。减少网络消耗，提升性能 生产消息 存储消息 消费消息 批量发送 磁盘顺序写入 消费者群组 异步发送 页缓存 批量拉取 消息压缩 稀疏索引 并行消费 并行发送 零拷贝 分区和副本