Kafka学习---消费者（分区消费、分区平衡策略、offset、漏消费和重复消费）-阿里云开发者社区

1、消费者

1.1 Kafka消费方式

1、pull（拉）模式：consumer采用从broker中主动拉取数据。

2、push（推）模式：Kafka没有采用这种方式。因为broker决定消息发生速率，很难适应所有消费者的消费速率。例如推送的速度是50M/s，Consumer1、Consumer2就来不及处理消息。

pull模式不足之处是如果Kafka没有数据，消费者可能会陷入循环中，一直返回空数据。

1.2 Kafka消费者工作流程

1.2.1 消费者总体工作流程

1.2.2 消费者组原理

Consumer Group（CG）：消费者组，由多个consumer组成。形成一个消费者组的条件，是所有消费者的groupid相同。

消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费。

消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。

1、消费者组初始化流程

（1）coordinator：辅助实现消费者组的初始化和分区的分配

coordinator节点选择=groupid的hashcode值%50（_consumer_offsets的分区数量）

例如：groupid的hashcode值=1,1%50=1，那么_consumer_offsets主题的1号分区，在哪个broker上，就选择这个节点的coordinator作为这个消费者组的老大。消费者组下的所有的消费者提交offset的时候就往这个分区去提交offset。

2、消费者组详细消费流程

6.5 subscribe 与 assign 的区别

通过 subscribe()方法订阅主题具有消费者自动再均衡功能；

在多个消费者的情况下可以根据分区分配策略来自动分配各个消费者与分区的关系。当消费组的消费

者增加或减少时，分区分配关系会自动调整，以实现消费负载均衡及故障自动转移。

assign() 方法订阅分区时，是不具备消费者自动均衡的功能的；

其实这一点从 assign()方法参数可以看出端倪，两种类型 subscribe()都有 ConsumerRebalanceListener

类型参数的方法，而 assign()方法却没有

kafka的消费组再均衡现象观察

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import java.io.IOException;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Properties;
public class ConsumerDemo3 {
    public static void main(String[] args) throws IOException {
        // jj 主题： 3个分区
        // xx 主题： 2个分区
        Properties properties = new Properties();
        properties.load(ConsumerDemo3.class.getClassLoader().getResourceAsStream("consumer.properties"));
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"d30-2");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
//        consumer.subscribe(Collections.singletonList(""));
        consumer.subscribe(Arrays.asList("jj", "xx"), new ConsumerRebalanceListener() {
            // 再均衡过程中，消费者会被取消先前所分配的主题，分区
            // 取消了之后，consumer底层就会调用下面的方法
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                System.out.println("我被取消了如下主题，分区：" + partitions);
            }
            // 再均衡过程中，消费者会被重新分配到新的主题，分区
            // 分配好了新的主题，分区后，consumer底层调用下面的方法
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                System.out.println("我又被分配了如下主题，分区：" + partitions);
            }
        });
        while (true){
            consumer.poll(Duration.ofMillis(Long.MAX_VALUE));
        }
    }
}

组内各消费者，再均衡分区，只针对有多个消费者共同都订阅的主题
jvm:STOP THE WORLD
STW
通过 subscribe()方法订阅主题具有消费者自动再均衡功能 ；
在多个消费者的情况下可以根据分区分配策略来自动分配各个消费者与分区的关系。当消费组的消费
者增加或减少时，分区分配关系会自动调整，以实现消费负载均衡及故障自动转移。

当在用一个消费者去消费jj

先定每个消费者要分几个蛋糕，然后按patition编号的范围，区划分，如果共同订阅的主题比较多,那么排在前面的消费者拿到的分区明显多于后面的消费者

消费者组再均衡分区分配策略

消费者组的意义何在？

为了提高数据处理的并行度！

将分区的消费权从一个消费者移到另一个消费者称为再均衡（rebalance），如何 rebalance 也涉及到

分区分配策略。

kafka 有两种的分区分配策略：range（默认） 和 round robin（新版本中又新增了另外 2 种）

我们可以通过 partition.assignment.strategy 参数选择 range 或 roundrobin。

partition.assignment.strategy 参数默认的值是 range。

partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor

