一、幂等性 1.1、Exactly-Once 保证在消息重发的时候，消费者不会重复处理。即使在消费者收到重复消息的时候，重复处理，也要保证最终结果的一致性。 所谓幂等性，数学概念就是：f(f(x)) = f(x) 。f 函数表示对消息的处理。 比如，银行转账，如果失败，需要重试。不管重试多少次，都要保证最终结果一定是一致的。 Exactly-Once：即精准一次，来保证幂等性。Exactly once 是 Kafka 从版本 0.11 之后提供的高级特性。 1.2、为什么需要幂等性 在使用 Kafka 时，需要确保 Exactly-Once 语义。分布式系统中，一些不可控因素有很多，比如网络、OOM、FullGC 等。在 Kafka Broker 确认 Ack 前，有可能出现网络异常、FullGC、OOM 等问题时导致 Ack 超时，Producer 会进行重复发送。注，在未达到最大重试次数前，会自动重试（非应用程序代码写的重试）。 Kafka 在引入幂等性之前，Producer 向 Broker 发送消息，然后 Broker 将消息追加到消息流中后给 Producer 返回 Ack 信号值。实现流程如下： 生产中，会出现各种不确定的因素，比如在 Producer 在发送给 Broker 的时候出现网络异常。比如以下这种异常情况的出现： 上图这种情况，当 Producer 第一次发送消息给 Broker 时，Broker 将消息 (x2,y2) 追加到了消息流中，但是在返回 Ack 信号给 Producer 时失败了（比如网络异常） 。此时，Producer 端触发重试机制，将消息 (x2,y2) 重新发送给 Broker，Broker 接收到消息后，再次将该消息追加到消息流中，然后成功返回 Ack 信号给 Producer。这样下来，消息流中就被重复追加了两条相同的 (x2,y2) 的消息。 因此需要保证幂等性保证即使多次发送也要让最终的结果一样。 1.3、如何实现幂等性 Kafka 在 0.11.0.0 之后加入的幂等性。他是通过添加唯一 ID，类似于数据库的主键，用于唯一标记一个消息。它在底层设计架构中引入了 ProducerID 和 SequenceNumber。 PID：ProducerID，每个生产者启动时，Kafka 都会给它分配一个 ID，ProducerID 是生产者的唯一标识，需要注意的是，Kafka 重启也会重新分配 PID 。 SequenceNumber：对于每个 ProducerID，Producer 发送数据的每个 Topic 和 Partition 都对应一个从 0 开始单调递增的 SequenceNumber 值。 同样，下图一种理想状态下的发送流程。实际情况下，会有很多不确定的因素，比如 Broker 在发送 Ack 信号给 Producer 时出现网络异常，导致发送失败。异常情况如下图所示： 当 Producer 发送消息 (x2,y2) 给 Broker 时，Broker 接收到消息并将其追加到消息流中。此时，Broker 返回 Ack 信号给 Producer 时，发生异常导致 Producer 接收 Ack 信号失败。对于 Producer 来说，会触发重试机制，将消息 (x2,y2) 再次发送，但是，由于引入了幂等性，在每条消息中附带了 PID（ProducerID）和 SequenceNumber。相同的 PID 和 SequenceNumber 发送给 Broker，而之前 Broker 缓存过之前发送的相同的消息，那么在消息流中的消息就只有一条 (x2,y2)，不会出现重复发送的情况。 1.4、使用幂等 通过如下配置使用幂等： enable.idempotence=true。enable.idempotence 配置项表示是否使用幂等性。当 enable.idempotence 配置为 true 时，acks 必须配置为 all。并且建议 max.in.flight.requests.per.connection 的值小于 5。 acks=all。 1.5、幂等性的限制条件 单独只使用 Producer 的幂等性是存在一些限制条件的： 只能保证 Producer 在单个会话内不丟不重 ，如果 Producer 出现意外挂掉再重启是无法保证的（幂等性情况下，是无法获取之前的状态信息，因此是无法做到跨会话级别的不丢不重）; 幂等性不能跨多个 Topic-Partition，只能保证单个 partition 内的幂等性 ，当涉及多个 Topic-Partition 时，这中间的状态并没有同步。 如果需要跨会话、跨多个 topic-partition 的情况，需要使用 Kafka 的事务性来实现。这种幂等性只是保证了再生产端实现了幂等性，在实际场景中往往需要在消息者端实现幂等性，可以最大程度避免重复消费。 1.6、幂等性的实现原理 每个新的 Producer 在初始化的时候会被分配一个唯一的 PID（凡是开启幂等性都是需要生成 PID，只不过未开启事务的 PID 可以在任意 broker 生成，而开启事务只能在 TransactionCoordinator 节点生成），该 PID 对用户完全透明而不会暴露给用户。Broker 端也会为每个 <PID, Topic, Partition> 维护一个序号，并且每次 Commit 一条消息时将其对应序号递增。对于接收的每条消息，如果其消息序号比 Broker 维护的序号（即最后一次 Commit 的消息的序号）大一，则 Broker 会接受它，否则将其丢弃： 如果消息序号比 Broker 维护的序号大于 1 以上，说明中间有数据尚未写入，也即乱序，此时 Broker 拒绝该消息，Producer 抛出 InvalidSequenceNumber。 如果消息序号小于等于 Broker 维护的序号，说明该消息已被保存，即为重复消息，Broker 直接丢弃该消息，Producer 抛出 DuplicateSequenceNumber。 上述设计解决了 0.11.0 之前版本中的两个问题： Broker 保存消息后，发送 ACK 前宕机，Producer 认为消息未发送成功并重试，造成数据重复。 前一条消息发送失败，后一条消息发送成功，前一条消息重试后成功，造成数据乱序。 producer_id 是从 Kafka 服务端请求获取的（通过 ProducerIdManager 的 generateProducerId () 方法产生，维护在 zk 中的 /latest_producer_id_block 节点），消息序列号是 Producer 端生成的，初始值为 0，之后自增加 1，每个分区都有独立的序列号。这里需要说明下，Kafka 发送消息都是以 batch 的格式发送，batch 包含了多条消息。所以 Producer 发送消息 batch 的时候，只会设置该 batch 的第一个消息的序列号，后面消息的序列号可以根据第一个消息的序列号计算出来。 二、事务 2.1、为什么需要事务 Kafka 的 Exactly Once 幂等性只能保证单次会话内的精准一次性，不能解决跨会话和跨分区的问题。 假如有如下问题： 1、producer 发的多条消息组成一个事务，这些消息需要对 consumer 同时可见或者同时不可见。 2、producer 可能会给多个 topic，多个 partition 发消息，这些消息也需要能放在一个事务里面，这就形成了一个典型的分布式事务。 3、kafka 的应用场景经常是应用先消费一个 topic，然后做处理再发到另一个 topic，这个 consume-transform-produce 过程需要放到一个事务里面，比如在消息处理或者发送的过程中如果失败了，消费偏移量也不能提交。 4、producer 或者 producer 所在的应用可能会挂掉，新的 producer 启动以后需要知道怎么处理之前未完成的事务 。 在一个原子操作中，根据包含的操作类型，可以分为三种情况： 1、只有 Producer 生产消息。 2、消费消息和生产消息并存，这个是事务场景中最常用的情况，就是我们常说的 consume-transform-produce 模式。 3、只有 consumer 消费消息。 前两种情况需要引入事务，第 3 种情况不需要引入，这种操作其实没有什么意义，跟使用手动提交效果一样，而且也不是事务属性引入的目的。 2.2、开启事务 对于 Producer，需要设置 transactional.id 属性，这个属性的作用下文会提到。设置了 transactional.id 属性后，enable.idempotence 属性会自动设置为 true。 对于 Consumer，需要设置 isolation.level = read_committed，这样 Consumer 只会读取已经提交了事务的消息。另外，需要设置 enable.auto.commit = false 来关闭自动提交 Offset 功能。 2.3、事务保证 2.3.1、事务恢复的保证 为了实现有状态的应用也可以保证重启后从断点处继续处理，也即事务恢复。应用程序必须提供一个稳定的（重启后不变）唯一的 ID，也即 Transaction ID。Transactin ID 与 PID 可能一一对应。区别在于 Transaction ID 由用户提供，而 PID 是内部的实现对用户透明。 另外，为了保证新的 Producer 启动后，旧的具有相同 Transaction ID 的 Producer 即失效，每次 Producer 通过 Transaction ID 拿到 PID 的同时，还会获取一个单调递增的 epoch。由于旧的 Producer 的 epoch 比新 Producer 的 epoch 小，Kafka 可以很容易识别出该 Producer 是老的 Producer 并拒绝其请求。 有了 Transaction ID 和 epoch 后，Kafka 可保证： 跨 Session 的数据幂等发送。当具有相同 Transaction ID 的新的 Producer 实例被创建且工作时，旧的且拥有相同 Transaction ID 的 Producer 将不再工作。 跨 Session 的事务恢复。如果某个应用实例宕机，新的实例可以保证任何未完成的旧的事务要么 Commit 要么 Abort，使得新实例从一个正常状态开始工作。 2.3.2、事务原子性的保证 事务原子性是指 Producer 将多条消息作为一个事务批量发送，要么全部成功要么全部失败。 引入了一个服务器端的模块，名为 Transaction Coordinator，用于管理 Producer 发送的消息的事务性。 该 Transaction Coordinator 维护 Transaction Log，该 log 存于一个内部的 Topic 内。由于 Topic 数据具有持久性，因此事务的状态也具有持久性。 Producer 并不直接读写 Transaction Log，它与 Transaction Coordinator 通信，然后由 Transaction Coordinator 将该事务的状态插入相应的 Transaction Log。 Transaction Log 的设计与 Offset Log 用于保存 Consumer 的 Offset 类似。 2.3.3、事务中 Offset 的提交保证 在 Kafka Stream 应用中同时包含 Consumer 和 Producer（即 Consumer-Transform-Producer），前者负责从 Kafka 中获取消息，后者负责将处理完的数据写回 Kafka 的其它 Topic 中。 为了实现该场景下的事务的原子性，Kafka 需要保证对 Consumer Offset 的 Commit 与 Producer 对发送消息的 Commit 包含在同一个事务中。否则，如果在二者 Commit 中间发生异常，根据二者 Commit 的顺序可能会造成数据丢失和数据重复： 如果先 Commit Producer 发送数据的事务再 Commit Consumer 的 Offset，即 At Least Once 语义，可能造成数据重复。 如果先 Commit Consumer 的 Offset，再 Commit Producer 数据发送事务，即 At Most Once 语义，可能造成数据丢失。 2.3.4、用于事务特性的控制型消息 为了区分写入 Partition 的消息被 Commit 还是 Abort，Kafka 引入了一种特殊类型的消息，即 Control Message。该类消息的 Value 内不包含任何应用相关的数据，并且不会暴露给应用程序。它只用于 Broker 与 Client 间的内部通信。 对于 Producer 端事务，Kafka 以 Control Message 的形式引入一系列的 Transaction Marker。Consumer 即可通过该标记判定对应的消息被 Commit 了还是 Abort 了，然后结合该 Consumer 配置的隔离级别决定是否应该将该消息返回给应用程序。 Kafka 事务的回滚，并不是删除已写入的数据，而是将写入数据的事务标记为 Rollback/Abort 从而在读数据时过滤该数据。 2.4、事务流程 事务原理流程如下： 上图中的 Transaction Coordinator 运行在 Kafka 服务端，下面简称 TC 服务。 __transaction_state 是 TC 服务持久化事务信息的 topic 名称，下面简称事务 topic。 Producer 向 TC 服务发送的 commit 消息，下面简称事务提交消息。 TC 服务向分区发送的消息，下面简称事务结果消息。 1、寻找 TC 服务地址 Producer 会首先从 Kafka 集群中选择任意一台机器，然后向其发送请求，获取 TC 服务的地址。Kafka 有个特殊的事务 topic，名称为__transaction_state ，负责持久化事务消息。这个 topic 有多个分区，默认有 50 个，每个分区负责一部分事务。事务划分是根据 transaction id， 计算出该事务属于哪个分区。这个分区的 leader 所在的机器，负责这个事务的 TC 服务地址。 2、事务初始化 Producer 在使用事务功能，必须先自定义一个唯一的 transaction id。有了 transaction id，即使客户端挂掉了，它重启后也能继续处理未完成的事务。 Kafka 实现事务需要依靠幂等性，而幂等性需要指定 producer id 。所以 Producer 在启动事务之前，需要向 TC 服务申请 producer id。TC 服务在分配 producer id 后，会将它持久化到事务 topic。 3、发送消息 Producer 在接收到 producer id 后，就可以正常的发送消息了。不过发送消息之前，需要先将这些消息的分区地址，上传到 TC 服务。TC 服务会将这些分区地址持久化到事务 topic。然后 Producer 才会真正的发送消息，这些消息与普通消息不同，它们会有一个字段，表示自身是事务消息。 这里需要注意下一种特殊的请求，提交消费位置请求，用于原子性的从某个 topic 读取消息，并且发送消息到另外一个 topic。我们知道一般是消费者使用消费组订阅 topic，才会发送提交消费位置的请求，而这里是由 Producer 发送的。Producer 首先会发送一条请求，里面会包含这个消费组对应的分区（每个消费组的消费位置都保存在 __consumer_offset topic 的一个分区里），TC 服务会将分区持久化之后，发送响应。Producer 收到响应后，就会直接发送消费位置请求给 GroupCoordinator。 4、发送提交请求 Producer 发送完消息后，如果认为该事务可以提交了，就会发送提交请求到 TC 服务。Producer 的工作至此就完成了，接下来它只需要等待响应。这里需要强调下，Producer 会在发送事务提交请求之前，会等待之前所有的请求都已经发送并且响应成功。 5、提交请求持久化 TC 服务收到事务提交请求后，会先将提交信息先持久化到事务 topic 。持久化成功后，服务端就立即发送成功响应给 Producer。然后找到该事务涉及到的所有分区，为每 个分区生成提交请求，存到队列里等待发送。 在一般的二阶段提交中，协调者需要收到所有参与者的响应后，才能判断此事务是否成功，最后才将结果返回给客户。那如果 TC 服务在发送响应给 Producer 后，还没来及向分区发送请求就挂掉了，那么 Kafka 是如何保证事务完成。因为每次事务的信息都会持久化，所以 TC 服务挂掉重新启动后，会先从 事务 topic 加载事务信息，如果发现只有事务提交信息，却没有后来的事务完成信息，说明存在事务结果信息没有提交到分区。 6、发送事务结果信息给分区 后台线程会不停的从队列里，拉取请求并且发送到分区。当一个分区收到事务结果消息后，会将结果保存到分区里，并且返回成功响应到 TC 服务。当 TC 服务收到所有分区的成功响应后，会持久化一条事务完成的消息到事务 topic。至此，一个完整的事务流程就完成了。 2.5、事务原理 2.5.1、FindCoordinatorRequest 对应图中的 1。 由于 Transaction Coordinator 是分配 PID 和管理事务的核心，因此 Producer 要做的第一件事情就是通过向任意一个 Broker 发送 FindCoordinator 请求找到 Transaction Coordinator 的位置。 注意：只有应用程序为 Producer 配置了 Transaction ID 时才可使用事务特性，也才需要这一步。另外，由于事务性要求 Producer 开启幂等特性，因此通过将 transactional.id 设置为非空从而开启事务特性的同时也需要通过将 enable.idempotence 设置为 true 来开启幂等特性。 通过请求附带的事务 ID，计算出 __transaction_state 的分区 ID，而其对应的 Leader 副本 Broker 即是负责当前事务的 Transaction Coordinator。 2.5.2、InitProducerIdRequest 对应图中的 2 。 找到 Transaction Coordinator 后，具有幂等特性的 Producer 必须发起 InitPidRequest 请求以获取 PID。 注意：只要开启了幂等特性即必须执行该操作，而无须考虑该 Producer 是否开启了事务特性。 2.5.3、如果事务特性被开启 InitPidRequest 会发送给 Transaction Coordinator。如果 Transaction Coordinator 是第一次收到包含有该 Transaction ID 的 InitPidRequest 请求，它将会把该 < TransactionID, PID > 存入 Transaction Log，如上图中步骤 2.1 所示。这样可保证该对应关系被持久化，从而保证即使 Transaction Coordinator 宕机该对应关系也不会丢失。 除了返回 PID 外，InitPidRequest 还会执行如下任务： 增加该 PID 对应的 epoch。具有相同 PID 但 epoch 小于该 epoch 的其它 Producer（如果有）新开启的事务将被拒绝。 恢复（Commit 或 Abort）之前的 Producer 未完成的事务（如果有）。 注意：InitPidRequest 的处理过程是同步阻塞的。一旦该调用正确返回，Producer 即可开始新的事务。 另外，如果事务特性未开启，InitPidRequest 可发送至任意 Broker，并且会得到一个全新的唯一的 PID。该 Producer 将只能使用幂等特性以及单一 Session 内的事务特性，而不能使用跨 Session 的事务特性。 2.5.4、开启事务回话 Kafka 从 0.11.0.0 版本开始，提供 beginTransaction () 方法用于开启一个事务。调用该方法后，Producer 本地会记录已经开启了事务（标记生产者状态机处于 IN_TRANSACTION 状态），但 Transaction Coordinator 只有在 Producer 发送第一条消息后才认为事务已经开启。 2.5.5、流的处理与转发阶段 Consume-Transform-Produce 模式。这一阶段，包含了整个事务的数据处理过程，并且包含了多种请求。 1、AddPartitionsToTxnRequest 一个 Producer 可能会给多个 < Topic, Partition > 发送数据，给一个新的 < Topic, Partition > 发送数据前，它需要先向 Transaction Coordinator 发送 AddPartitionsToTxnRequest。 Transaction Coordinator 会将该 < Transaction, Topic, Partition > 存于 Transaction Log 内，并将其状态置为 BEGIN，如上图中步骤 4.1 所示。有了该信息后，我们才可以在后续步骤中为每个 Topic, Partition > 设置 COMMIT 或者 ABORT 标记（如上图中步骤 5.2 所示）。 另外，如果该 < Topic, Partition > 为该事务中第一个 < Topic, Partition>，Transaction Coordinator 还会启动对该事务的计时（每个事务都有自己的超时时间）。 2、ProduceRequest Producer 通过一个或多个 ProduceRequest 发送一系列消息。除了应用数据外，该请求还包含了 PID，epoch，和 Sequence Number。该过程如上图中步骤 4.2 所示。 3、AddOffsetsToTxnRequest 为了提供事务性，Producer 新增了 sendOffsetsToTransaction 方法，该方法将多组消息的发送和消费放入同一批处理内。 该方法先判断在当前事务中该方法是否已经被调用并传入了相同的 Group ID。若是，直接跳到下一步；若不是，则向 Transaction Coordinator 发送 AddOffsetsToTxnRequests 请求，Transaction Coordinator 将对应的所有 < Topic, Partition > 存于 Transaction Log 中，并将其状态记为 BEGIN，如上图中步骤 4.3 所示。该方法会阻塞直到收到响应。 4、TxnOffsetCommitRequest 作为 sendOffsetsToTransaction 方法的一部分，在处理完 AddOffsetsToTxnRequest 后，Producer 也会发送 TxnOffsetCommit 请求给 Consumer Coordinator 从而将本事务包含的与读操作相关的各 < Topic, Partition > 的 Offset 持久化到内部的__consumer_offsets 中，如上图步骤 4.4 所示。 在此过程中，Consumer Coordinator 会通过 PID 和对应的 epoch 来验证是否应该允许该 Producer 的该请求。 这里需要注意： （1）写入__consumer_offsets 的 Offset 信息在当前事务 Commit 前对外是不可见的。也即在当前事务被 Commit 前，可认为该 Offset 尚未 Commit，也即对应的消息尚未被完成处理。 （2）Consumer Coordinator 并不会立即更新缓存中相应 < Topic, Partition > 的 Offset，因为此时这些更新操作尚未被 COMMIT 或 ABORT。 2.5.6、提交或回滚事务 1、EndTxnRequest（对应 5.1） 当用户调用 KafkaProducer#commitTransaction 或者 abortTransaction 方法，生产者会往 Transaction Coordinator 发送附带提交或中止的事务结果的 EndTxnRequest 请求。 当 Transaction Coordinator 在收到请求后， 把 PREPARE 消息写到 __transaction_state。（对应 5.1a） 通过 WriteTxnMarkersRequest 请求，向事务关联的所有 TopicPartitions 主副本写入 EndTransactionMarker 标记。（详细见下文） 最终，把封装了 COMMITTED 或 ABORTED 状态的 EndTransactionMarker 标记写到 __transaction_state。（对应 5.3a） 2、WriteTxnMarkersRequest（对应 5.2a） Transaction Coordinator 向关联的 TopicPartitions 主副本提交 WriteTxnMarkersRequest 请求，请求中将附带生产者 ID，以用于过滤掉交叉不相关联生产者的日志。 在日后，当消费者读取某生产者的 Aborted 段日志时，可通过上文提及的 .txnindex 索引文件提前过滤，而读取 Committed 段的则无需格外处理。（具体参考上文的设计分解） 另外，如果 __consumer_offsets 也作为事务的一部分，同样写入 EndTransactionMarker 标记并更新 Offsets 可见性。 2.5.7、超时事务中止 默认情况，根据事务的起始时间戳，Transaction Coordinator 每 10s 轮询进行中的事务是否已超时，若发现超时事务，将推进 Epoch、中止当前事务（相当于 Transaction Coordinator 作为新的生产者）；而在未来，老的生产者将收到 ProducerFencedException 异常。 2.6、拒绝僵尸实例 在分布式系统中，一个 instance 的宕机或失联，集群往往会自动启动一个新的实例来代替它的工作。此时若原实例恢复了，那么集群中就产生了两个具有相同职责的实例，此时前一个 instance 就被称为 “僵尸实例（Zombie Instance）”。在 Kafka 中，两个相同的 producer 同时处理消息并生产出重复的消息（read-process-write 模式），这样就严重违反了 Exactly Once Processing 的语义。这就是僵尸实例问题。 Kafka 事务特性通过 transaction-id 属性来解决僵尸实例问题。所有具有相同 transaction-id 的 Producer 都会被分配相同的 pid，同时每一个 Producer 还会被分配一个递增的 epoch。Kafka 收到事务提交请求时，如果检查当前事务提交者的 epoch 不是最新的，那么就会拒绝该 Producer 的请求。从而达成拒绝僵尸实例的目标。 2.7、与分布式事务机制对比 2.7.1、两阶段提交原理 二阶段提交的算法思路可以概括为：协调者询问参与者是否准备好了提交，并根据所有参与者的反馈情况决定向所有参与者发送 commit 或者 rollback 指令（协调者向所有参与者发送相同的指令）。 所谓的两个阶段是指 准备阶段 又称投票阶段。在这一阶段，协调者询问所有参与者是否准备好提交，参与者如果已经准备好提交则回复 Prepared，否则回复 Non-Prepared。 提交阶段 又称执行阶段。协调者如果在上一阶段收到所有参与者回复的 Prepared，则在此阶段向所有参与者发送 commit 指令，所有参与者立即执行 commit 操作；否则协调者向所有参与者发送 rollback 指令，参与者立即执行 rollback 操作。 两阶段提交中，协调者和参与方的交互过程如下图所示： 2.7.2、Kafka 两阶段提交对比 Kafka 的事务机制与上述所介绍的两阶段提交机制看似相似，都分 PREPARE 阶段和最终 COMMIT 阶段，但又有很大不同。 Kafka 事务机制中，PREPARE 时即要指明是 PREPARE_COMMIT 还是 PREPARE_ABORT，并且 只须在 Transaction Log 中标记即可 ，无须其它组件参与。而两阶段提交的 PREPARE 需要发送给所有的分布式事务参与方，并且事务参与方需要尽可能准备好，并根据准备情况返回 Prepared 或 Non-Prepared 状态给事务管理器。 Kafka 事务中，一但发起 PREPARE_COMMIT 或 PREPARE_ABORT，则确定该事务最终的结果应该是被 COMMIT 或 ABORT 。而分布式事务中，PREPARE 后由各事务参与方返回状态，只有所有参与方均返回 Prepared 状态才会真正执行 COMMIT，否则执行 ROLLBACK Kafka 事务机制中，某几个 Partition 在 COMMIT 或 ABORT 过程中变为不可用，只影响该 Partition 不影响其它 Partition。两阶段提交中，若唯一收到 COMMIT 命令参与者 Crash，其它事务参与方无法判断事务状态从而使得整个事务阻塞。 Kafka 事务机制引入事务超时机制，有效避免了挂起的事务影响其它事务的问题。 Kafka 事务机制中存在多个 Transaction Coordinator 实例，而分布式事务中只有一个事务管理器。 2.8、事务操作 Kafka 事务操作常用的 API 如下： // 初始化事务，需要注意确保 transation.id 属性被分配 void initTransactions (); // 开启事务 void beginTransaction () throws ProducerFencedException; // 为 Consumer 提供的在事务内 Commit Offsets 的操作 void sendOffsetsToTransaction (Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) throws ProducerFencedException; // 提交事务 void commitTransaction () throws ProducerFencedException; // 放弃事务，类似于回滚事务的操作 void abortTransaction () throws ProducerFencedException; 案例 1： 单个 Producer，使用事务保证消息的仅一次发送： import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; import java.util.HashMap; import java.util.Map; public class MyTransactionalProducer { public static void main (String [] args) { Map<String, Object> configs = new HashMap<>(); configs.put (ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092"); configs.put (ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); configs.put (ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); // 提供客户端 ID configs.put (ProducerConfig.CLIENT_ID_CONFIG, "tx_producer"); // 事务 ID configs.put (ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my_tx_id"); // 要求 ISR 都确认 configs.put (ProducerConfig.ACKS_CONFIG, "all"); KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs); // 初始化事务 producer.initTransactions (); // 开启事务 producer.beginTransaction (); try { //producer.send (new ProducerRecord<>("tp_tx_01", "tx_msg_01")); producer.send (new ProducerRecord<>("tp_tx_01", "tx_msg_02")); //int i = 1 / 0; // 提交事务 producer.commitTransaction (); } catch (Exception ex) { // 中止事务 producer.abortTransaction (); } finally { // 关闭生产者 producer.close (); } } } 案例 2： 在 消费 - 转换 - 生产 模式，使用事务保证仅一次发送： import org.apache.kafka.clients.consumer.*; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; import java.util.Collections; import java.util.HashMap; import java.util.Map; public class MyTransactional { public static KafkaProducer<String, String> getProducer () { Map<String, Object> configs = new HashMap<>(); configs.put (ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092"); configs.put (ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); configs.put (ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); // 设置 client.id configs.put (ProducerConfig.CLIENT_ID_CONFIG, "tx_producer_01"); // 设置事务 id configs.put (ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx_id_02"); // 需要所有的 ISR 副本确认 configs.put (ProducerConfig.ACKS_CONFIG, "all"); // 启用幂等性 configs.put (ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs); return producer; } public static KafkaConsumer<String, String> getConsumer (String consumerGroupId) { Map<String, Object> configs = new HashMap<>(); configs.put (ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092"); configs.put (ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); configs.put (ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); // 设置消费组 ID configs.put (ConsumerConfig.GROUP_ID_CONFIG, "consumer_grp_02"); // 不启用消费者偏移量的自动确认，也不要手动确认 configs.put (ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); configs.put (ConsumerConfig.CLIENT_ID_CONFIG, "consumer_client_02"); configs.put (ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 只读取已提交的消息 //configs.put (ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"); KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(configs); return consumer; } public static void main (String [] args) { String consumerGroupId = "consumer_grp_id_101"; KafkaProducer<String, String> producer = getProducer (); KafkaConsumer<String, String> consumer = getConsumer (consumerGroupId); // 事务的初始化 producer.initTransactions (); // 订阅主题 consumer.subscribe (Collections.singleton ("tp_tx_01")); final ConsumerRecords<String, String> records = consumer.poll (1_000); // 开启事务 producer.beginTransaction (); try { Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(); for (ConsumerRecord<String, String> record : records) { System.out.println (record); producer.send (new ProducerRecord<String, String>("tp_tx_out_01", record.key (), record.value ())); offsets.put (new TopicPartition (record.topic (), record.partition ()), new OffsetAndMetadata (record.offset () + 1)); // 偏 移量表示下一条要消费的消息 } // 将该消息的偏移量提交作为事务的一部分，随事务提交和回滚（不提交消费偏移 量） producer.sendOffsetsToTransaction (offsets, consumerGroupId); // 提交事务 producer.commitTransaction (); } catch (Exception e) { e.printStackTrace (); // 回滚事务 producer.abortTransaction (); } finally { // 关闭资源 producer.close (); consumer.close (); } } } 三、相关配置 3.1、Broker configs transactional.id.timeout.ms 在 ms 中，事务协调器在生产者 TransactionalId 提前过期之前等待的最长时间，并且没有从该生产者 TransactionalId 接收到任何事务状态更新。默认是 604800000 (7 天)。这允许每周一次的生产者作业维护它们的 id max.transaction.timeout.ms 事务允许的最大超时。如果客户端请求的事务时间超过此时间，broke 将在 InitPidRequest 中返回 InvalidTransactionTimeout 错误。这可以 防止客户机超时过大，从而导致用户无法从事务中包含的主题读取内容。默认值为 900000 (15 分钟)。这是消息事务需要发送的时间的保守上限。 transaction.state.log.replication.factor 事务状态 topic 的副本数量。默认值：3 transaction.state.log.num.partitions 事务状态主题的分区数。默认值：50 transaction.state.log.min.isr 事务状态主题的每个分区 ISR 最小数量。默认值：2 transaction.state.log.segment.bytes 事务状态主题的 segment 大小。默认值：104857600 字节 3.2、Producer configs enable.idempotence 开启幂等 transaction.timeout.ms 事务超时时间 事务协调器在主动中止正在进行的事务之前等待生产者更新事务状态的最长时间。这个配置值将与 InitPidRequest 一起发送到事务 协调器。如果该值大于 max.transaction.timeout：在 broke 中设置 ms 时，请求将失败，并出现 InvalidTransactionTimeout 错误。默认是 60000。这使得交易不会阻塞下游消费超过一分钟，这在实时应用程序中通常是允许的。 transactional.id 用于事务性交付的 TransactionalId。这支持跨多个生产者会话的可靠性语义，因为它允许客户端确保使用相同 TransactionalId 的事务在启动任何新事务之前已经完成。如果没有提供 TransactionalId，则生产者仅限于幂等交付。 3.3、Consumer configs isolation.level 隔离级别。 read_uncommitted: 以偏移顺序使用已提交和未提交的消息。 read_committed: 仅以偏移量顺序使用非事务性消息或已提交事务性消息。为了维护偏移排序，这个设置意味着我们必须在使用者中缓冲消息，直到看到给定事务中的所有消息。 四、总结 KAFKA 的事务机制，在底层依赖于幂等生产者，幂等生产者是 kafka 事务的必要不充分条件； 开启 kafka 事务时，kafka 会自动开启幂等生产者。 Transaction Marker 与 PID 提供了识别消息是否应该被读取的能力，从而实现了事务的隔离性。 通过事务机制，KAFKA 实现了对多个 topic 的多个 partition 的原子性的写入（Atomic multi-partition writes）； KAFKA 的事务机制，在底层依赖于幂等生产者，幂等生产者是 kafka 事务的必要不充分条件：用户可以根据需要，配置使用幂等生产者但不开启事务；也可以根据需要开启 kafka 事务，此时 kafka 会使用幂等生产者； 为支持事务机制，KAFKA 引入了两个新的组件：Transaction Coordinator 和 Transaction Log，其中 transaction coordinator 是运行在每个 kafka broker 上的一个模块，是 kafka broker 进程承载的新功能之一（不是一个独立的新的进程）；而 transaction log 是 kakafa 的一个内部 topic； 为支持事务机制，kafka 将日志文件格式进行了扩展：日志中除了普通的消息，还有一种消息专门用来标志一个事务的结束，它就是控制消息 controlBatch, 它有两种类型：commit 和 abort，分别用来表征事务已经成功提交或已经被成功终止。 开启了事务的生产者，生产的消息最终还是正常写到目标 topic 中，但同时也会通过 transaction coordinator 使用两阶段提交协议，将事务状态标记 transaction marker，也就是控制消息 controlBatch，写到目标 topic 中，控制消息共有两种类型 commit 和 abort，分别用来表征事务已经成功提交或已经被成功终止； 开启了事务的消费者，如果配置读隔离级别为 read-committed, 在内部会使用存储在目标 topic-partition 中的事务控制消息，来过滤掉没有提交的消息，包括回滚的消息和尚未提交的消息，从而确保只读到已提交的事务的 message； 开启了事务的消费者，过滤消息时，KAFKA consumer 不需要跟 transactional coordinator 进行 rpc 交互，因为 topic 中存储的消息，包括正常的数据消息和控制消息，包含了足够的元数据信息来支持消息过滤； 当然 kakfa 的 producer 和 consumer 是解耦的，你也可以使用非 transactional consumer 来消费 transactional producer 生产的消息，此时目标 topic-partition 中的所有消息都会被返回，不会进行过滤，此时也就丢失了事务 ACID 的支持；
Kafka 分区副本重分配、数据迁移、副本扩缩容
一、脚本的使用介绍 该脚本是 kafka 提供用来重新分配分区的脚本工具； 1.1、生成推荐配置脚本 关键参数 --generate。 在进行分区副本重分配之前，最好是用下面方式获取一个合理的分配文件； 编写 move-json-file.json 文件；这个文件就是告知想对哪些 Topic 进行重新分配的计算。 { "topics": [ {"topic": "test_create_topic1"} ], "version": 1 } 然后执行下面的脚本，--broker-list "0,1,2,3" 这个参数是你想要分配的 Brokers。 sh bin/kafka-reassign-partitions.sh --zookeeper xxx:2181 --topics-to-move-json-file config/move-json-file.json --broker-list "0,1,2,3" --generate 执行完毕之后会打印： Current partition replica assignment// 当前副本分配方式 {"version":1,"partitions":[{"topic":"test_create_topic1","partition":2,"replicas":[1],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":1,"replicas":[3],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":0,"replicas":[2],"log_dirs":["any"]}]} Proposed partition reassignment configuration// 期望的重新分配方式 {"version":1,"partitions":[{"topic":"test_create_topic1","partition":2,"replicas":[2],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":1,"replicas":[1],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":0,"replicas":[0],"log_dirs":["any"]}]} 需求注意的是，此时分区移动尚未开始，它只是告诉你当前的分配和建议。保存当前分配，以防你想要回滚它。 1.2. 执行 Json 文件 关键参数 --execute。 将上面得到期望的重新分配方式文件保存在一个 json 文件里面 reassignment-json-file.json。 {"version":1,"partitions":[{"topic":"test_create_topic1","partition":2,"replicas":[2],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":1,"replicas":[1],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":0,"replicas":[0],"log_dirs":["any"]}]} 然后执行 sh bin/kafka-reassign-partitions.sh --zookeeper xxxxx:2181 --reassignment-json-file config/reassignment-json-file.json --execute 迁移过程注意流量陡增对集群的影响： Kafka 提供一个 broker 之间复制传输的流量限制，限制了副本从机器到另一台机器的带宽上限，当重新平衡集群，引导新 broker，添加或移除 broker 时候，这是很有用的。因为它限制了这些密集型的数据操作从而保障了对用户的影响、 例如上面的迁移操作加一个限流选项 --throttle 50000000。 在后面加上一个 --throttle 50000000 参数，那么执行移动分区的时候，会被限制流量在 50000000 B/s 加上参数后可以看到 The throttle limit was set to 50000000 B/s Successfully started reassignment of partitions. 需要注意的是，如果你迁移的时候包含副本跨路径迁移 (同一个 Broker 多个路径) 那么这个限流措施不会生效，需要再加上 --replica-alter-log-dirs-throttle 这个限流参数，它限制的是同一个 Broker 不同路径直接迁移的限流。 如果想在重新平衡期间修改限制，增加吞吐量，以便完成的更快。可以重新运行 execute 命令，用相同的 reassignment-json-file。 在重新平衡期间，管理员可以使用指标监控复制是否正在进行： kafka.server:type=FetcherLagMetrics,name=ConsumerLag,clientId=([-.\w]+),topic=([-.\w]+),partition=([0-9]+) 在复制过程中，滞后应该不断减少。如果指标没有减少，管理员应该如上所述增加节流吞吐量。 1.3、验证 关键参数 --verify。 该选项用于检查分区重新分配的状态，同时 --throttle 流量限制也会被移除掉；否则可能会导致定期复制操作的流量也受到限制。 sh bin/kafka-reassign-partitions.sh --zookeeper xxxx:2181 --reassignment-json-file config/reassignment-json-file.json --verify 二、副本扩缩 kafka 并没有提供一个专门的脚本来支持副本的扩缩，不像 kafka-topic.sh 脚本一样，是可以扩分区的；想要对副本进行扩缩，只能是曲线救国了，利用 kafka-reassign-partitions.sh 来重新分配副本。 2.1、副本扩容 假设我们当前的情况是 3 分区 1 副本，为了提供可用性，我想把副本数升到 2。 2.1.1、计算副本分配方式 用步骤 1.1 的 --generate 获取一下当前的分配情况，得到如下 json。 { "version": 1, "partitions": [{ "topic": "test_create_topic1", "partition": 2, "replicas": [2], "log_dirs": ["any"] }, { "topic": "test_create_topic1", "partition": 1, "replicas": [1], "log_dirs": ["any"] }, { "topic": "test_create_topic1", "partition": 0, "replicas": [0], "log_dirs": ["any"] }] } 我们想把所有分区的副本都变成 2, 那我们只需修改 "replicas": [] 里面的值了，这里面是 Broker 列表，排在第一个的是 Leader，所以我们根据自己想要的分配规则修改一下 json 文件就变成如下。 { "version": 1, "partitions": [{ "topic": "test_create_topic1", "partition": 2, "replicas": [2,0], "log_dirs": ["any","any"] }, { "topic": "test_create_topic1", "partition": 1, "replicas": [1,2], "log_dirs": ["any","any"] }, { "topic": "test_create_topic1", "partition": 0, "replicas": [0,1], "log_dirs": ["any","any"] }] } 注意 log_dirs 里面的数量要和 replicas 数量匹配；或者直接把 log_dirs 选项删除掉；这个 log_dirs 是副本跨路径迁移时候的绝对路径。 2.1.2、执行–execute 如果想在重新平衡期间修改限制，增加吞吐量，以便完成的更快。可以重新运行 execute 命令，用相同的 reassignment-json-file。 2.1.3、验证–verify 副本数量增加了。 2.2、副本缩容 副本缩容跟扩容是一个意思；当副本分配少于之前的数量时候，多出来的副本会被删除。 比如刚刚新增了一个副本，想重新恢复到一个副本。 { "version": 1, "partitions": [{ "topic": "test_create_topic1", "partition": 2, "replicas": [2], "log_dirs": ["any"] }, { "topic": "test_create_topic1", "partition": 1, "replicas": [1], "log_dirs": ["any"] }, { "topic": "test_create_topic1", "partition": 0, "replicas": [0], "log_dirs": ["any"] }] } 执行之后可以看到其他的副本就被标记为删除了。一会就会被清理掉。 用这样一种方式我们虽然是实现了副本的扩缩容，但是副本的分配需要我们自己来把控好，要做到负载均衡等等；那肯定是没有 kafka 自动帮我们分配比较合理一点。 --generate 本质上也是调用了这个方法： dminUtils.assignReplicasToBrokers (brokerMetadatas, assignment.size, replicas.size) 三、分区扩容 kafka 的分区扩容是 kafka-topis.sh 脚本实现的，不支持缩容。 四、分区迁移 分区迁移跟上面同理，请看 1.1,1.2,1.3 部分。 五、副本跨路径迁移 为什么线上 Kafka 机器各个磁盘间的占用不均匀，经常出现 “一边倒” 的情形？ 这是因为 Kafka 只保证分区数量在各个磁盘上均匀分布，但它无法知晓每个分区实际占用空间，故很有可能出现某些分区消息数量巨大导致占用大量磁盘空间的情况。在 1.1 版本之前，用户对此毫无办法，因为 1.1 之前 Kafka 只支持分区数据在不同 broker 间的重分配，而无法做到在同一个 broker 下的不同磁盘间做重分配。1.1 版本正式支持副本在不同路径间的迁移。 怎么在一台 Broker 上用多个路径存放分区呢？ ############################# Log Basics ############################## A comma separated list of directories under which to store log fileslog.dirs=kafka-logs-5,kafka-logs-6,kafka-logs-7,kafka-logs-8 注意同一个 Broker 上不同路径只会存放不同的分区，而不会将副本存放在同一个 Broker; 不然那副本就没有意义了 (容灾) 怎么针对跨路径迁移呢？ 迁移的 json 文件有一个参数是 log_dirs; 默认请求不传的话 它是 "log_dirs": ["any"] （这个数组的数量要跟副本保持一致） 但是你想实现跨路径迁移，只需要在这里填入绝对路径就行了，例如下面。 迁移的 json 文件示例： { "version": 1, "partitions": [ { "topic": "test_create_topic4", "partition": 2, "replicas": [ 0 ], "log_dirs": [ "/Users/xxxxx/work/IdeaPj/source/kafka/kafka-logs-5" ] }, { "topic": "test_create_topic4", "partition": 1, "replicas": [ 0 ], "log_dirs": [ "/Users/xxxxx/work/IdeaPj/source/kafka/kafka-logs-6" ] } ] } 然后执行脚本： sh bin/kafka-reassign-partitions.sh --zookeeper xxxxx --reassignment-json-file config/reassignment-json-file.json --execute --bootstrap-serverxxxxx:9092 --replica-alter-log-dirs-throttle 10000 注意 --bootstrap-server 在跨路径迁移的情况下，必须传入此参数。 如果需要限流的话 加上参数 --replica-alter-log-dirs-throttle ; 跟 --throttle 不一样的是 --replica-alter-log-dirs-throttle 限制的是 Broker 内不同路径的迁移流量。
两种迁移场景，分别是：同集群数据迁移、跨集群数据迁移。 一、同集群迁移 同集群之间数据迁移，比如在已有的集群中新增了一个 Broker 节点，此时需要将原来集群中已有的 Topic 的数据迁移部分到新的集群中，缓解集群压力。 将新的节点添加到 Kafka 集群很简单，只需为它们分配一个唯一的 Broker ID，并在新服务器上启动 Kafka。但是，这些新服务器节点不会自动分配任何数据分区，因此除非将分区移动到新增的节点，否则在创建新 Topic 之前新节点不会执行任何操作。因此，通常在将新服务器节点添加到 Kafka 集群时，需要将一些现有数据迁移到这些新的节点。 迁移数据的过程是手动启动的，执行过程是完全自动化的。在 Kafka 后台服务中，Kafka 将添加新服务器作为其正在迁移的分区的 Follower，并允许新增节点完全复制该分区中的现有数据。当新服务器节点完全复制此分区的内容并加入同步副本（ISR）时，其中一个现有副本将删除其分区的数据。 Kafka 系统提供了一个分区重新分配工具 kafka-reassign-partitions.sh，该工具可用于在 Broker 之间迁移分区。理想情况下，将确保所有 Broker 的数据和分区均匀分配。分区重新分配工具无法自动分析 Kafka 群集中的数据分布并迁移分区以实现均匀的负载均衡。因此，管理员在操作的时候，必须弄清楚应该迁移哪些 Topic 或分区。 分区重新分配工具可以在 3 种互斥模式下运行： --generate：在此模式下，给定 Topic 列表和 Broker 列表，该工具会生成候选重新​​分配，以将指定 Topic 的所有分区迁移到新 Broker 中。此选项仅提供了一种方便的方法，可在给定 Topic 和目标 Broker 列表的情况下生成分区重新分配计划。 --execute：在此模式下，该工具将根据用户提供的重新分配计划启动分区的重新分配。 （使用 --reassignment-json-file 选项）。由管理员手动制定自定义重新分配计划，也可以使用 --generate 选项提供。 --verify：在此模式下，该工具将验证最后一次 --execute 期间列出的所有分区的重新分配状态。状态可以有成功、失败或正在进行等状态。 1.1、迁移过程实现 分区重新分配工具可用于将一些 Topic 从当前的 Broker 节点中迁移到新添加的 Broker 中。这在扩展现有集群时通常很有用，因为将整个 Topic 移动到新的 Broker 变得更容易，而不是一次移动一个分区。当执行此操作时，用户需要提供已有的 Broker 节点的 Topic 列表，以及到新节点的 Broker 列表（源 Broker 到新 Broker 的映射关系）。然后，该工具在新的 Broker 中均匀分配给指定 Topic 列表的所有分区。在迁移过程中，Topic 的复制因子保持不变。 现有如下实例，将 Topic 为 ke01，ke02 的所有分区从 Broker1 中移动到新增的 Broker2 和 Broker3 中。由于该工具接受 Topic 的输入列表作为 JSON 文件，因此需要明确迁移的 Topic 并创建 json 文件，如下所示： > cat topic-to-move.json {"topics": [{"topic": "ke01"}, {"topic": "ke02"}], "version":1 } 准备好 JSON 文件，然后使用分区重新分配工具生成候选分配，命令如下： > bin/kafka-reassign-partitions.sh --zookeeper dn1:2181 --topics-to-move-json-file topics-to-move.json --broker-list "1,2" --generate 执行命名之前，Topic（ke01、ke02）的分区如下图所示： 执行完成命令之后，控制台出现如下信息： 该工具生成一个候选分配，将所有分区从 Topic ke01，ke02 移动到 Broker1 和 Broker2。需求注意的是，此时分区移动尚未开始，它只是告诉你当前的分配和建议。保存当前分配，以防你想要回滚它。新的赋值应保存在 JSON 文件（例如 expand-cluster-reassignment.json）中，以使用 --execute 选项执行。JSON 文件如下： {"version":1,"partitions":[{"topic":"ke02","partition":0,"replicas":[2]},{"topic":"ke02","partition":1,"replicas":[1]},{"topic":"ke02","partition":2,"replicas":[2]},{"topic":"ke01","partition":0,"replicas":[2]},{"topic":"ke01","partition":1,"replicas":[1]},{"topic":"ke01","partition":2,"replicas":[2]}]} 执行命令如下所示： > ./kafka-reassign-partitions.sh --zookeeper dn1:2181 --reassignment-json-file expand-cluster-reassignment.json --execute 最后，--verify 选项可与该工具一起使用，以检查分区重新分配的状态。需要注意的是，相同的 expand-cluster-reassignment.json（与 --execute 选项一起使用）应与 --verify 选项一起使用，执行命令如下： > ./kafka-reassign-partitions.sh --zookeeper dn1:2181 --reassignment-json-file expand-cluster-reassignment.json --verify 执行结果如下图所示： 同时，可以通过 Kafka Eagle 工具来查看 Topic 的分区情况。 二、跨集群迁移 这里跨集群迁移，指的是在 Kafka 多个集群之间复制数据 “镜像” 的过程，以避免与单个集群中的节点之间发生的复制混淆。 Kafka 附带了一个用于在 Kafka 集群之间镜像数据的工具。该工具从源集群使用并生成到目标集群。这种镜像的一个常见用例是在另一个数据中心提供副本。 另外，你可以运行许多此类镜像进程以提高吞吐量和容错（如果一个进程终止，其他进程将占用额外负载）。将从源集群中的 Topic 读取数据，并将其写入目标集群中具有相同名称的主题。事实上，“镜像” 数据只不过是一个 Kafka 将消费者和生产者联系在了一起。 源集群和目标集群是完全独立的实体，它们可以具有不同数量的分区，并且偏移量将不相同。出于这个原因，镜像集群并不是真正意图作为容错机制（因为消费者的位置会有所不同）; 为此，建议使用正常的集群内复制。但是，镜像进程将保留并使用消息 Key 进行分区，因此可以按 Key 保留顺序。 下面是一个跨集群的单 Topic 实例，命令如下： ./kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist ke03 需要注意的是，consumer.properties 文件配置源 Kafka 集群 Broker 地址，producer.properties 文件配置目标 Kafka 集群地址。如果需要迁移多个 Topic，可以使用 --whitelist 'A|B'，如果需要迁移所有的 Topic，可以使用 --whitelist '*'。 三、结果预览 执行跨集群迁移命令后，目标集群中使用 Kafka Eagle 中查看 Topic Size 大小看是否与源集群的 Topic Size 大小相等，或者使用 SQL 语句，验证是否有数据迁移过来，结果如下图所示： select * from "ke03" where "partition" in (0) 跨集群迁移数据的本质是，Kafka 启动了消费者读取源集群数据，并将消费后的数据写入到目标集群，在迁移的过程中，可以启动多个实例，提供迁出的吞吐量。
一、Security 在 Kafka 的官方文档中介绍了如下几种安全方式，以满足用户不同场景对安全的需求。 SASL/GSSAPI (Kerberos) - starting at version 0.9.0.0。主要是为 Kerberos 使用，如果当前已有 Kerberos 认证，只需要为集群中每个 Broker 和访问用户申请 Principle ，然后在 Kafka 配置文件中开启 Kerberos 的支持即可。官方文档详见 Authentication using SASL/Kerberos。 SASL/PLAIN - starting at version 0.10.0.0。一个简单的用户名和密码身份认证机制，通常与 TLS/SSL 一起用于加密，以实现身份验证。是一种比较容易使用的方式，但是也有一个很明显的缺点，这种方式会把用户账户文件配置到静态文件中，每次想要添加新的账户都需要重启 Kafka 去加载静态文件，才能生效，十分不方便。官方文档详见 Authentication using SASL/PLAIN。 SASL/SCRAM-SHA-256 和 SASL/SCRAM-SHA-512 - starting at version 0.10.2.0。通过将认证信息保存在 ZooKeeper 里面，从而动态的获取用户信息，相当于把 ZK 用作一个认证中心使用。这种认证可以在使用过程中，使用 Kafka 提供的命令动态地创建和删除用户，无需重启整个集群，十分方便。官方文档详见 Authentication using SASL/SCRAM 。 SASL/OAUTHBEARER - starting at version 2.0。Kafka 引入的新认证机制，主要是为了实现与 OAuth2 框架的集成，Kafka 不提倡单纯使用 OAUTHBEARER，因为它生成的不安全 Json Web Token，必须配以 SSL 加密才能在生产环境中使用。官方文档详见 Authentication using SASL/OAUTHBEARER 。 二、SASL+ACL 实现用户及权限认证 2.1、下载 wget https://repo.huaweicloud.com/apache/kafka/3.3.1/kafka_2.12-3.3.1.tgz -P /data cd /data tar -zxf kafka_2.12-3.3.1.tgz ln -s kafka_2.12-3.3.1 kafka cd /data/kafka 2.2、Kafka 服务配置 Kafka server 的 jaas 文件如下： /data/kafka/config/jaas/kafka_server_jaas.conf 这里创建五个用户，密码分别如下，后会会对其进行分别授权。 用户 密码 admin admin123 kwriter kwriter123 kreader kreader123 sharga sharga123 shargb shargb123 创建文件： mkdir /data/kafka/config/jaas/ vim /data/kafka/config/jaas/kafka_server_jaas.conf 配置如下： KafkaServer { org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin123" user_admin="admin123" user_kwriter="kwriter123" user_kreader="kreader123" user_sharga="sharga123" user_shargb="shargb123"; }; 2.3、修改 Kafka 服务启动脚本 1、复制一份作为带安全的启动脚本： cp /data/kafka/bin/kafka-server-start.sh/data/kafka/bin/sasl_kafka-server-start.sh 2、修改大概 44 行为如下，引入上面的 jaas.conf 文件，修改为如下 export LOG_DIR=/var/log/kafka-logs export KAFKA_OPTS="-Djava.security.auth.login.config=/data/kafka/config/jaas/kafka_server_jaas.conf" exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@" 2.4、配置 server.properties 拷贝一份 ： cp /data/kafka/config/server.properties/data/kafka/config/jaas/server.properties 配置如下： broker.id=0 listeners=SASL_PLAINTEXT://192.168.179.3:9092 advertised.listeners=SASL_PLAINTEXT://192.168.179.3:9092 security.inter.broker.protocol=SASL_PLAINTEXT sasl.mechanism.inter.broker.protocol=PLAIN sasl.enabled.mechanisms=PLAIN # ACL 入口类 authorizer.class.name=kafka.security.authorizer.AclAuthorizer # 设置 admin 用户为超级管理员 super.users=User:admin auto.create.topics.enable=true log.dirs=/data/kafka/data/kafka-logs zookeeper.connect=192.168.179.3:2181/kafka 创建需要的目录： mkdir -p /data/kafka/data/{kafka-logs,zookeeper} 2.5、启动 Zookeeper ZK 服务地址可以使用 Kafka 自带的，也可以使用已部署在的 ZK。如果已有 ZK 可以跳过这一步。这里以 Kafka 自带的为例。 配置 zookeeper.properties： dataDir=/data/kafka/data/zookeeper clientPort=2181 maxClientCnxns=0 admin.enableServer=false 启动 ZK： /data/kafka/bin/zookeeper-server-start.sh -daemon /data/kafka/config/zookeeper.properties 查看日志： tail -f /var/log/zookeeper/server.log 可以通过如下命令进一步查看 znode 信息： /data/kafka/bin/zookeeper-shell.sh 192.168.179.3:2181 2.6、启动 Kafka 集群 # 1 启动 Kafka /data/kafka/bin/sasl_kafka-server-start.sh -daemon /data/kafka/config/jaas/server.properties # 2 查看启动日志 tail -f /var/log/kafka-logs/server.log 关闭的命令如下： /data/kafka/bin/kafka-server-stop.sh /data/kafka/bin/zookeeper-server-stop.sh 如果需要调整 Kafka JVM 参数，可以如下进行修改，调整期堆大小，或者其它参数： vim /data/kafka/bin/sasl_kafka-server-start.sh +28 2.7、ACL 接下来对用户进行授权，大概按照如下进行设置。 用户 密码 权限 说明 admin admin123 All 超级管理员用户 kwriter kwriter123 producer 生产者类权限，Writer、Create、Describe kreader kreader123 consumer 消费者类权限，Read、Describe sharga sharga123 All 仅对 sharga-* 类 Topic 有所有权 shargb shargb123 All 仅对 shargb-* 类 Topic 有所有权 命令： # 1 授予 kwriter 用户为生产者权限 /data/kafka/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.179.3:2181/kafka \ --add --allow-principal User:kwriter --producer --topic "*" # 2 授予 kreader 用户为消费者权限 /data/kafka/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.179.3:2181/kafka \ --add --allow-principal User:kreader --consumer --topic "*" --group "*" # 3 对 sharga 用户进行授权，前缀模式 /data/kafka/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.179.3:2181/kafka \ --add --allow-principal User:sharga --group "*" --topic "sharga" --resource-pattern-type prefixed # 4 对 shargb 用户进行授权，前缀模式 /data/kafka/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.179.3:2181/kafka \ --add --allow-principal User:shargb --group "*" --topic "shargb" --resource-pattern-type prefixed # 5 查看 ACL 权限列表 /data/kafka/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.179.3:2181/kafka --list 小的测试验证（在没有使用正确用户时，执行如下会失败）： # 1 创建 Topic（客户端没有使用认证是，创建 topic 会失败） /data/kafka/bin/kafka-topics.sh --create --bootstrap-server node01:9092 \ --partitions 1 --replication-factor 1 --topic test_sasl # 2 查看 topic（客户端没有使用认证是，获取 topic 会失败） /data/kafka/bin/kafka-topics.sh --list --bootstrap-server node01:9092 | grep test # 3 查看 topic 详情（客户端没有使用认证是，获取 topic 详情会失败） /data/kafka/bin/kafka-topics.sh --bootstrap-server 192.168.179.3:9092 \ --describe --topic test_sasl # 4 生产者（此时会因为权限问题，而报错，下面来解决这个问题） /data/kafka/bin/kafka-console-producer.sh --bootstrap-server 192.168.179.3:9092 \ --topic test_sasl # 5 消费者（此时会因为权限问题，而报错，下面来解决这个问题） /data/kafka/bin/kafka-console-consumer.sh --bootstrap-server node01:9092 \ --from-beginning --topic test_sasl 2.7.1、admin 新增 admin 配置属性文件： vim /data/kafka/config/jaas/admin.properties 内容如下： sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin123"; security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN 以 admin 身份执行如下命令： # 1 再次创建 topic，此时可以成功创建 /data/kafka/bin/kafka-topics.sh --create --bootstrap-server node01:9092 \ --command-config /data/kafka/config/jaas/admin.properties \ --partitions 1 --replication-factor 1 --topic test_sasl # 2 list /data/kafka/bin/kafka-topics.sh --list --bootstrap-server node01:9092 \ --command-config /data/kafka/config/jaas/admin.properties # 3 topic 详新信息 /data/kafka/bin/kafka-topics.sh --bootstrap-server node01:9092 \ --command-config /data/kafka/config/jaas/admin.properties \ --describe --topic test_sasl 如果需要删除 Topic，可使用如下命令 /data/kafka/bin/kafka-topics.sh --bootstrap-server node01:9092 \ --command-config /data/kafka/config/jaas/admin.properties \ --delete --topic test_sasl 其它（可以执行完下面生产者或消费者命令后再执行查看） # 1 查看当前的消费者组 /data/kafka/bin/kafka-consumer-groups.sh --bootstrap-server node01:9092 --command-config \ /data/kafka/config/jaas/admin.properties --list # 2 查看消费者消费进度 /data/kafka/bin/kafka-consumer-groups.sh --bootstrap-server node01:9092 --command-config \ /data/kafka/config/jaas/admin.properties --describe --group test-consumer-group # 3 查看 Kafka Topic 的 offset 信息 /data/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server node01:9092 \ --command-config /data/kafka/config/jaas/admin.properties --time -1 --topic test_sasl 2.7.2、生产者 新增生产者 kwrite 用户的配置属性文件： vim /data/kafka/config/jaas/kwriter.properties 内容如下： sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kwriter" password="kwriter123"; security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN 使用客户端命令行测试： /data/kafka/bin/kafka-console-producer.sh --bootstrap-server node01:9092 \ --producer.config/data/kafka/config/jaas/kwriter.properties --topic test_sasl 发送如下的数据： {"id":100,"name":"one","age":1} {"id":101,"name":"two","age":2} {"id":102,"name":"three","age":3} 2.7.3、消费者 新增生产者 kreader 用户的配置属性文件： vim /data/kafka/config/jaas/kreader.properties 内容如下： group.id=test-consumer-group sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kreader" password="kreader123"; security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN 消费者测试： /data/kafka/bin/kafka-console-consumer.sh --bootstrap-server node01:9092 \ --consumer.config/data/kafka/config/jaas/kreader.properties --from-beginning --topic test_sasl 2.7.4、sharga 用户 新增生产者 sharga 用户的配置属性文件： vim /data/kafka/config/jaas/sharga.properties 内容如下： group.id=test-sharga-group sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="sharga" password="sharga123"; security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN 执行如下命令： # 1 创建自己的 topic ## sharga-t01 /data/kafka/bin/kafka-topics.sh --create --bootstrap-server node01:9092 --command-config \ /data/kafka/config/jaas/sharga.properties --partitions 1 --replication-factor 1 --topic sharga-t01 # 2 生产者 /data/kafka/bin/kafka-console-producer.sh --bootstrap-server node01:9092 --producer.config \ /data/kafka/config/jaas/sharga.properties --topic sharga-t01 向 sharga-t01 的 Topic 发送如下消息： {"id":100,"name":"one","age":1} {"id":101,"name":"two","age":2} {"id":102,"name":"three","age":3} #3 同样创建 sharga-t02 Topic /data/kafka/bin/kafka-topics.sh --create --bootstrap-server node01:9092 --command-config \ /data/kafka/config/jaas/sharga.properties --partitions 1 --replication-factor 1 --topic sharga-t02 # 4 生产者 /data/kafka/bin/kafka-console-producer.sh --bootstrap-server node01:9092 --producer.config \ /data/kafka/config/jaas/sharga.properties --topic sharga-t02 向 sharga-t02 的 Topic 发送如下消息： {"id":100,"sex":"man"} {"id":101,"sex":"man"} {"id":102,"sex":"woman"} # 5 查看当前的 topic 列表 /data/kafka/bin/kafka-topics.sh --list --bootstrap-server node01:9092 --command-config \ /data/kafka/config/jaas/sharga.properties sharga 用户仅能查看到自己的 Topic。 2.7.6、说明 这种方式是将用户和密码相关的信息保存在了服务器本地文件中了，经过前面的配置，基本涉及到安全认证的配置都放置到了 /data/kafka/config/jaas 目录下，因此如果需要在真实服务上使用时，这个目录需要经读写权限详细的分配给系统的每个用户，对于不允许使用此用户配置信息的其他用户，这个文件的读写权限可以不授予给这个用户。 2.8、生产者客户端代码设置 Properties props = new Properties (); //... props.put ("security.protocol", "SASL_PLAINTEXT"); props.put ("sasl.mechanism", "PLAIN"); props.put ("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='kwriter' password='kwriter123';"); 2.9、消费者客户端代码设置 Properties props = new Properties (); //... props.put ("security.protocol", "SASL_PLAINTEXT"); props.put ("sasl.mechanism", "PLAIN"); props.put ("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='kreader' password='kreader123';"); 2.10、Spring-kafka 需配置如下 spring: #…… kafka: bootstrap-servers: node01:9092 #指定 kafka server 的地址，集群配多个，中间，逗号隔开 producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer properties: # sasl 安全认证 security.protocol: SASL_PLAINTEXT sasl.mechanism: PLAIN sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username='kwriter' password='kwriter123'; consumer: group-id: app_consumer_group #群组 ID enable-auto-commit: true auto-commit-interval: 1000s key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer properties: security.protocol: SASL_PLAINTEXT sasl.mechanism: PLAIN sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username='kreader' password='kreader123'; 2.11、Flink SQL create table kafka_sharga_t01 ( id int, name string, age int ) with ( 'format' = 'json', 'properties.bootstrap.servers' = '192.168.179.3:9092', 'connector' = 'kafka', 'topic' = 'sharga-t01', 'properties.group.id' = 'group1', 'scan.startup.mode' = 'earliest-offset', 'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.plain.PlainLoginModule required username="sharga" password="sharga123";', 'properties.security.protocol' = 'SASL_PLAINTEXT', 'properties.sasl.mechanism' = 'PLAIN' );
一、Topic Command 1.0、Kafka 启停 启动 /opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties 停止 /opt/module/kafka/bin/kafka-server-stop.sh 1.1、Topic 创建 ​​bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3 --topic test​​ 1.2、删除 Topic ​bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic test​​ 1.3、Topic 分区扩容 单个 Topic 扩容 ​bin/kafka-topics.sh --bootstrap-server broker_host:port --alter --topic test_create_topic1 --partitions 4​​ 批量扩容 (将所有正则表达式匹配到的 Topic 分区扩容到 4 个) ​​sh bin/kafka-topics.sh --topic ".*?" --bootstrap-server localhost:9092 --alter --partitions 4​​ 1.4、查询 Topic 描述 1、查询单个 Topic bin/kafka-topics.sh --topic test --bootstrap-server localhost:9092 --describe --exclude-internal​​ 2、批量查询 Topic (正则表达式匹配，下面是查询所有 Topic) ​​sh bin/kafka-topics.sh --topic ".*?" --bootstrap-server localhost:9092 --describe --exclude-internal​​ 支持正则表达式匹配 Topic, 只需要将 topic 用双引号包裹起来。 1.5、查询 Topic 列表 1、查询所有 Topic 列表 bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --list --exclude-internal​​ 2、查询匹配 Topic 列表 (正则表达式) 查询​​test_create_​​开头的所有 Topic 列表。 bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --list --exclude-internal --topic "user.*" 二、Config Command 2.1、查询配置 1、查询单个 Topic 配置 (只列举动态配置) ​sh bin/kafka-configs.sh --describe --bootstrap-server 127.0.0.1:9092 --topic test​​ 或者 ​​sh bin/kafka-configs.sh --describe --bootstrap-server 127.0.0.1:9092 --entity-type topics --entity-name test​​ 2、查询所有 Topic 配置 (包括内部 Topic)(只列举动态配置) ​​sh bin/kafka-configs.sh --describe --bootstrap-server 127.0.0.1.85:9092 --entity-type topics​​ 3、查询 Topic 的详细配置 (动态 + 静态) 只需要加上一个参数​​--all​​ 其他配置 /clients/users/brokers/broker-loggers 的查询 同理 ；只需要将​​--entity-type​​ 改成对应的类型就行了 (topics/clients/users/brokers/broker-loggers) 查询 kafka 版本信息 bin/kafka-configs.sh --describe --bootstrap-server 127.0.0.1:9092 --version​​ 2.2、增删改配置 ​​--alter​​ 删除配置: ​​--delete-config​​ k1=v1,k2=v2 添加 / 修改配置: ​​--add-config​​ k1,k2 选择类型: ​​--entity-type​​ (topics/clients/users/brokers/broker-loggers) 类型名称: ​​--entity-name​​ Topic 添加 / 修改动态配置​​ --add-config​​ bin/kafka-configs.sh --bootstrap-server 127.0.0.1:9092 --alter --entity-type topics --entity-name test_create_topic1 --add-config file.delete.delay.ms=222222,retention.ms=999999​​ Topic 删除动态配置 --delete-config​​ bin/kafka-configs.sh --bootstrap-server 127.0.0.1:9092 --alter --entity-type topics --entity-name test_create_topic1 --delete-config file.delete.delay.ms,retention.ms​​ 添加 / 删除配置同时执行 bin/kafka-configs.sh --bootstrap-server 127.0.0.1:9092 --alter --entity-type brokers --entity-default --add-config log.segment.bytes=788888888 --delete-config log.retention.ms​​ 其他配置同理，只需要类型改下​​--entity-type​​ 类型有: (topics/clients/users/brokers/broker- loggers) 默认配置 配置默认​​--entity-default​​ sh bin/kafka-configs.sh --bootstrap-server 127.0.0.1:9092 --alter --entity-type brokers --entity-default --add-config log.segment.bytes=88888888​​ 动态配置的默认配置是使用了节点 ​​​​; 优先级：指定动态配置 > 默认动态配置 > 静态配置 三、副本扩缩、分区迁移、跨路径迁移 kafka-reassign-partitions 【kafka 运维】副本扩缩容、数据迁移、副本重分配、副本跨路径迁移​​ 【​​kafka 实战】分区重分配可能出现的问题和排查问题思路 四、Topic 的发送 kafka-console-producer.sh 4.1、生产无 key 消息 ## 生产者 bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test --producer.config config/producer.properties 4.2、生产有 key 消息 加上属性​​--property parse.key=true​​ ## 生产者 bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test --producer.config config/producer.properties --property parse.key=true 默认消息 key 与消息 value 间使用 “Tab 键” 进行分隔，所以消息 key 以及 value 中切勿使用转义字符 (\t)。 五、Topic 的消费 kafka-console-consumer.sh 5.1、新客户端从头消费​​--from-beginning​​ (注意这里是新客户端，如果之前已经消费过了是不会从头消费的) 下面没有指定客户端名称，所以每次执行都是新客户端都会从头消费： sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning 5.2、正则表达式匹配 topic 进行消费​​--whitelist​​ ​​消费所有的 topic​​ sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --whitelist ‘.*’ ​​消费所有的 topic，并且还从头消费​​ sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --whitelist ‘.*’ --from-beginning 5.3、显示 key 进行消费​​--property print.key=true​​ sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --property print.key=true 5.4、指定分区消费​​--partition​​ 指定起始偏移量消费​​--offset​​ sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --partition 0 --offset 100 5.5、给客户端命名​​--group​​ 注意给客户端命名之后，如果之前有过消费，那么​​--from-beginning​​就不会再从头消费了 sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --group test-group 5.6、添加客户端属性​​--consumer-property​​ 这个参数也可以给客户端添加属性，但是注意 不能多个地方配置同一个属性，他们是互斥的；比如在下面的基础上还加上属性​​--group test-group​​ 那肯定不行 sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test ​​--consumer-property group.id=test-consumer-group​​ 5.7、添加客户端属性​​--consumer.config​​ 跟​​--consumer-property​​​ 一样的性质，都是添加客户端的属性，不过这里是指定一个文件，把属性写在文件里面，​​--consumer-property​​​ 的优先级大于 ​​--consumer.config​​ sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --consumer.config config/consumer.properties 六、kafka-leader-election Leader 重新选举 6.1、指定 Topic 指定分区用重新​​PREFERRED：优先副本策略​​ 进行 Leader 重选举 sh bin/kafka-leader-election.sh --bootstrap-server 127.0.0.1:9092 --topic test_create_topic4 --election-type PREFERRED --partition 0 6.2、所有 Topic 所有分区用重新​​PREFERRED：优先副本策略​​ 进行 Leader 重选举 sh bin/kafka-leader-election.sh --bootstrap-server 127.0.0.1:9092 --election-type preferred --all-topic-partitions 6.3、设置配置文件批量指定 topic 和分区进行 Leader 重选举 先配置 leader-election.json 文件 { "partitions": [ { "topic": "test_create_topic4", "partition": 1 }, { "topic": "test_create_topic4", "partition": 2 } ] } sh bin/kafka-leader-election.sh --bootstrap-server xxx:9090 --election-type preferred --path-to-json-file config/leader-election.json 七、持续批量推送消息 kafka-verifiable-producer.sh 单次发送 100 条消息​​--max-messages 100​​ 一共要推送多少条，默认为 - 1，-1 表示一直推送到进程关闭位置 sh bin/kafka-verifiable-producer.sh --topic test_create_topic4 --bootstrap-server localhost:9092 ​​--max-messages 100​​ 每秒发送最大吞吐量不超过消息 ​​--throughput 100​​ 推送消息时的吞吐量，单位 messages/sec。默认为 - 1，表示没有限制 sh bin/kafka-verifiable-producer.sh --topic test_create_topic4 --bootstrap-server localhost:9092 ​​--throughput 100​​ 发送的消息体带前缀​​--value-prefix​​ sh bin/kafka-verifiable-producer.sh --topic test_create_topic4 --bootstrap-server localhost:9092 ​​--value-prefix 666​​ 注意​​--value-prefix 666​​​必须是整数，发送的消息体的格式是加上一个 点号​​.​​​ 例如： ​​666.​​ 其他参数： ​​--producer.config CONFIG_FILE​​ 指定 producer 的配置文件 ​​--acks ACKS​​ 每次推送消息的 ack 值，默认是 - 1 八、持续批量拉取消息 kafka-verifiable-consumer 持续消费 sh bin/kafka-verifiable-consumer.sh --group-id test_consumer --bootstrap-server localhost:9092 --topic test_create_topic4 单次最大消费 10 条消息​​--max-messages 10​​ sh bin/kafka-verifiable-consumer.sh --group-id test_consumer --bootstrap-server localhost:9092 --topic test_create_topic4 ​​--max-messages 10​​ 九、生产者压力测试 kafka-producer-perf-test.sh 9.1、发送 1024 条消息​​--num-records 100​​并且每条消息大小为 1KB​​--record-size 1024​​ 最大吞吐量每秒 10000 条​​--throughput 100​​ sh bin/kafka-producer-perf-test.sh --topic test_create_topic4 --num-records 100 --throughput 100000 --producer-props bootstrap.servers=localhost:9092 --record-size 1024 9.2、用指定消息文件​​--payload-file​​发送 100 条消息最大吞吐量每秒 100 条​​--throughput 100​​ 先配置好消息文件​​batchmessage.txt​​ 然后执行命令 发送的消息会从​​batchmessage.txt​​里面随机选择；注意这里我们没有用参数​​--payload-delimeter​​指定分隔符，默认分隔符是 \n 换行； bin/kafka-producer-perf-test.sh --topic test_create_topic4 --num-records 1024 --throughput 100 --producer-props bootstrap.servers=localhost:9090 --payload-file config/batchmessage.txt 十、消费者压力测试 kafka-consumer-perf-test.sh 消费 100 条消息​​--messages 100​​ sh bin/kafka-consumer-perf-test.sh -topic test_create_topic4 --bootstrap-server localhost:9090 --messages 100 十一、删除指定分区的消息 kafka-delete-records.sh 删除指定 topic 的某个分区的消息删除至 offset 为 1024 先配置 json 文件​​offset-json-file.json​​ {"partitions": [{"topic": "test1", "partition": 0, "offset": 1024}], "version":1 } 在执行命令 sh bin/kafka-delete-records.sh --bootstrap-server 172.23.250.249:9090 --offset-json-file config/offset-json-file.json 十二、查看 Broker 磁盘信息 kafka-log-dirs.sh 查询指定 topic 磁盘信息​​--topic-list topic1,topic2​​ sh bin/kafka-log-dirs.sh --bootstrap-server 127.0.0.1:9092 --describe --topic-list test2 查询指定 Broker 磁盘信息​​--broker-list 0 broker1,broker2​​ sh bin/kafka-log-dirs.sh --bootstrap-server 127.0.0.1:9092 --describe --topic-list test2 --broker-list 0 例如我一个 3 分区 3 副本的 Topic 的查出来的信息 ​​logDir​​ Broker 中配置的​​log.dir​​ { "version": 1, "brokers": [{ "broker": 0, "logDirs": [{ "logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-0", "error": null, "partitions": [{ "partition": "test2-1", "size": 0, "offsetLag": 0, "isFuture": false }, { "partition": "test2-0", "size": 0, "offsetLag": 0, "isFuture": false }, { "partition": "test2-2", "size": 0, "offsetLag": 0, "isFuture": false }] }] }, { "broker": 1, "logDirs": [{ "logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-1", "error": null, "partitions": [{ "partition": "test2-1", "size": 0, "offsetLag": 0, "isFuture": false }, { "partition": "test2-0", "size": 0, "offsetLag": 0, "isFuture": false }, { "partition": "test2-2", "size": 0, "offsetLag": 0, "isFuture": false }] }] }, { "broker": 2, "logDirs": [{ "logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-2", "error": null, "partitions": [{ "partition": "test2-1", "size": 0, "offsetLag": 0, "isFuture": false }, { "partition": "test2-0", "size": 0, "offsetLag": 0, "isFuture": false }, { "partition": "test2-2", "size": 0, "offsetLag": 0, "isFuture": false }] }] }, { "broker": 3, "logDirs": [{ "logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-3", "error": null, "partitions": [] }] }] } 十三、消费者组管理 kafka-consumer-groups.sh 13.1、查看消费者列表​​--list​​ ​sh bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --list​​ 先调用​​MetadataRequest​​拿到所有在线 Broker 列表 再给每个 Broker 发送​​ListGroupsRequest​​请求获取 消费者组数据 13.2、查看消费者组详情​​--describe​​ ​​DescribeGroupsRequest​​ 查看消费组详情​​--group​​ 或 ​​--all-groups​​ 查看指定消费组详情​​--group​​​ ​​sh bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group test2_consumer_group​​ 查看所有消费组详情​​--all-groups​​​ ​​sh bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --all-groups 查看该消费组 消费的所有 Topic、及所在分区、最新消费 offset、Log 最新数据 offset、Lag 还未消费数量、消费者 ID 等等信息。 查询消费者成员信息​​--members​​ ​​sh bin/kafka-consumer-groups.sh --describe --all-groups --members --bootstrap-server xxx:9090​ 指定消费组成员信息 ​sh bin/kafka-consumer-groups.sh --describe --members --group test2_consumer_group --bootstrap-server 127.0.0.1:9092​​ 查询消费者状态信息​​--state​​ ​​sh bin/kafka-consumer-groups.sh --describe --all-groups --state --bootstrap-server 127.0.0.1:9092 指定消费组状态信息 ​sh bin/kafka-consumer-groups.sh --describe --state --group test2_consumer_group --bootstrap-server 127.0.0.1:9092​​ 13.3、删除消费者组​​--delete​​ ​​DeleteGroupsRequest​​ 删除消费组–delete 删除指定消费组​​--group​​​ ​sh bin/kafka-consumer-groups.sh --delete --group test2_consumer_group --bootstrap-server 127.0.0.1:9092​​ 删除所有消费组​​--all-groups​​​ sh bin/kafka-consumer-groups.sh --delete --all-groups --bootstrap-server 127.0.0.1:9092​​ PS: 想要删除消费组前提是这个消费组的所有客户端都停止消费 / 不在线才能够成功删除；否则会报下面异常 Error: Deletion of some consumer groups failed: Group 'test2_consumer_group' could not be deleted due to: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.GroupNotEmptyException: The group is not empty. 13.4、重置消费组的偏移量 ​​--reset-offsets​​ 能够执行成功的一个前提是 消费组这会是不可用状态； 下面的示例使用的参数是：​​--dry-run​​ ; 这个参数表示预执行，会打印出来将要处理的结果； 等你想真正执行的时候请换成参数​​--execute​​ ; 下面示例 重置模式都是 ​​--to-earliest​​ 重置到最早的； 请根据需要参考下面 相关重置 Offset 的模式 换成其他模式； 重置指定消费组的偏移量 ​​--group​​ 重置指定消费组的所有 Topic 的偏移量​​--all-topic​​​ ​​sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --group test2_consumer_group --bootstrap-server 127.0.0.1:9092 --dry-run --all-topic ​​重置指定消费组的指定 Topic 的偏移量​​--topic​​​ ​​sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --group test2_consumer_group --bootstrap-server 127.0.0.1:9092 --dry-run --topic test2​​ 重置所有消费组的偏移量 ​​--all-group​​ 重置所有消费组的所有 Topic 的偏移量​​--all-topic​​​ ​​sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --all-group --bootstrap-server 127.0.0.1:9092 --dry-run --all-topic ​​重置所有消费组中指定 Topic 的偏移量​​--topic​​​ ​​sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --all-group --bootstrap-server 127.0.0.1:9092 --dry-run --topic test2​​ ​​--reset-offsets​​ 后面需要接重置的模式 13.5、删除偏移量​​delete-offsets​​ 能够执行成功的一个前提是 消费组这会是不可用状态； 偏移量被删除了之后，Consumer Group 下次启动的时候，会从头消费； ​sh bin/kafka-consumer-groups.sh --delete-offsets --group test2_consumer_group2 --bootstrap-server 127.0.0.1:9092 --topic test2​​ 十四、查看日志文件 kafka-dump-log.sh ​​sh bin/kafka-dump-log.sh --files kafka-logs-0/test2-0/00000000000000000300.log​​ 查询 Log 文件具体信息 ​​--print-data-log​​ ​sh bin/kafka-dump-log.sh --files kafka-logs-0/test2-0/00000000000000000300.log --print-data-log​​ 查询 index 文件具体信息 ​sh bin/kafka-dump-log.sh --files kafka-logs-0/test2-0/00000000000000000300.index​​ 配置项为​​log.index.size.max.bytes​​； 来控制创建索引的大小； 查询 timeindex 文件 ​​sh bin/kafka-dump-log.sh --files kafka-logs-0/test2-0/00000000000000000300.timeindex​​ kafka 重新消费已消费的记录： 其中，<kafka_host > 和 < port > 是 Kafka 服务器的主机名和端口号，<group_name > 是消费者组的名称，<topic_name > 是要消费的主题名称，<offset_number > 是要重置的偏移量位置。 kafka-consumer-groups.sh --bootstrap-server <kafka_host>:<port> --group <group_name> --topic <topic_name> --reset-offsets --to-offset <offset_number> --execute 将 Kafka 消费者的偏移量设置到某个时间前： 其中，<kafka_host > 和 < port > 是 Kafka 服务器的主机名和端口号，<group_name > 是消费者组的名称，<topic_name > 是要消费的主题名称， 是要设置的时间点。请注意， 的格式应为 YYYY-MM-DDTHH:mm:ss.sss，例如 2022-01-01T00:00:00.000 kafka-consumer-groups.sh --bootstrap-server <kafka_host>:<port> --group <group_name> --topic <topic_name> --reset-offsets --to-datetime <datetime> --execute 参考： https://blog.51cto.com/szzdzhp/5302196
一、MM2 简介 在介绍 MM2 之前先谈一下 MM1, 这个是 kafka 很早之前就有的组件，本质来说就是实现了 consumer + producer， 从集群 A 将数据同步到集群 B， 使用的是 kafka client 的 high level api。如果网络不稳定，会有频繁 rebalance 问题。一个 MM1 实例是解决部分 topic 从集群 A—>B 的同步问题。所以如果我们有 N 个 kafka 集群， 那么为了满足所有同步需求，理论上我们要维护 N (N-1) 个实例， 如果算上每个实例只处理部分 topic 同步的话，那么需要维护的实例会更多。 MM2 是 MM1 的升级替代品，用于 kafka 集群间的数据同步，解决了 MM1 的很多问题，同时部署维护成本大大降低，以 N 个 kafka 集群为例，只需要部署 N 个 MM2 集群即可满足需求。基于更高抽象的 Kafka Connect framework 解决容错和水平扩展问题。内部基于 low level (assign) 来实现 topic 订阅， 没有频繁 rebalance 的问题， 对同步有更精准的控制。 MM2 具有如下特性: 基于 Kafka Connect framework 和生态 自动探测新 topic, partition 自动同步 topic 配置，自动同步 topic acl 支持 active-active 集群对，以及任意数量的 active 集群 支持跨 IDC 同步， aggregation 和其他复杂拓扑 consumer offset 等 meta 信息的同步和翻译 no rebalance，减少同步波动 提供广泛的指标，例如跨多个数据中心 / 集群的端到端复制延迟 容错和水平可扩展 MM2 架构如下: MM2 基于 kafka connect framework 实现集群部署，通过 Connect Worker 来同步数据， 容错与水平扩展等由 Connect 框架保障。 MM2 主要有两种运行模式：Connector cluster 和 Driver, 我们采用的是 Driver 模式， 因为这种模式封装得更好，配置简洁，也是官方推荐的模式。 MM2 对于同步 topic 有前缀命名约定，默认情况下是 source clusterName, 当然也可以自定义，如下图所有，所有同步 topic 都包含有 source clusterName 前缀，这样做有两个好处：一是可以清楚知晓数据来源， 二是可以避免循环同步的问题。 下面介绍几个比较重要的内部 topic，可以说 MM2 的主要特性都依赖于这几个 topic: checkpoint topic: 同步集群间 consumer group state，其中记录了 consumer offset 的映射关系，在 2.7.0 + 版本后还支持自动提交到 target cluster 的__consumer_offsets， 也可通过 translateOffsets () 接口手动翻译后提交 offset_sync topic: 同步集群间 broker 端 offset 映射关系，分为 upstreamOffset 和 downstreamOffset, 其中 upstreamOffset 是上游 topic 的 offset (source 端)， downstreamOffset 是下游 topic 的 offset (target 端) heartbeat topic: 用于监控 replication flows, 可以动态发现 replication topology，便于 MM2 动态分配同步任务 关于 MM2 的部署我们采用的方案是：每个 kafka 集群 (作为 target cluster) 部署一个 MM2 集群实例，部署在 k8s 上方便运维管理， 通过 jmx_prometheus_javaagent 将 metrics 暴露成 exporter 供 prometheus server 抓取， 然后通过 grafana 展示。 二、跨 IDC kafka 热备多活 在没有跨 IDC kafka 热备方案之前， 我们只有冷备方案，就是在集群 A 出现问题的情况下，手动将业务迁移到集群 B 上， 这种操作的代价是巨大的， 而且短时间内无法恢复，离线业务可能还好，实时业务是完全无法接受的。 下面列举了在冷备的情况下业务迁移面临的问题: 业务上下游的依赖问题，因为涉及到生产和消费的先后问题，各业务必须协调好顺序 沟通上下游各个业务做迁移，沟通成本很高 数据安全得不到保障，在生产端没有及时切换的情况下可能会丢数据，下游 pipeline 是否能处理完整数据也要打个问号 数据恢复时间过长，对于实时业务是无法容忍的 所以我们想是不是可以做到跨 IDC 的热备多活， 在一个 IDC 集群出问题的情况下， 业务可以快速切换恢复。 在没有 MM2 之前， 要实现两个集群的实时同步，特别是 consumer group state 的实时同步是比较困难的， 一个是日志的实时同步稳定性，之前已经说过 MM1 的稳定性和可维护性是比较差的，还有就是 offset 信息 (分为 broker offset, consumer offset) 两个集群是完全独立的， 需要一个 offset mapping 实时映射。之前 uber 自己实现过一个多 kafka 集群热备的方案，基于 uReplication 和自研组件，但是实现上有点复杂，需要维护的组件也较多。 在 MM2 发布之后，我们发现已经比较完美地解决了以上问题， 所以我们觉得基于 M2 实现热备是一个很好的方案， 当然在实现的过程中也遇到了很多问题， 下面也会逐一讲解。 三、设计目标 跨 IDC 容灾，在一个 IDC 不可用的情况下，业务可快速恢复 业务上下游 pipeline 可以解耦，各个业务切换完全独立，不需要协调上下游 切换代价尽量小，最好能不重启业务情况下切换集群 产品流程使用上尽量简单，将复杂度对业务屏蔽掉 这个是我们要实现的一个总体目标，主要是为了解决冷备情况下的痛点问题， 同时我们想通过产品化流程化将这个热备切换的过程对业务使用尽量简单。 四、设计方案 为了实现 Kafka 集群跨 IDC 容灾，首先要考虑的是如何同步，做单向同步还是双向同步，我们的需求是两个集群的 topic 数据是一致，且要上下游模块完全解耦， 显然单向同步不能满足我们的需求，因为对于在集群 A 的 producer 写入 topicT, 在 B 集群也必须要有 topicT, 否则 producer 就是无法迁移的， 所以我们只能使用双向同步的方案：即在集群 A, 集群 B 都有 topicT，然后双向同步。双向同步后无论 producer 写入的是哪个集群， 两个集群的 topic 数据是一致的。 下图以 IDC1 与 IDC2 的两个 Kafka 集群 Cluster A 与 Cluster B 之间双向同步 topicT 为例，consumer 消费 A 集群，其 offset 信息会通过 checkpoint topic 被实时同步给 B 集群， 这样当 A 集群不可用时就可以切换到 B 集群的 consumer 继续消费。对于 producer 的切换更加简单，只要切换写入 B 集群 topicT 即可，两个集群数据是一致的。 以下是我们热备多活方案简单示意图， producer 只要写成功一个集群即为成功， consumer 因为 group state 实时同步，所以 standby 的 consumer 随时可以切换消费。 3.1、MM2 支持动态修改同步配置 目前 MM2 不支持动态修改同步配置，但业务有对 topic 动态增减的需求，如果每次修改都需要重启 MM2 来实现的话，那么代价太大了。我们发现可以通过实现 TopicFilter，GroupFilter 接口来解决这个问题。我们使用 mysql 来存储 MM2 的同步 meta 信息， 那么在接口中动态查询获取 mysql 的信息即可满足动态加载同步信息的需求。重载 configure 方法，定期 update 数据库中同步配置信息，以下是代码片段： 3.2、MM2 不支持反向同步 consumer group state 如上所示，对于 B.topicT 到 topicT 的 offset mapping 是不支持的，也就是 mm2 的 checkpoint 同步是单向的。比如 topicT 从 cluster A 同步到 cluster B, consumer c 在 cluster A 消费 topicT. cluster B 会通过 A.checkpoints.internel topic 获取 consumer c 的 state 信息，其中有 topicT —>A.topicT 的映射关系，但 B.topicT --> topicT 的反向映射关系是没有的， 但为了做 consumer offset 实时同步，这个反向映射也是要支持的，否则就是不完整的。 那么如何获得 offset 反向映射呢， 我们要利用 B 集群的 mm2-offset-syncs.A.internal 这个 topic， 里面记录的是 A 集群同步到 B 集群的 topic 的 broker offset mapping 信息。 我们可以从 B.topicT 的 offset 信息加上两个集群的 broker offset 的差值 delta 来计算出 B 集群中 topicT 对应的 offset 信息，计算方法如下： 现在我们假设 B.topicT 的 offset 为 offset，upstreamOffset 与 downstreamOffset 是 mm2-offset-syncs.A.internal 这个 topic 中记录的 broker offset 映射关系 delta = upstreamOffset - downstreamOffset reverseOffset = offset + delta 3.3、业务触发式切换集群实现 默认情况下，在集群切换的过程中还是需要业务介入的， 需要修改 target 集群的配置信息，起码 bootstrap.servers, topic 列表是要变更的， 然后重启恢复业务。 但重启业务对线上服务总是会有影响的，有没有办法可以在不重启业务的情况下完成切换呢？ 我们基于 librdkafka/kafka clients 封装了 c++, java 的客户端， 可以在客户端框架层面完成对切换动作的封装，这样上层业务就不需要做重启了。具体做法是基于 “客户端框架 + 配置中心” 的方式来实现。目前配置中心我们支持 apollo, mysql 的触发方式。业务在我们 ultron 平台发起切换流程，审批通过后触发切换集群，业务无需重启业务。 产品化 为了让业务容易理解使用热备方案，将双向同步，offset 映射等概念封装屏蔽掉， 我们在产品化过程中引入了热备属性的概念。 热备的属性与定义如下: 热备 topic：topic 在两个集群间双向同步 热备 producer：只生产热备 topic 的 producer， 且所有 topic 热备的 target 集群是需要一致的 热备 consumer：只消费热备 topic 的 consumer， 且所有 topic 热备的 target 集群是需要一致的， 两个热备 consumer group state 实时同步 业务申请热备 topic, producer,consumer 之后，我们 ultron 平台会完成如下动作: 对热备 topic 配置双向同步 两个集群热备的 consumer group state 实时同步 授权每个项目用户对应的 topic, group 读写权限 四、产品化 为了让业务容易理解使用热备方案，将双向同步，offset 映射等概念封装屏蔽掉， 在产品化过程中引入了热备属性的概念。 热备的属性与定义如下: 热备 topic：topic 在两个集群间双向同步 热备 producer：只生产热备 topic 的 producer， 且所有 topic 热备的 target 集群是需要一致的 热备 consumer：只消费热备 topic 的 consumer， 且所有 topic 热备的 target 集群是需要一致的， 两个热备 consumer group state 实时同步 业务申请热备 topic, producer,consumer 之后，平台会完成如下动作: 对热备 topic 配置双向同步 两个集群热备的 consumer group state 实时同步 授权每个项目用户对应的 topic, group 读写权限 对于热备 topic, 我们支持业务级别的切换，即 producer, consumer 级别的切换，因为有时候业务可能是写入部分 partition 或消费某个 group 有问题，这个时候只需要某个模块的按需切换即可。 如果是整个 kafka 集群的不可用或 IDC 不可用这种情况， 那么各个业务独立完成切换即可。 如果某个 IDC 的 kafka 集群不可用了，那么业务只需要在界面上申请切换集群，审批通过后，当前生效集群及配置会动态改变，由于我们的 consumer offset 信息是实时同步的， 业务修改参数后重启即可。如果业务基于我们的客户端框架则可实现触发式切换，业务无需重启。 五、需要注意的风险 这里主要考虑数据完整性风险。首先依赖于 producer 端的设置： producer 端有 retries=Long.MAX_VALUE，acks=ALL 设置， 在 send fail 情况下有持久化机制， 在切换集群后可以恢复发送。 如果 producer 没有 retry 机制，也没有持久化机制，那么 producer 端就会丢数据，那么数据保障性就无从谈起 其次是 MM2 日志同步的完整性，以上述集群 A (主集群), 集群 B (备集群) 为例： 如果集群 A 宕机且无法恢复， 那么 MM2 是会有少量日志丢失的。 如果集群 A 宕机且能恢复那么 MM2 能接着同步完，这时候 MM2 支持 at least once 语义。 参考： 基于 MM2 的跨 IDC kafka 热备多活方案 跨数据中心下的 Kafka 高可用架构分析
一、下载安装 下载地址：Offset Explorer 根据不同的系统下载对应的版本。 [atguigu@hadoop102 kafka]$ find ./libs/ -name *kafka_* | head -1 | grep -o 'kafka[^ ]*' kafka_2.12-3.0.0.jar 二、简单使用 2.1、连接kafka 设置完了，点击Test测试是否能连接，连接通了，然后点击Add，添加完成设置。出现如下界面。 2.2、简单使用 配置以字符串的形式显示kafka消息体，Tools -> Setting -> Topics，把Key、Message的Byte Array改成String，点ok。 或者通过如下界面配置。 注释：更改完Content Types，要点击Update和Refresh按钮。 再次查看kafka的数据：
如何保障kafka消费的顺序性，宗旨就是通过将消息绑定到定向的分区或者队列来保证顺序性，通过增加分区或者线程来提升消费能力。 一、生产者 要保证生产者发消息发的是顺序性的消息，发消息的时候指定一下key相同的key会发送到一个分区中，而分区时有序的在发消息的时候多个操作（下单，支付）保证顺序的话，保证这些操作在一个topic下的同一个分区，发送的时候指定key相同的key会去到同一个分区中，kafkaTemplate.send（topic,partition,key,data）有参数可以指定key。意思就是保证发消息的时候有序，发到同一个分区中，那么消费的时候也是有序的，因为分区可以理解为一个队列，消息先进先出。 当有消息被生产出来的时候，如果没有指定分区或者指定 key ，那么消费会按照【轮询】的方式均匀地分配到所有可用分区中，但不一定按照分区顺序来分配。 在 Kafka 中消费者可以订阅一个或多个主题，并被分配一个或多个分区。 如果一个消费者消费了多个分区，某些场景下消费者需要顺序地消费消息，但消息并不是按照顺序分配给分区的，所以就不一定能够保证消息消费的全局顺序性。 如何尽可能地保证消息消费的全局顺序性？（即这些消息具有因果关系）要想消费消息 B 必须先消费消息 A。 要注意的是，Kafka 的设计目标是提供高吞吐量和低延迟，而不是强制保证全局有序性。 所以该文探讨的是需要强调全局顺序性场景下的 Kafka 应用。 二、单分区 最简单粗暴的方法，虽然 Kafka 不能保证全局消费顺序性，但是能够保证分区内的消息顺序性。 可以只创建一个分区，并让消费者消费这个分区，这样就能够保证消费的消息是有序的。 但是这样做大大降低了吞吐量和处理效率，容易使得性能出现瓶颈。 三、基于key 在 Kafka 中，基于 key 的消息分配策略是通过消息中的键（key）来确定消息发送到哪个分区。 当生产者发送消息时，可以指定一个键（key），Kafka 使用这个键通过哈希算法来确定消息被发送到哪个分区。 四、代码 实现顺序性原理：消息发送的时候保证有序，设置相同的key会把消息投递到同一个分区的topic中，再由一个消费者来消费该分区topic。 topic: "topic_query_p3r1" 分配了三个partition分区。 4.1、producer 投递顺序消息 同一组行为设置相同的key，会把这组数据投递到同一分区topic中。 /** * 投递顺序性消息，根据用户id做取模推送到不同分区的topic中 * 相同的key推送到同一分区中 * 第一个参数：topic * 第二个参数：key * 第三个参数：发送的消息内容 * 三个参数全部是sring类型 */ @RequestMapping("/kafka2") public String testKafka2() { for (int userId = 0; userId < 3; userId++) { kafkaTemplate.send("topic_query_p3r1", userId + "", "insert" + userId); kafkaTemplate.send("topic_query_p3r1", userId + "", "update" + userId); kafkaTemplate.send("topic_query_p3r1", userId + "", "delete" + userId); } return null; 4.2、consumer 消费顺序消息 前提是生产者发消息的时候指定key了。 1、直接进行消费 因为投递的相同行为的消息是有序的，所以直接消费也不会有问题。 /** * 消费topic_query_p3r1主题，ConsumerGroupId1消费组 */ @KafkaListener(topics = "topic_query_p3r1", groupId = "ConsumerGroupId1") public void p3r2ConsumerGroupId0(ConsumerRecord<?, ?> consumer) throws InterruptedException { System.out.println("消费者A topic名称:" + consumer.topic() + ", key:" + consumer.key() + ", value:" + consumer.value() + ", 分区位置:" + consumer.partition() + ", 下标" + consumer.offset()+" "+Thread.currentThread().getId()); Thread.sleep(10); 2、一个消费者来指定具体分区进行消费 指定具体分区来进行消费。 /** * 消费者，解决消息顺序性 * 注解参数：partitions=0表示：只消费该主题中0分区的数据。 */ @KafkaListener(topicPartitions = {@TopicPartition(topic = "topic_query_p3r1", partitions = {"0"})}, groupId = "ConsumerGroupId1") public void receive(ConsumerRecord<?, ?> consumer) { System.out.println("消费者C topic名称:" + consumer.topic() + ",key:" + consumer.key() + "," + ",value:" + consumer.value() + "," + "分区位置:" + consumer.partition() + ", 下标" + consumer.offset()); } 3、多个消费者来指定不同分区进行消费。 写多个消费者方法来分别指向不同分区，提高消费速度，但是此方法不灵活。 /** * 消费0分区的topic_query_p3r1主题消费者，ConsumerGroupId1消费组 */ @KafkaListener(topicPartitions = {@TopicPartition(topic = "topic_query_p3r1", partitions = {"0"})}, groupId = "ConsumerGroupId1") public void p3r2ConsumerGroupId0(ConsumerRecord<?, ?> consumer) throws InterruptedException { System.out.println("消费者A topic名称:" + consumer.topic() + ", key:" + consumer.key() + ", value:" + consumer.value() + ", 分区位置:" + consumer.partition() + ", 下标" + consumer.offset()+" "+Thread.currentThread().getId()); Thread.sleep(10); } /** * 消费1分区的topic_query_p3r1主题消费者，ConsumerGroupId1消费组 */ @KafkaListener(topicPartitions = {@TopicPartition(topic = "topic_query_p3r1", partitions = {"1"})}, groupId = "ConsumerGroupId1") public void p3r2ConsumerGroupId1(ConsumerRecord<?, ?> consumer) throws InterruptedException { System.out.println("消费者A topic名称:" + consumer.topic() + ", key:" + consumer.key() + ", value:" + consumer.value() + ", 分区位置:" + consumer.partition() + ", 下标" + consumer.offset()+" "+Thread.currentThread().getId()); Thread.sleep(10); } /** * 消费2分区的topic_query_p3r1主题消费者，ConsumerGroupId1消费组 */ @KafkaListener(topicPartitions = {@TopicPartition(topic = "topic_query_p3r1", partitions = {"2"})}, groupId = "ConsumerGroupId1") public void p3r2ConsumerGroupId2(ConsumerRecord<?, ?> consumer) throws InterruptedException { System.out.println("消费者A topic名称:" + consumer.topic() + ", key:" + consumer.key() + ", value:" + consumer.value() + ", 分区位置:" + consumer.partition() + ", 下标" + consumer.offset()+" "+Thread.currentThread().getId()); Thread.sleep(10); } 4、多线程顺序消费 @RestController @Slf4j public class ShunXuConsumerMoreThread implements ApplicationRunner { @Autowired private KafkaTemplate<String, String> kafkaTemplate; // 使用两个内存队列 final int queueLingth = 2; // 创建两个内存队列 Queue<Map> queueA = new ConcurrentLinkedQueue<>(); Queue<Map> queueB = new ConcurrentLinkedQueue<>(); /** * 投递顺序性消息，根据用户id做取模推送到不同分区的topic中 * 相同的key推送到相同的分区中 */ @RequestMapping("/kafka2") public String testKafka2() { for (int userId = 0; userId < 300; userId++) { kafkaTemplate.send("topic_query_2", userId + "", "insert" + userId); kafkaTemplate.send("topic_query_2", userId + "", "update" + userId); kafkaTemplate.send("topic_query_2", userId + "", "delete" + userId); } return null; } /** * 主题消费者-把相同行为的数据放到同一内存队列中 */ @KafkaListener(topics = "topic_query_2", groupId = "ConsumerGroupId1") public void p3r2ConsumerGroupId0(ConsumerRecord<?, ?> consumer){ // 1.封装消息参数 Map param = new HashMap(); param.put("topic", consumer.topic()); param.put("key", consumer.key()); param.put("value", consumer.value()); param.put("p", consumer.partition()); // 2.把相同行为(key)数据添加到同一内存队列中 int queueHash = consumer.key().hashCode() % queueLingth; if (queueHash == 0) { queueA.add(param); } if (queueHash == 1) { queueB.add(param); } } // 开启两个线程消费内存队列中的消息 ApplicationRunner接口常用于项目启动后，（也就是ApringApplication.run()执行结束），立马执行某些逻辑。 //这里是立即启动两个线程 @Override public void run(ApplicationArguments args) { new Thread() { @Override public void run() { while (true) { if (queueA.size() > 0) { Map poll = queueA.poll(); //业务逻辑 System.out.println("Thrend-Id: " + Thread.currentThread().getId() + " topic：" + poll.get("topic") + " key:" + poll.get("key") + " value:" + poll.get("value") + " partition:" + poll.get("p")); try { Thread.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); } } } } }.start(); new Thread() { @Override public void run() { while (true) { if (queueB.size() > 0) { Map poll = queueB.poll(); //业务逻辑 System.out.println("Thrend-Id: " + Thread.currentThread().getId() + " topic：" + poll.get("topic") + " key:" + poll.get("key") + " value:" + poll.get("value") + " partition:" + poll.get("p")); try { Thread.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); } } } } }.start(); } } 此时消费是按照顺序的。
一、简介 Kafka 目前主要作为一个分布式的发布订阅式的消息系统使用，也是目前最流行的消息队列系统之一。本文集成的框架 spring-kafka。 源码解读： @Target ({ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @Retention (RetentionPolicy.RUNTIME) @MessageMapping @Documented @Repeatable (KafkaListeners.class) public @interface KafkaListener { /** * 消费者的 id，当 GroupId 没有被配置的时候，默认 id 为 GroupId */ String id () default ""; /** * 监听容器工厂，当监听时需要区分单数据还是多数据消费需要配置 containerFactory 属性 */ String containerFactory () default ""; /** * 需要监听的 Topic，可监听多个，和 topicPattern 属性互斥 */ String [] topics () default {}; /** * 需要监听的 Topic 的正则表达。和 topics,topicPartitions 属性互斥 */ String topicPattern () default ""; /** * 可配置更加详细的监听信息，必须监听某个 Topic 中的指定分区，或者从 offset 为 200 的偏移量开始监听，可配置该参数，和 topicPattern 属性互斥 */ TopicPartition [] topicPartitions () default {}; /** * 侦听器容器组 */ String containerGroup () default ""; /** * 监听异常处理器，配置 BeanName */ String errorHandler () default ""; /** * 消费组 ID */ String groupId () default ""; /** * id 是否为 GroupId */ boolean idIsGroup () default true; /** * 消费者 Id 前缀 */ String clientIdPrefix () default ""; /** * 真实监听容器的 BeanName，需要在 BeanName 前加 "__" */ String beanRef () default "__listener"; } 二、案例 2.1、ConsumerRecord 类消费 使用 ConsumerRecord 类接收有一定的好处，ConsumerRecord 类里面包含分区信息、消息头、消息体等内容，如果业务需要获取这些参数时，使用 ConsumerRecord 会是个不错的选择。如果使用具体的类型接收消息体则更加方便，比如说用 String 类型去接收消息体。 这里编写一个 Listener 方法，监听 topic1，并把 ConsumerRecord 里面所包含的内容打印到控制台中： @Component public class Listener { private static final Logger log = LoggerFactory.getLogger (Listener.class); @KafkaListener (id = "consumer", topics = "topic1") public void consumerListener (ConsumerRecord record) { log.info ("topic.quick.consumer receive :" + record.toString ()); } } 2.2、批量消费 批量消费在现实业务场景中是很有实用性的。因为批量消费可以增大 kafka 消费吞吐量，提高性能。 批量消费实现步骤： 1、重新创建一份新的消费者配置，配置为一次拉取 10 条消息。 2、创建一个监听容器工厂，命名为：batchContainerFactory，设置其为批量消费并设置并发量为 5，这个并发量根据分区数决定，必须小于等于分区数，否则会有线程一直处于空闲状态。 3、创建一个分区数为 8 的 Topic。 4、创建监听方法，设置消费 id 为 batchConsumer，clientID 前缀为 batch，监听 batch，使用 batchContainerFactory 工厂创建该监听容器。 @Component public class BatchListener { private static final Logger log = LoggerFactory.getLogger (BatchListener.class); private Map consumerProps () { Map props = new HashMap<>(); props.put (ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put (ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); props.put (ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); props.put (ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000"); // 一次拉取消息数量 props.put (ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "10"); props.put (ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, NumberDeserializers.IntegerDeserializer.class); props.put (ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class); return props; } @Bean ("batchContainerFactory") public ConcurrentKafkaListenerContainerFactory listenerContainer () { ConcurrentKafkaListenerContainerFactory container = new ConcurrentKafkaListenerContainerFactory (); container.setConsumerFactory (new DefaultKafkaConsumerFactory (consumerProps ())); // 设置并发量，小于或等于 Topic 的分区数 container.setConcurrency (5); // 必须 设置为批量监听 container.setBatchListener (true); return container; } @Bean public NewTopic batchTopic () { return new NewTopic ("topic.batch", 8, (short) 1); } @KafkaListener (id = "batchConsumer", clientIdPrefix = "batch" , topics = {"topic.batch"}, containerFactory = "batchContainerFactory") public void batchListener (List data) { log.info ("topic.batch receive :"); for (String s : data) { log.info (s); } } } 2.3、监听 Topic 中指定的分区 使用 @KafkaListener 注解的 topicPartitions 属性监听不同的 partition 分区。 @TopicPartition：topic 需要监听的 Topic 的名称，partitions 需要监听 Topic 的分区 id，partitionOffsets 可以设置从某个偏移量开始监听。 @PartitionOffset：partition 分区 Id，非数组，initialOffset 初始偏移量。 @Bean public NewTopic batchWithPartitionTopic () { return new NewTopic ("topic.batch.partition", 8, (short) 1); } @KafkaListener (id = "batchWithPartition",clientIdPrefix = "bwp",containerFactory = "batchContainerFactory", topicPartitions = { @TopicPartition (topic = "topic1",partitions = {"1","3"}), @TopicPartition (topic = "topic2",partitions = {"0","4"}, partitionOffsets = @PartitionOffset (partition = "2",initialOffset = "100")) } ) public void batchListenerWithPartition (List data) { log.info ("topic.batch.partition receive :"); for (String s : data) { log.info (s); } } 2.4、注解方式获取消息头及消息体 当你接收的消息包含请求头，以及你监听方法需要获取该消息非常多的字段时可以通过这种方式。。这里使用的是默认的监听容器工厂创建的，如果你想使用批量消费，把对应的类型改为 List 即可，比如 List data ， List key。 @Payload 获取的是消息的消息体，也就是发送内容。 @Header (KafkaHeaders.RECEIVED_MESSAGE_KEY) 获取发送消息的 key。 @Header (KafkaHeaders.RECEIVED_PARTITION_ID) 获取当前消息是从哪个分区中监听到的。 @Header (KafkaHeaders.RECEIVED_TOPIC) 获取监听的 TopicName。 @Header (KafkaHeaders.RECEIVED_TIMESTAMP) 获取时间戳。 @KafkaListener (id = "params", topics = "topic.params") public void otherListener (@Payload String data, @Header (KafkaHeaders.RECEIVED_MESSAGE_KEY) Integer key, @Header (KafkaHeaders.RECEIVED_PARTITION_ID) int partition, @Header (KafkaHeaders.RECEIVED_TOPIC) String topic, @Header (KafkaHeaders.RECEIVED_TIMESTAMP) long ts) { log.info ("topic.params receive : \n"+ "data :"+data+"\n"+ "key :"+key+"\n"+ "partitionId :"+partition+"\n"+ "topic :"+topic+"\n"+ "timestamp :"+ts+"\n" ); } 2.5、使用 Ack 机制确认消费 Kafka 是通过最新保存偏移量进行消息消费的，而且确认消费的消息并不会立刻删除，所以我们可以重复的消费未被删除的数据，当第一条消息未被确认，而第二条消息被确认的时候，Kafka 会保存第二条消息的偏移量，也就是说第一条消息再也不会被监听器所获取，除非是根据第一条消息的偏移量手动获取。Kafka 的 ack 机制可以有效的确保消费不被丢失。因为自动提交是在 kafka 拉取到数据之后就直接提交，这样很容易丢失数据，尤其是在需要事物控制的时候。 使用 Kafka 的 Ack 机制比较简单，只需简单的三步即可： 设置 ENABLE_AUTO_COMMIT_CONFIG=false，禁止自动提交。 设置 AckMode=MANUAL_IMMEDIATE。 监听方法加入 Acknowledgment ack 参数。 使用 Consumer.seek 方法，可以指定到某个偏移量的位置. @Component public class AckListener { private static final Logger log = LoggerFactory.getLogger (AckListener.class); private Map consumerProps () { Map props = new HashMap<>(); props.put (ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put (ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); props.put (ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); props.put (ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000"); props.put (ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class); props.put (ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); return props; } @Bean ("ackContainerFactory") public ConcurrentKafkaListenerContainerFactory ackContainerFactory () { ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory (); factory.setConsumerFactory (new DefaultKafkaConsumerFactory (consumerProps ())); factory.getContainerProperties ().setAckMode (AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE); factory.setConsumerFactory (new DefaultKafkaConsumerFactory (consumerProps ())); return factory; } @KafkaListener (id = "ack", topics = "topic.ack", containerFactory = "ackContainerFactory") public void ackListener (ConsumerRecord record, Acknowledgment ack) { log.info ("topic.quick.ack receive :" + record.value ()); ack.acknowledge (); } } 2.6、解决重复消费 上一节中使用 ack 手动提交偏移量时，假如 consumer 挂了重启，那它将从 committed offset 位置开始重新消费，而不是 consume offset 位置。这也就意味着有可能重复消费。 在 0.9 客户端中，有 3 种 ack 策略： 策略 1： 自动的，周期性的 ack。 策略 2：consumer.commitSync ()，调用 commitSync，手动同步 ack。每处理完 1 条消息，commitSync 1 次。 策略 3：consumer. commitASync ()，手动异步 ack。 那么使用策略 2，提交每处理完 1 条消息，就发送一次 commitSync。那这样是不是就可以解决 “重复消费” 了呢？如下代码： while (true) { List buffer = new ArrayList<>(); ConsumerRecords records = consumer.poll (100); for (ConsumerRecord record : records) { buffer.add (record); } insertIntoDb (buffer); // 消除处理，存到 db consumer.commitSync (); // 同步发送 ack buffer.clear (); } 答案是否定的！因为上面的 insertIntoDb 和 commitSync 做不到原子操作：如果在数据处理完成，commitSync 的时候挂了，服务器再次重启，消息仍然会重复消费。 那么如何解决重复消费的问题呢？答案是自己保存 committed offset，而不是依赖 kafka 的集群保存 committed offset，把消息的处理和保存 offset 做成一个原子操作，并且对消息加入唯一 id, 进行判重。 依照官方文档，要自己保存偏移量，需要: enable.auto.commit=false, 禁用自动 ack。 每次取到消息，把对应的 offset 存下来。 下次重启，通过 consumer.seek 函数，定位到自己保存的 offset，从那开始消费。 更进一步处理可以对消息加入唯一 id, 进行判重。 参考： 【spring-kafka】@KafkaListener详解与使用 Spring Boot与Kafka整合：实现百万级高吞吐消息处理的最佳实践
一、Kafka 硬件配置选择 1.1、场景说明 100 万日活，每人每天 100 条日志，每天总共的日志条数是 100 万 * 100 条 = 1 亿条。 1 亿 / 24 小时 / 60 分 / 60 秒 = 1150 条 / 每秒钟。 每条日志大小：0.5k - 2k（取 1k）。 1150 条 / 每秒钟 * 1k ≈ 1m/s 。 高峰期每秒钟：1150 条 * 20 倍 = 23000 条。 每秒多少数据量：20MB/s。 1.2、服务器台数选择 服务器台数 = 2 * （生产者峰值生产速率 * 副本 / 100） + 1 = 2 * （20m/s * 2 / 100） + 1 = 3 台 建议 3 台服务器。 1.3、磁盘选择 kafka 底层主要是顺序写，固态硬盘和机械硬盘的顺序写速度差不多。 建议选择普通的机械硬盘。 每天总数据量：1 亿条 * 1k ≈ 100g 100g * 副本 2 * 保存时间 3 天 / 0.7 ≈ 1T 建议三台服务器硬盘总大小，大于等于 1T。 1.4、内存选择 Kafka 内存组成：堆内存 + 页缓存 1）Kafka 堆内存建议每个节点：10g ~ 15g 在 kafka-server-start.sh 中修改 if ["x$KAFKA_HEAP_OPTS" = "x"]; then export KAFKA_HEAP_OPTS="-Xmx10G -Xms10G" fi （1）查看 Kafka 进程号。 [atguigu@hadoop102 kafka]$ jps 2321 Kafka 5255 Jps 1931 QuorumPeerMain （2）根据 Kafka 进程号，查看 Kafka 的 GC 情况。 参数说明： S0C：第一个幸存区的大小； S1C：第二个幸存区的大小 S0U：第一个幸存区的使用大小； S1U：第二个幸存区的使用大小 EC：伊甸园区的大小； EU：伊甸园区的使用大小 OC：老年代大小； OU：老年代使用大小 MC：方法区大小； MU：方法区使用大小 CCSC: 压缩类空间大小； CCSU: 压缩类空间使用大小 YGC：年轻代垃圾回收次数； YGCT：年轻代垃圾回收消耗时间 FGC：老年代垃圾回收次数； FGCT：老年代垃圾回收消耗时间 GCT：垃圾回收消耗总时间； （3）根据 Kafka 进程号，查看 Kafka 的堆内存。 [atguigu@hadoop102 kafka]$ jmap -heap 2321 Attaching to process ID 2321, please wait... Debugger attached successfully. Server compiler detected. JVM version is 25.212-b10 using thread-local object allocation. Garbage-First (G1) GC with 8 thread (s) Heap Configuration: MinHeapFreeRatio = 40 MaxHeapFreeRatio = 70 MaxHeapSize = 2147483648 (2048.0MB) NewSize = 1363144 (1.2999954223632812MB) MaxNewSize = 1287651328 (1228.0MB) OldSize = 5452592 (5.1999969482421875MB) NewRatio = 2 SurvivorRatio = 8 MetaspaceSize = 21807104 (20.796875MB) CompressedClassSpaceSize = 1073741824 (1024.0MB) MaxMetaspaceSize = 17592186044415 MB G1HeapRegionSize = 1048576 (1.0MB) Heap Usage: G1 Heap: regions = 2048 capacity = 2147483648 (2048.0MB) used = 246367744 (234.95458984375MB) free = 1901115904 (1813.04541015625MB) 11.472392082214355% used G1 Young Generation: Eden Space: regions = 83 capacity = 105906176 (101.0MB) used = 87031808 (83.0MB) free = 18874368 (18.0MB) 82.17821782178218% used Survivor Space: regions = 7 capacity = 7340032 (7.0MB) used = 7340032 (7.0MB) free = 0 (0.0MB) 100.0% used G1 Old Generation: regions = 147 capacity = 2034237440 (1940.0MB) used = 151995904 (144.95458984375MB) free = 1882241536 (1795.04541015625MB) 7.471886074420103% used 13364 interned Strings occupying 1449608 bytes. 2）页缓存：页缓存是 Linux 系统服务器的内存。我们只需要保证 1 个 segment（1g）中 25% 的数据在内存中就好。 每个节点页缓存大小 =（分区数 * 1g * 25%）/ 节点数。例如 10 个分区，页缓存大小 =（10 * 1g * 25%）/ 3 ≈ 1g 建议服务器内存大于等于 11G。 1.5、CPU 选择 num.io.threads = 8 负责写磁盘的线程数，整个参数值要占总核数的 50%。 num.replica.fetchers = 1 副本拉取线程数，这个参数占总核数的 50% 的 1/3。 num.network.threads = 3 数据传输线程数，这个参数占总核数的 50% 的 2/3。 建议 32 个 cpu core。 1.6、网络选择 网络带宽 = 峰值吞吐量 ≈ 20MB/s 选择千兆网卡即可。 100Mbps 单位是 bit；10M/s 单位是 byte ; 1byte = 8bit，100Mbps/8 = 12.5M/s。 一般百兆的网卡（100Mbps ）、千兆的网卡（1000Mbps）、万兆的网卡（10000Mbps）。 二、Kafka 生产者 3.1.1 Updating Broker Configs From Kafka version 1.1 onwards, some of the broker configs can be updated without restarting the broker. See the Dynamic Update Mode column in Broker Configs for the update mode of each broker config. read-only: Requires a broker restart for update per-broker: May be updated dynamically for each broker cluster-wide: May be updated dynamically as a cluster-wide default. May also be updated as a per-broker value for testing. 2.1、Kafka 生产者核心参数配置 参数： bootstrap.servers 生产者连接集群所需的 broker 地址清单。例如 hadoop102:9092,hadoop103:9092,hadoop104:9092，可以设置 1 个或者多个，中间用逗号隔开。注意这里并非需 要所有的 broker 地址，因为生产者从给定的 broker 里查 找到其他 broker 信息。 key.serializer 和 value.serializer 指定发送消息的 key 和 value 的序列化类型。一定要写全类名。 buffer.memory RecordAccumulator 缓冲区总大小，默认 32m。 batch.size 缓冲区一批数据最大值，默认 16k。适当增加该值，可 以提高吞吐量，但是如果该值设置太大，会导致数据传 输延迟增加。 linger.ms 如果数据迟迟未达到 batch.size，sender 等待 linger.time 之后就会发送数据。单位 ms，默认值是 0ms，表示没有 延迟。生产环境建议该值大小为 5-100ms 之间。 acks 0：生产者发送过来的数据，不需要等数据落盘应答。 1：生产者发送过来的数据，Leader 收到数据后应答。 -1（all）：生产者发送过来的数据，Leader + 和 isr 队列 里面的所有节点收齐数据后应答。默认值是 - 1，-1 和 all 是等价的。 max.in.flight.requests.per.connection 允许最多没有返回 ack 的次数，默认为 5，开启幂等性 要保证该值是 1-5 的数字。 retries 当消息发送出现错误的时候，系统会重发消息。retries 表 示重试次数。默认是 int 最大值，2147483647。如果设置了重试，还想保证消息的有序性，需要设置 MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1 否则在重试此失败消息的时候，其他的消息可能发送成 功了。 retry.backoff.ms 两次重试之间的时间间隔，默认是 100ms。 enable.idempotence 是否开启幂等性，默认 true，开启幂等性。 compression.type 生产者发送的所有数据的压缩方式。默认是 none，也就是不压缩。支持压缩类型：none、gzip、snappy、lz4 和 zstd。 2.2、生产者如何提高吞吐量 buffer.memory RecordAccumulator 缓冲区总大小，默认 32m。 batch.size 缓冲区一批数据最大值，默认 16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。 linger.ms 如果数据迟迟未达到 batch.size，sender 等待 linger.time 之后就会发送数据。单位 ms，默认值是 0ms，表示没有延迟。生产环境建议该值大小为 5-100ms 之间。 compression.type 生产者发送的所有数据的压缩方式。默认是 none，也就是不压缩。支持压缩类型：none、gzip、snappy、lz4 和 zstd。 2.3、数据可靠性 acks 0：生产者发送过来的数据，不需要等数据落盘应答。 1：生产者发送过来的数据，Leader 收到数据后应答。 -1（all）：生产者发送过来的数据，Leader + 和 isr 队列 里面的所有节点收齐数据后应答。默认值是 - 1，-1 和 all 是等价的。 至少一次（At Least Once）= ACK 级别设置为 - 1 + 分区副本大于等于 2 + ISR 里应 答的最小副本数量大于等于 2 2.4、数据去重 1）配置参数 enable.idempotence 是否开启幂等性，默认 true，表示开启幂等性。 2）Kafka 的事务一共有如下 5 个 API // 1 初始化事务 void initTransactions (); // 2 开启事务 void beginTransaction () throws ProducerFencedException; // 3 在事务内提交已经消费的偏移量（主要用于消费者） void sendOffsetsToTransaction (Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) throws ProducerFencedException; // 4 提交事务 void commitTransaction () throws ProducerFencedException; // 5 放弃事务（类似于回滚事务的操作） void abortTransaction () throws ProducerFencedException; 2.5、数据有序 单分区内，有序（有条件的，不能乱序）；多分区，分区与分区间无序； 2.6、数据乱序 enable.idempotence 是否开启幂等性，默认 true，表示开启幂等性。 max.in.flight.requests.per.connection 允许最多没有返回 ack 的次数，默认为 5，开启幂等性要保证该值是 1-5 的数字。 三、Kafka Broker 3.1、Broker 核心参数配置 Kafka Broker 总体工作流程 参数： replica.lag.time.max.ms ISR 中，如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值，默认 30s。 auto.leader.rebalance.enable 默认是 true。 自动 Leader Partition 平衡。建议关闭。 leader.imbalance.per.broker.percentage 默认是 10%。每个 broker 允许的不平衡的 leader 的比率。如果每个 broker 超过了这个值，控制器会触发 leader 的平衡。 leader.imbalance.check.interval.seconds 默认值 300 秒。检查 leader 负载是否平衡的间隔时间。 log.segment.bytes Kafka 中 log 日志是分成一块块存储的，此配置是指 log 日志划分 成块的大小，默认值 1G。 log.index.interval.bytes 默认 4kb，kafka 里面每当写入了 4kb 大小的日志（.log），然后就往 index 文件里面记录一个索引。 log.retention.hours Kafka 中数据保存的时间，默认 7 天。 log.retention.minutes Kafka 中数据保存的时间，分钟级别，默认关闭。 log.retention.ms Kafka 中数据保存的时间，毫秒级别，默认关闭。 log.retention.check.interval.ms 检查数据是否保存超时的间隔，默认是 5 分钟。 log.retention.bytes 默认等于 - 1，表示无穷大。超过设置的所有日志总大小，删除最早的 segment。 log.cleanup.policy 默认是 delete，表示所有数据启用删除策略；如果设置值为 compact，表示所有数据启用压缩策略。 num.io.threads 默认是 8。负责写磁盘的线程数。整个参数值要占总核数的 50%。 num.replica.fetchers 默认是 1。副本拉取线程数，这个参数占总核数的 50% 的 1/3 num.network.threads 默认是 3。数据传输线程数，这个参数占总核数的 50% 的 2/3 。 log.flush.interval.messages 强制页缓存刷写到磁盘的条数，默认是 long 的最大值，9223372036854775807。一般不建议修改，交给系统自己管理。 log.flush.interval.ms 每隔多久，刷数据到磁盘，默认是 null。一般不建议修改，交给系统自己管理。 3.2、服役新节点 / 退役旧节点 （1）创建一个要均衡的主题。 [atguigu@hadoop102 kafka]$ vim topics-to-move.json { "topics": [ {"topic": "first"} ], "version": 1 } （2）生成一个负载均衡的计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate （3）创建副本存储计划（所有副本存储在 broker0、broker1、broker2、broker3 中）。 [atguigu@hadoop102 kafka]$ vim increase-replication-factor.json （4）执行副本存储计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increasereplication-factor.json --execute （5）验证副本存储计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increasereplication-factor.json --verify 3.3、增加分区 1）修改分区数（注意：分区数只能增加，不能减少） [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 3 3.4、增加副本因子 1）创建 topic [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 3 --replication-factor 1 --topic four 2）手动增加副本存储 （1）创建副本存储计划（所有副本都指定存储在 broker0、broker1、broker2 中）。 [atguigu@hadoop102 kafka]$ vim increase-replication-factor.json 输入如下内容： {"version":1,"partitions":[{"topic":"four","partition":0,"replicas":[0,1,2]},{"topic":"four","partition":1,"replicas":[0,1,2]},{"topic":"four","partition":2,"replicas":[0,1,2]}]} （2）执行副本存储计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increasereplication-factor.json --execute 3.5、手动调整分区副本存储 （1）创建副本存储计划（所有副本都指定存储在 broker0、broker1 中）。 [atguigu@hadoop102 kafka]$ vim increase-replication-factor.json 输入如下内容： { "version":1, "partitions":[{"topic":"three","partition":0,"replicas":[0,1]}, {"topic":"three","partition":1,"replicas":[0,1]}, {"topic":"three","partition":2,"replicas":[1,0]}, {"topic":"three","partition":3,"replicas":[1,0]}] } （2）执行副本存储计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increasereplication-factor.json --execute （3）验证副本存储计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increasereplication-factor.json --verify 3.6、Leader Partition 负载平衡 auto.leader.rebalance.enable 默认是 true。自动 Leader Partition 平衡。生产环境中，leader 重选举的代价比较大，可能会带来性能影响，建议设置为 false 关闭。 leader.imbalance.per.broker.percentage 默认是 10%。每个 broker 允许的不平衡的 leader 的比率。如果每个 broker 超过了这个值，控制器会触发 leader 的平衡。 leader.imbalance.check.interval.seconds 默认值 300 秒。检查 leader 负载是否平衡的间隔时间。 3.7、自动创建主题 如果 broker 端配置参数 auto.create.topics.enable 设置为 true（默认值是 true），那么当生产者向一个未创建的主题发送消息时，会自动创建一个分区数为 num.partitions（默认值为 1）、副本因子为 default.replication.factor（默认值为 1）的主题。除此之外，当一个消费者开始从未知主题中读取消息时，或者当任意一个客户端向未知主题发送元数据请求时，都会自动创建一个相应主题。这种创建主题的方式是非预期的，增加了主题管理和维护的难度。生产环境建议将该参数设置为 false。 1）向一个没有提前创建 five 主题发送数据 [atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic five >hello world 2）查看 five 主题的详情 [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic five 四、Kafka 消费者 4.1、Kafka 消费者核心参数配置 消费者组初始化流程 1、coordinator：辅助实现消费者组的初始化和分区的分配。 coordinator 节点选择 = groupid 的 hashcode 值 % 50（ __consumer_offsets 的分区数量） 例如：groupid 的 hashcode 值 = 1，1% 50 = 1，那么__consumer_offsets 主题的 1 号分区，在哪个 broker 上，就选择这个节点的 coordinator 作为这个消费者组的老大。消费者组下的所有的消费者提交 offset 的时候就往这个分区去提交 offset。 消费者组详细消费流程 参数： bootstrap.servers 向 Kafka 集群建立初始连接用到的 host/port 列表。 key.deserializer 和 value.deserializer 指定接收消息的 key 和 value 的反序列化类型。一定要写全类名。 group.id 标记消费者所属的消费者组。 enable.auto.commit 默认值为 true，消费者会自动周期性地向服务器提交偏移量。 auto.commit.interval.ms 如果设置了 enable.auto.commit 的值为 true， 则该值定义了消费者偏移量向 Kafka 提交的频率，默认 5s。 auto.offset.reset 当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在（如，数据被删除了），该如何处理？ earliest：自动重置偏移量到最早的偏移量。 latest：默认，自动重置偏移量为最新的偏移量。 none：如果消费组原来的（previous）偏移量不存在，则向消费者抛异常。 anything：向消费者抛异常。 offsets.topic.num.partitions __consumer_offsets 的分区数，默认是 50 个分区。不建议修改。 heartbeat.interval.ms Kafka 消费者和 coordinator 之间的心跳时间，默认 3s。该条目的值必须小于 session.timeout.ms ，也不应该高于 session.timeout.ms 的 1/3。不建议修改。 session.timeout.ms Kafka 消费者和 coordinator 之间连接超时时间，默认 45s。超过该值，该消费者被移除，消费者组执行再平衡。 max.poll.interval.ms 消费者处理消息的最大时长，默认是 5 分钟。超过该值，该消费者被移除，消费者组执行再平衡。 fetch.min.bytes 默认 1 个字节。消费者获取服务器端一批消息最小的字节数。 fetch.max.wait.ms 默认 500ms。如果没有从服务器端获取到一批数据的最小字节数。该时间到，仍然会返回数据。 fetch.max.bytes 默认 Default: 52428800（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受 message.max.bytes （brokerconfig）or max.message.bytes （topic config）影响。 max.poll.records 一次 poll 拉取数据返回消息的最大条数，默认是 500 条。 4.2、消费者再平衡 heartbeat.interval.ms Kafka 消费者和 coordinator 之间的心跳时间，默认 3s。该条目的值必须小于 session.timeout.ms，也不应该高于 session.timeout.ms 的 1/3。 session.timeout.ms Kafka 消费者和 coordinator 之间连接超时时间，默认 45s。超过该值，该消费者被移除，消费者组执行再平衡。 max.poll.interval.ms 消费者处理消息的最大时长，默认是 5 分钟。超过该值，该消费者被移除，消费者组执行再平衡。 partition.assignment.strategy 消费者分区分配策略，默认策略是 Range +CooperativeSticky。Kafka 可以同时使用多个分区分配策略。可以选择的策略包括：Range、RoundRobin、Sticky、CooperativeSticky 4.3、指定 Offset 消费 kafkaConsumer.seek (topic, 1000); 4.4、指定时间消费 HashMap<TopicPartition, Long> timestampToSearch = new HashMap<>(); timestampToSearch.put (topicPartition, System.currentTimeMillis () -1 * 24 * 3600 * 1000); kafkaConsumer.offsetsForTimes (timestampToSearch); 4.5、消费者事务 4.6、消费者如何提高吞吐量 增加分区数； [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-serverhadoop102:9092 --alter --topic first --partitions 3 参数： fetch.max.bytes 默认 Default: 52428800（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受 message.max.bytes （broker config）or max.message.bytes （topic config）影响。 max.poll.records 一次 poll 拉取数据返回消息的最大条数，默认是 500 条 五、Kafka 总体 5.1、如何提升吞吐量 1）提升生产吞吐量 （1）buffer.memory：发送消息的缓冲区大小，默认值是 32m，可以增加到 64m。 （2）batch.size：默认是 16k。如果 batch 设置太小，会导致频繁网络请求，吞吐量下降；如果 batch 太大，会导致一条消息需要等待很久才能被发送出去，增加网络延时。 （3）linger.ms，这个值默认是 0，意思就是消息必须立即被发送。一般设置一个 5-100 毫秒。如果 linger.ms 设置的太小，会导致频繁网络请求，吞吐量下降；如果 linger.ms 太长，会导致一条消息需要等待很久才能被发送出去，增加网络延时。 （4）compression.type：默认是 none，不压缩，但是也可以使用 lz4 压缩，效率还是不错的，压缩之后可以减小数据量，提升吞吐量，但是会加大 producer 端的 CPU 开销。 2）增加分区 3）消费者提高吞吐量 （1）调整 fetch.max.bytes 大小，默认是 50m。 （2）调整 max.poll.records 大小，默认是 500 条。 4）增加下游消费者处理能力 5.2、数据精准一次 1）生产者角度 acks 设置为 - 1 （acks=-1）。 幂等性（enable.idempotence = true） + 事务 。 2）broker 服务端角度 分区副本大于等于 2 （--replication-factor 2）。 ISR 里应答的最小副本数量大于等于 2 （min.insync.replicas = 2）。 3）消费者 事务 + 手动提交 offset （enable.auto.commit = false）。 消费者输出的目的地必须支持事务（MySQL、Kafka）。 5.3、合理设置分区数 （1）创建一个只有 1 个分区的 topic。 （2）测试这个 topic 的 producer 吞吐量和 consumer 吞吐量。 （3）假设他们的值分别是 Tp 和 Tc，单位可以是 MB/s。 （4）然后假设总的目标吞吐量是 Tt，那么分区数 = Tt /min（Tp，Tc）。 例如：producer 吞吐量 = 20m/s；consumer 吞吐量 = 50m/s，期望吞吐量 100m/s； 分区数 = 100 / 20 = 5 分区 分区数一般设置为：3-10 个 分区数不是越多越好，也不是越少越好，需要搭建完集群，进行压测，再灵活调整分区个数。 5.4、单条日志大于 1m 参数： message.max.bytes 默认 1m，broker 端接收每个批次消息最大值。 max.request.size 默认 1m，生产者发往 broker 每个请求消息最大值。针对 topic 级别设置消息体的大小。 replica.fetch.max.bytes 默认 1m，副本同步数据，每个批次消息最大值。 fetch.max.bytes 默认 Default: 52428800（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受 message.max.bytes （broker config）or max.message.bytes （topic config）影响。 5.5、服务器挂了 在生产环境中，如果某个 Kafka 节点挂掉。 正常处理办法： （1）先尝试重新启动一下，如果能启动正常，那直接解决。 （2）如果重启不行，考虑增加内存、增加 CPU、网络带宽。 （3）如果将 kafka 整个节点误删除，如果副本数大于等于 2，可以按照服役新节点的方式重新服役一个新节点，并执行负载均衡。 5.6、集群压力测试 1）Kafka 压测 用 Kafka 官方自带的脚本，对 Kafka 进行压测。 生产者压测：kafka-producer-perf-test.sh 消费者压测：kafka-consumer-perf-test.sh 2）Kafka Producer 压力测试 （1）创建一个 test topic，设置为 3 个分区 3 个副本 [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrapserver hadoop102:9092 --create --replication-factor 3 --partitions 3 --topic test （2）在 /opt/module/kafka/bin 目录下面有这两个文件。 [atguigu@hadoop105 kafka]$ bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=16384 linger.ms=0 参数说明： record-size 是一条信息有多大，单位是字节，本次测试设置为 1k。 num-records 是总共发送多少条信息，本次测试设置为 100 万条。 throughput 是每秒多少条信息，设成 - 1，表示不限流，尽可能快的生产数据，可测出生产者最大吞吐量。本次实验设置为每秒钟 1 万条。 producer-props 后面可以配置生产者相关参数，batch.size 配置为 16k。 （3）调整 batch.size 大小 1、batch.size 默认值是 16k。本次实验 batch.size 设置为 32k。 [atguigu@hadoop105 kafka]$ bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=32768 linger.ms=0 2、batch.size 默认值是 16k。本次实验 batch.size 设置为 4k。 [atguigu@hadoop105 kafka]$ bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=4096 linger.ms=0 （4）调整 linger.ms 时间 linger.ms 默认是 0ms。本次实验 linger.ms 设置为 50ms [atguigu@hadoop105 kafka]$ bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=4096 linger.ms=50 （5）调整压缩方式 1、默认的压缩方式是 none。本次实验 compression.type 设置为 snappy。 [atguigu@hadoop105 kafka]$ bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=4096 linger.ms=50 compression.type=snappy 2、默认的压缩方式是 none。本次实验 compression.type 设置为 zstd。 [atguigu@hadoop105 kafka]$ bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=4096 linger.ms=50 compression.type=zstd 3、默认的压缩方式是 none。本次实验 compression.type 设置为 gzip。 [atguigu@hadoop105 kafka]$ bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=4096 linger.ms=50 compression.type=gzip 4、默认的压缩方式是 none。本次实验 compression.type 设置为 lz4。 [atguigu@hadoop105 kafka]$ bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=4096 linger.ms=50 compression.type=lz4 （6）调整缓存大小 默认生产者端缓存大小 32m。本次实验 buffer.memory 设置为 64m。 [atguigu@hadoop105 kafka]$ bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=4096 linger.ms=50 buffer.memory=67108864 3）Kafka Consumer 压力测试 （1）修改 /opt/module/kafka/config/consumer.properties 文件中的一次拉取条数为 500 max.poll.records=500 （2）消费 100 万条日志进行压测 [atguigu@hadoop105 kafka]$ bin/kafka-consumer-perf-test.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --topic test --messages 1000000 --consumer.config config/consumer.properties 参数说明： --bootstrap-server 指定 Kafka 集群地址。 --topic 指定 topic 的名称。 --messages 总共要消费的消息个数。本次实验 100 万条。 （3）一次拉取条数为 2000 1、修改 /opt/module/kafka/config/consumer.properties 文件中的一次拉取条数为 2000。 max.poll.records=2000 2、再次执行 [atguigu@hadoop105 kafka]$ bin/kafka-consumer-perf-test.sh --broker-list hadoop102:9092,hadoop103:9092,hadoop104:9092 --topic test --messages 1000000 --consumer.config config/consumer.properties （4）调整 fetch.max.bytes 大小为 100m 1、修改 /opt/module/kafka/config/consumer.properties 文件中的拉取一批数据大小 100m。 fetch.max.bytes=104857600 2、再次执行 [atguigu@hadoop105 kafka]$ bin/kafka-consumer-perf-test.sh --broker-list hadoop102:9092,hadoop103:9092,hadoop104:9092 --topic test --messages 1000000 --consumer.config config/consumer.properties