partition.assignment.strategy=org.apache.kafka.clients.consumer.RangeAssignor

这个参数属于“消费者”参数！

消费者组再均衡流程

消费组在消费数据的时候，有两个角色进行组内的各事务的协调；

角色 1： Group Coordinator （组协调器）位于服务端（就是某个 broker）

角色 2： Group Leader （组长）位于消费端（就是消费组中的某个消费者）

分区.分配.策略

// 指定消费者再均衡策略
properties.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RangeAssignor.class.getName());

类名或类类型的列表（按首选项排序）支持的分区分配策略，客户端将在使用组管理时使用者实例之间分配分区所有权。可用选项包括：

org.apache.kafka.clients.consumer.RangeAssignor：按主题分配分区。
org.apache.kafka.clients.consumer.RoundRobinAssignor：以循环方式将分区分配给使用者。
org.apache.kafka.clients.consumer.StickyAssignor：保证分配最大程度的平衡，同时保留尽可能多的现有分区分配。
org.apache.kafka.clients.consumer.CooperativeStickyAssignor：遵循相同的 StickyAssignor 逻辑，但允许协作再平衡。

默认分配器是 [RangeAssignor， CooperativeStickyAssignor]，默认情况下它将使用 RangeAssignor，但允许升级到 CooperativeStickyAssignor，只需一次滚动跳出即可从列表中删除 RangeAssignor。

通过实现该接口，您可以插入自定义分配策略。org.apache.kafka.clients.consumer.ConsumerPartitionAssignor

类型：	列表
违约：	类 org.apache.kafka.clients.consumer.RangeAssignor，类 org.apache.kafka.clients.consumer.CooperativeStickyAssignor
取值范围：	非 null 字符串
重要性：	中等

Sticky Strategy

对应的类叫做： org.apache.kafka.clients.consumer.StickyAssignor

sticky 策略的特点：

要去打成最大化的均衡

尽可能保留各消费者原来分配的分区

再均衡的过程中，还是会让各消费者先取消自身的分区，然后再重新分配（只不过是分配过程中会尽

量让原来属于谁的分区依然分配给谁

Cooperative Sticky Strategy

对应的类叫做： org.apache.kafka.clients.consumer.ConsumerPartitionAssignor

sticky 策略的特点：

逻辑与 sticky 策略一致

支持 cooperative 再均衡机制（再均衡的过程中，不会让所有消费者取消掉所有分区然后再进行重

分配）

定位 Group Coordinator

每一个消费组,记录消费的偏移量，是会记录在_consumer_offsets主题中的第一个固定分区中,_consumer_offsets主题有50个分区(可配置server.properties)
组id_hadecode%50
x组协调器是在哪台broker?
就是x组写消费位移的上述分区的leader副本所在的broker

coordinator 在我们组记偏移量的__consumer_offsets 分区的 leader 所在 broker 上

查找 Group Coordinator 的方式：

先根据消费组 groupid 的 hashcode 值计算它应该所在＿consumer_offsets 中的分区编号；

Utils.abc(groupId.hashCode) % groupMetadataTopicPartitionCount

groupMetadataTopicPartitionCount 为 __consumer_offsets 的分区总数，这个可以通过 broker 端参数

offset.topic.num.partitions 来配置，默认值是 50；

阶段 2：加入组 Join The Group

此阶段的重要操作之 1：选举消费组的 leader

private val members = new mutable.HashMap[String, MemberMetadata]

var leaderid = members.keys.head

此阶段的重要操作之 2：选择分区分配策略

最终选举的分配策略基本上可以看作被各个消费者支持的最多的策略，具体的选举过程如下：

（1）收集各个消费者支持的所有分配策略，组成候选集 candidates

（2）每个消费者从候选集 candidates 找出第一个自身支持的策略，为这个策略投上一票。

（3）计算候选集中各个策略的选票数，选票数最多的策略即为当前消费组的分配策略。

其实，此逻辑并不需要 consumer 来执行，而是由 Group Coordinator 来执行

消费组 leader 的选举，策略就是：随机！

阶段 3：组信息同步 SYNC Group

此阶段，主要是由消费组 leader 将分区分配方案，通过 Group Coordinator 来转发给组中各消费者

阶段 4：心跳联系 HEART BEAT

进入这个阶段之后，消费组中的所有消费者就会处于正常工作状态。

各消费者在消费数据的同时，保持与 Group Coordinator 的心跳通信；

消费者的心跳间隔时间由参数 heartbeat.interval.ms 指定，默认值为 3000 ，即这个参数必须比

session.timeout.ms 参数设定的值要小；一般情况下 heartbeat.interval.ms 的配置值不能超过

session.timeout.ms 配置值的 1/3 。这个参数可以调整得更低，以控制正常重新平衡的预期时间；

如果一个消费者发生崩溃，并停止读取消息，那么 GroupCoordinator 会等待一小段时间确认这个消费

者死亡之后才会触发再均衡。在这一小段时间内，死掉的消费者并不会读取分区里的消息。

这个一小段时间由 session.timeout. ms 参数控制，该参数的配置值必须在 broker 端参数

group.min.session.timeout. ms （默认值为 6000 ，即 6 秒）和 group.max.session. timeout. ms （默认

值为 300000 ，即 5 分钟）允许的范围内。

1.2.3 消费者重要参数

1.3 消费者API

6.5 subscribe 与 assign 的区别
 通过 subscribe()方法订阅主题具有消费者自动再均衡功能 ；
在多个消费者的情况下可以根据分区分配策略来自动分配各个消费者与分区的关系。当消费组的消费
者增加或减少时，分区分配关系会自动调整，以实现消费负载均衡及故障自动转移。
 assign() 方法订阅分区时，是不具备消费者自动均衡的功能的；
其实这一点从 assign()方法参数可以看出端倪，两种类型 subscribe()都有 ConsumerRebalanceListener
类型参数的方法，而 assign()方法却没有。

1.3.1 独立消费者案例（订阅主题）

1、需求

创建一个独立的消费者，消费first主题中的数据

2、实现步骤

（1）创建包名：com.zhm.consumer

（2）编写代码

package org.zhm.consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;
/**
 * @ClassName CustomConsumer
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/6/14 9:08
 * @Version 1.0
 */
public class CustomConsumer {
    public static void main(String[] args) {
        //1、创建消费者的配置对象
        Properties properties=new Properties();
        //2、给消费者配置对象添加参数
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
        //配置反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        //配置消费者组（组内任意起名） 必须
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test2");
//        //修改分区分配策略
//        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");
        //测试：修改分区分配策略
        ArrayList<String> startegys=new ArrayList<>();
        startegys.add("org.apache.kafka.clients.consumer.StickyAssignor");
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,startegys);
        //3、创建消费者对象
        KafkaConsumer<String,String> kafkaConsumer=new KafkaConsumer<String, String>(properties);
        //注册要消费的主题（可以消费多个主题）
        ArrayList<String> topics=new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);
        //拉取数据打印
        while (true){
            //设置1s中消费一批数据
            ConsumerRecords<String,String> consumerRecords=kafkaConsumer.poll(Duration.ofSeconds(1));
            //打印消费到的数据
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}

独立消费者案例（订阅分区）

1、需求：创建一个独立消费者，消费first主题0号分区的数据

）实现步骤

（1）代码编写

package org.zhm.consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import javax.lang.model.type.ArrayType;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;
/**
 * @ClassName CustomConsumerPartition
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/6/14 9:18
 * @Version 1.0
 */
public class CustomConsumerPartition {
    public static void main(String[] args) {
        Properties properties=new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        //配置消费者组（必须）
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
        KafkaConsumer<String,String> kafkaConsumer=new KafkaConsumer<String, String>(properties);
        //消费某个主题的某个分区数据
        ArrayList<TopicPartition> topicPartitions=new ArrayList<>();
        topicPartitions.add(new TopicPartition("first",0));
        kafkaConsumer.assign(topicPartitions);
        while (true){
            ConsumerRecords<String,String> consumerRecords=kafkaConsumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}

3、测试

（1）在 IDEA 中执行消费者程序。

（2）在 IDEA 中执行生产者程序 CustomProducerCallback()在控制台观察生成几个 0 号分区的数据。

（3）在 IDEA 控制台，观察接收到的数据，只能消费到 0 号分区数据表示正确。

1.3.3 消费者组案例

2、案例实操

（1）复制一份基础消费者的代码，在 IDEA 中同时启动，即可启动同一个消费者组中的两个消费者。

（2）启动代码中的生产者发送消息，在 IDEA 控制台即可看到两个消费者在消费不同分区的数据（如果只发生到一个分区，可以在发送时增加延迟代码 Thread.sleep(2);）

（3）重新发送到一个全新的主题中，由于默认创建的主题分区数为 1，可以看到只能有一个消费者消费到数据。

1.4 生产经验——分区的分配以及再平衡

1、一个consumer group中有多个consumer组成，一个topic有多个partition组成，现在的问题是到底是由哪个consumer来消费哪个分区的数据。

2、kafka有四种主流的分区分配策略：Range、RoundRobin、Sticky、CooperativeSticky。可以通过配置参数partition.assignment.strategy，修改分区的分配策略。默认策略是Range+CooperativeSticky。kafka可以同时使用多个分区分配策略。

1.4.1 Range以及再平衡

1、Range分区策略原理

Range是对每个topic而言的。

首先对同一个topic里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。

假如现在有 7 个分区，3 个消费者，排序后的分区将会是0,1,2,3,4,5,6；消费者排序完之后将会是C0,C1,C2。

通过partition数/consumer数来决定每个消费者应该消费几个分区，如果除不尽，那么前面几个消费者将会多消费1个分区。

注意：如果只是针对一个topic而言，c0消费者多消费1个分区影响不是很大。但是如果有N多个topic，那么针对每个topic，消费者c0都将会多消费一个分区，topic越多，c0消费者的分区会比其他消费者明显多消费n个分区。

容易产生数据倾斜

2、Range分区分配策略案例

（1）修改主题first为7个分区

 bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 7

注意：分区数可以增加，但是不能减少

（2）复制 CustomConsumer 类，创建 CustomConsumer2。这样可以由三个消费者CustomConsumer、CustomConsumer1、CustomConsumer2 组成消费者组，组名都为“test”，同时启动 3 个消费者。

（3）启动 CustomProducer 生产者，发送 500 条消息，随机发送到不同的分区。

3、Range 分区分配再平衡案例

（1）停止掉 0 号消费者，快速重新发送消息观看结果（45s 以内，越快越好）。

1 号消费者：消费到4、5、6号分区数据。

2 号消费者：消费到2、3号分区数据。

0 号消费者的任务会整体被分配到 1 号消费者或者 2 号消费者。

说明：0 号消费者挂掉后，消费者组需要按照超时时间 45s 来判断它是否退出，所以需要等待，时间到了 45s 后，判断它真的退出就会把任务分配给其他 broker 执行。

（2）再次重新发送消息观看结果（45s 以后）。

1 号消费者：消费到 0、1、2、3 号分区数据。

2 号消费者：消费到 4、5、6 号分区数据。

说明：消费者 0 已经被踢出消费者组，所以重新按照 range 方式分配。

1.4.2 RoundRobin以及再平衡

1、RoundRobin 分区策略原理

RoundRobin针对集群中所以Topic而言。

RoundRobin轮询分区策略，是把所有的partition和所有的consumer都列出来，然后按照hashcode进行排序，最后通过轮询算法来分配partition给到各个消费者。

1.4.3 Sticky以及再平衡

粘性分区定义：可以理解为分配的结果带有“粘性的”。即在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可以节省大量的开销。

粘性分区是 Kafka 从 0.11.x 版本开始引入这种分配策略，首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。

1.5 offset位移

1.5.1 offset的默认维护位置

_consumer_offsets主题里面采用key和value的方式储存数据。key是group.id+topic+分区号，value值就是当前offset的值。每隔一段时间，kafka内部会对这个topic进行compact，也就是每个group.id+topic+分区号就保留最新数据。

1.5.2 自动提交offset

为了使我们能够专注于自己的业务逻辑，Kafka提供了自动提交offset的功能。

自动提交offset的相关参数：

enable.auto.commit:是否开启自动提交offset功能，默认是true

auto.commit.interval.ms:自动提交offset的时间间隔，默认是5s。

1.5.3 手动提交offset

虽然自动提交offset十分简单便利，但由于其是基于时间提交的，开发人员难以把握offset提交的时机。因此Kafka还提供了手动提交offset的API。

手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。两者的相同点是，都会将本次提交的一批数据最高的偏移量提交；不同点是，同步提交阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；而异步提交则没有失败重试机制，故有可能提交失败。

commitSync(同步提交)：必须等待offset提交完毕，再去消费下一批数据。

commitAsync（异步提交）：发送完提交offset请求后，就开始消费下一批数据。

1.5.4 指定offset消费

auto.offset.reset = earliest | latest | none 默认是 latest。

当 Kafka 中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量时（例如该数据已被删除），该怎么办？

（1）earliest：自动将偏移量重置为最早的偏移量，–from-beginning。

（2）latest（默认值）：自动将偏移量重置为最新偏移量。

（3）none：如果未找到消费者组的先前偏移量，则向消费者抛出异常。

（4）任意指定 offset 位移开始消费

1.5.5 漏消费和重复消费

重复消费：已经消费了数据，但是 offset 没提交。

漏消费：先提交 offset 后消费，有可能会造成数据的漏消费。

.6 消费者事务

如果想完成Consumer端的精准一次性消费，那么需要Kafka消费端将消费过程和提交offset过程做原子绑定。此时我们需要将Kafka的offset保存到支持事务的自定义介质（比如MySQL）。

1.7 数据积压（消费者如何提高吞吐量）

1、如果是Kafka消费能力不足，则可以考虑增加Topic的分区数，并且同时提升消费组的消费者数量，消费者数 = 分区数。（两者缺一不可）

2、如果是下游的数据处理不及时：提高每批次拉取的数量。批次拉取数据过少（拉取数据/处理时间 < 生产速度），

使处理的数据小于生产的数据，也会造成数据积压。

CAP理论:
生产者的数据写入原理
CAP原理：数据一致性，数据的可靠性，分布式容错性，kafka的事务机制
消费者在均衡原理

kafka编程练习题(需求)

一个正常的消费逻辑需要具备以下几个步骤：
（1）配置消费者客户端参数及创建相应的消费者实例；
（2）订阅主题 topic；
（3）拉取消息并消费；
（4）定期向__consumer_offsets 主题提交消费位移 offset；
（5）关闭消费者实例。

需求1：
* 写一个生产者，不断去生成 “用户行为事件”数据并写入kafka
* {"guid":1,"eventId":"pageview","timeStamp":1637868346789}
* {"guid":1,"eventId":"addcart","timeStamp":1637868346966}
* {"guid":2,"eventId":"applaunch","timeStamp":1637868346967}
* .....
*

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.RandomStringUtils;
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.Properties;
public class Kafka编程练习 {
    public static void main(String[] args) throws InterruptedException {
        MyDataGen myDataGen = new MyDataGen();
        myDataGen.genData();
    }
}
/**
 * 业务数据生成器
 */
class MyDataGen{
    KafkaProducer<String, String> producer;
    public MyDataGen(){
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"doit01:9092");
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producer = new KafkaProducer<>(props);
    }
    public void genData() throws InterruptedException {
        UserEvent userEvent = new UserEvent();
        while(true){
            // 造一条随机的用户行为事件数据对象
            userEvent.setGuid(RandomUtils.nextInt(1,10000));
            userEvent.setEventId(RandomStringUtils.randomAlphabetic(5,8));
            userEvent.setTimeStamp(System.currentTimeMillis());
            // 转成json串
            String json = JSON.toJSONString(userEvent);
            // 将业务数据封装成ProducerRecord对象
            ProducerRecord<String, String> record = new ProducerRecord<>("doit30-events", json);
            // 用producer写入kafka
            producer.send(record);
            // 控制发送的速度
            Thread.sleep(RandomUtils.nextInt(200,1500));
        }
    }
}
@NoArgsConstructor
@AllArgsConstructor
@Getter
@Setter
class UserEvent{
    private long guid;
    private String eventId;
    private long timeStamp;
    private Integer flag;
}

Kafka学习---消费者（分区消费、分区平衡策略、offset、漏消费和重复消费）