😀😀😀创作不易，各位看官点赞收藏.

文章目录

Kafka 学习笔记
- 1、消息队列 MQ
- 2、Kafka 下载安装
- - 2.1、Zookeeper 方式启动
  - 2.2、KRaft 协议启动
  - 2.3、Kafka 集群搭建
- 3、Kafka 之生产者
- - 3.1、Java 生产者 API
  - 3.2、Kafka 生产者生产分区
  - 3.3、Kafka 生产者常见问题
- 4、Kafka 之 Broker
- - 4.1、Broker 节点上下线
  - 4.2、Broker 副本
  - 4.3、Broker 文件存储机制
- 5、Kafka 之消费者
- - 5.1、消费者组
  - 5.2、Java 消费者 API
  - 5.3、消费者分区分配
  - 5.4、消费者 offset 维护
  - 5.5、消费者常见问题
- 6、Kafka-Eagle 监控
- 7、Spring Boot 整合 Kafka
- - 7.1、Kafka 生产者
  - 7.2、Kafka 消费者

Kafka 学习笔记

Kafka：是一个开源的分布式事件流平台，用于高性能数据管道、流分析、数据集成和关键任务应用。在一些大数据领域中通常使用 kafka 作为消息队列，在 JavaEE 开发中也有 ActiveMQ、RabbitMQ、RocketMQ 等等消息队列。

1、消息队列 MQ

消息队列是一种在分布式系统中用于不用组件之间传递和处理数据的通信机制，基于异步通信模式，允许发送者将消息发送到队列中，接受者从队列中获取消息数据并进行处理。

消息队列几种模式：

点对点模式：消息生产者将消息放入队列，每一条消息只能被一个消费者消费，消费者将消息处理完以后会将消息从队列中移除，这种适合单一消息被一个消费者处理的场景。

发布 - 订阅模式：生产者将消息发布到一个主题中，多个消费者可以订阅主题来接收消息，每一个消费者都会收到相同的消息，消息会被保存即使被消费也不会被删除。

应用场景：

限流消峰：MQ 可以将系统超量的请求进行暂存，以便后期系统进行处理调度，从而避免请求的的丢失和系统被压垮。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-gLys6siY-1692610909240)(https://jx-image-storage.oss-cn-hangzhou.aliyuncs.com/image/image-20221125185444191-16917431184751.png)]

异步和解耦：上游系统去调用下游系统时采用同步调用方式，系统的吞吐量会大大降低，并且上下游系统的耦合度增加。一般会在上下游系统之间添加一个MQ，上游系统将消息数据给 MQ 然后直接返回给用户，后面的所有操作由 MQ 进行请求下游操作，如果失败了就进行重试。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-bUhYcjMH-1692610909241)(https://jx-image-storage.oss-cn-hangzhou.aliyuncs.com/image/image-20221125190838306-16917431184763.png)]

数据收集：分布式系统会产生大量的数据，例如业务日志、监控数据等。针对这些数据进行实时采集和处理，然后对数据进行分析操作，MQ 也可以完成这类操作。

2、Kafka 下载安装

下载地址https://www.apache.org/dyn/closer.cgi?path=/kafka/3.5.0/kafka_2.13-3.5.0.tgz

# 解压
tar -zxvf kafka_2.13-3.5.0.tgz 
cd kafka_2.13-3.5.0

注意：

在 kafka2.8.0 之前必须需要依赖 zookeeper 组件，在之后可以选择不依赖 zookeeper 组件，而是以 KRaft 协议启动。
kafka 需要 Java 环境，需要配置环境变量。

2.1、Zookeeper 方式启动

配置文件 - server.properties ：

# 常用配置
# 身份唯一标识，不能重复
broker.id=0
# 日志文件
log.dirs=/tmp/kafka-logs
# 依赖的zookeeper节点地址，一般会加一个kafka节点
zookeeper.connect=localhost:2181/kafka	
# 与zookeeper连接超时时间
zookeeper.connection.timeout.ms=18000

启动 kafka 服务：

# 先启动一个 kafka 自带的 zookeeper，-daemon 后台运行
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

# 启动 kafka 服务
bin/kafka-server-start.sh -daemon config/server.properties

2.2、KRaft 协议启动

配置文件：修改 config/KRaft/server.properties 文件。

# 生成集群UUID(只执行一次)
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
# 格式化日志目录(只执行一次)
bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties
# 启动kafka服务
bin/kafka-server-start.sh -daemon config/kraft/server.properties
# 停止服务
bin/kafka-server-stop.sh

脚本简单使用：

# 创建一个主题
bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092
# 查看某个主题参数信息
bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092

# 向主题中写入消息
bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
# 阅读消息
bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092

2.3、Kafka 集群搭建

KRaft 方式搭建集群：

修改配置文件：

# 每一个节点的唯一标识id，不能重复
node.id=0

# 集群中每个 Controller IP地址和端口号
controller.quorum.voters=0@192.168.32.135:9093,1@192.168.32.136:9093,2@192.168.32.137
# 内网监听ip地址
listeners=PLAINTEXT://192.168.32.137:9092,CONTROLLER://192.168.32.137:9093
# 外网监听ip地址
advertised.listeners=PLAINTEXT://192.168.32.137:9092

生成集群唯一 UUID：

# 生成uuid，并把uuid记录下来
./bin/kafka-storage.sh random-uuid
gfCReVjpRqWi3RzL-sg7Lw

# -t 的参数就是生成的唯一集群id，每个节点都要根据这个id去执行命令
./bin/kafka-storage.sh format -t gfCReVjpRqWi3RzL-sg7Lw -c ./config/kraft/server.properties

启动服务：

# 启动 kafka 服务
./bin/kafka-server-start.sh -daemon ./config/kraft/server.properties

3、Kafka 之生产者

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-YZp3ESQp-1692610909242)(https://jx-image-storage.oss-cn-hangzhou.aliyuncs.com/image/image-20230814170630024.png)]

3.1、Java 生产者 API

导入依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.5.1</version>
</dependency>

代码编写：

public static void main(String[] args) throws ExecutionException, InterruptedException {
    Properties properties = new Properties();
    // Kafka服务端的主机名和端口号
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.32.135:9092,192.168.32.136:9092,192.168.32.137:9092");
    // 等待所有副本节点的应答
    properties.put(ProducerConfig.ACKS_CONFIG, "0");
    // 消息发送最大尝试次数，默认一直重试
    properties.put(ProducerConfig.RETRIES_CONFIG, 0);
    // 一批消息处理大小，默认16KB
    properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
    // 请求延时，默认0
    properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
    // 发送缓存区内存大小,默认32MB
    properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
    // key序列化
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    // value序列化
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    // kafka 生产者对象
    KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

    // 构建消息
    ProducerRecord<String, String> message = new ProducerRecord<>("quickstart-events", "key1", "value1");
    /**
         * 有两种消息发送方式：
         * 1：异步方式：send()方法返回一个异步Future对象，
         * 2：回调异步方式：可以在构建参数时设置一个回调方法
         * 3：同步发送：根据send()返回的future对象调用其get()方法进行阻塞主线程
         */
    //        Future<RecordMetadata> send = producer.send(message); // 异步方式
    producer.send(message, new Callback() { // 回调异步
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            System.out.println("消息发送成功");
        }
    });

    // 同步发送，get()方法会阻塞线程，知道上一批数据全部发送成功，返回结果包含了消息主题、分区等信息
    RecordMetadata metadata = producer.send(message).get();
    producer.close();
}

注意：主线程会先将数据发送到缓冲区，然后由 sender 线程进行异发送，而同步发送是一批数据发送到缓冲区由 sender 线程发送到 kafka 集群才会允许下一批数据进行发送。

3.2、Kafka 生产者生产分区

消息分区：将同一个主题的消息数据分区数据到不同的 broker 机器上。

便于合理使用存储资源，每个分区存储在一个 Broker 上，可以将海量数据按照分区分割成一块一块存储在多台 Broker 上。合理控制分区任务，可以实现负载均衡效果。
提高并行度，生产者可以以分区单位进行发送，消费者可以以分区单位进行消费，大大提高数据的处理能力。

分区策略：生产者写入消息到 topic，Kafka 将依据不同的策略将数据分配到不同的分区中。

轮询分区策略：如果生产消息时，对应 key 值是 null，则使用轮询方式最大限度均匀分配到某个分区。
key 分区策略：生产消息时，key 值不为 null，但是没有指定具体分区，则按照 key 的 hash 值去取余你的分区数量确定对应分区。
指定分区：生产消息时，指定对应的分区则严格按照指定分区存储。
自定义分区策略：实现 Partitioner 接口，通过配置可以创建自定义分区策略。

注意：如果发送的分区不存在，则客户端一直会进行等待连接，阻塞线程所有线程。

自定义分区器：可以根据业务需求自定义分区器，实现 Partitioner 接口重写 partition() 方法。

// 自定义分区器
public class CustomerPartitioner implements Partitioner {

    /**
     * 重写对应的分区策略
     * @param topic 主题
     * @param key key 值
     * @param keyBytes 序列化后的key字节值
     * @param value value 值
     * @param valueBytes 序列化后的value值
     * @param cluster 一些集群信息，可以通过主题获取有几个分区
     * @return 数据发送到哪个分区
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        if (key == null){
            return 0;
        }else {
            return 1;
        }
    }

    @Override
    public void close() {
    }

    @Override
    public void configure(Map<String, ?> configs) {
    }
}

// 配置对象配置对应的自定义分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomerPartitioner.class.getName());

3.3、Kafka 生产者常见问题

提高生产者吞吐量：主要是通过配置属性，结合实际生产环境调整配置。

batch.size：批次大小，默认 16 KB，可以根据需要修改对应大小。
linger.ms：缓冲时间，达到这个时间 sender 读取缓冲区数据进行发送，一般 5 ~ 100 ms，如果设置过长数据延迟性就变高。
buffer.memory：缓冲区大小，默认 32 MB，如果分区较多可以设置大一点，如果设置小了就会出现 sender() 数据不足导致等待。
compression.type：数据压缩方式，默认不压缩，可以使用压缩方式有：gzip、snappy、lz4、zstd，常用方式 snappy。

// 一批消息处理大小，默认16KB
properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
// 请求延时，默认0
properties.put(ProducerConfig.LINGER_MS_CONFIG, 5);
// 发送缓存区内存大小,默认32MB
properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
// 设置数据压缩方式
properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

数据可靠性：kafka 生产者在生产数据时有三种 ACK 应答级别，不同应答数据可靠性不一样，默认级别是 -1(all)。

“0”：生产者发送数据不需要等数据落盘直接响应，这样就可能出现数据丢失，但是效率高。
“1”：生产者发送数据只需要 Leader 落盘成功不用等副本复制就直接响应，也可以出现副本数据丢失，如果 Leader 宕机副本成为 Leader 就会出现数据丢失。对于一些数据量大并且允许少量数据丢失。
“-1”、“all”：生产者发送数据只有 Leader 和所有副本都落盘才会应答响应，不会出现数据丢失。==但是可能出现数据重复问题，Leader 和副本都落盘成功，但是没给到响应前 Leader 宕机，生成者由于没有收到应答就会重新给新 Leader 发出数据，但是新 Leader 已经存在这条数据。==用于数据可靠性要求较高。

在同步副本数据时，如果某个副本无法应答 Leader，Leader 也不会应答生产者。但是 Leader 维护了 ISR 一个动态副本队列，如果超过默认 30s 没有副本心跳就会把对应副本剔除队列，这样就不会长期去等待无法同步的副本。

最佳实践方式：（ACK 级别为 -1） + （分区副本 >= 2） + （ISR 中应答最小副本 >= 2）。

// ACK 应答级别
properties.put(ProducerConfig.ACKS_CONFIG, "-1");
// ISR 副本超时时间，默认30分钟
properties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000 * 60 * 30);

数据重复问题：

至少一次：ACK 级别为 -1，副本和 ISR 队列中数量大于等于2，但是存在数据重复问题。
最多一次：ACK 级别为 0，数据不会重复但是可能出现数据丢失问题。
精确一次：对于一些重要数据，数据可靠性高并且不能重复。

幂等性：指生产者向 Broker 发送多少条重复数据，Broker 都只会持久化一条数据。重复数据判断依据：PID(会话ID)、Partition(分区号)、SeqNumber(自增序列号)，三者都不相同则表示不同数据。

PID：每个客户端启动会生成一个 PID，重启会重新生成。(这就导致只能解决单会话内的数据重复问题)
Partition：数据存放的分区位置。
SeqNumber：消息的只增序列号。

/**
 * 开启幂等性：默认开启
 * 开启前提条件：max.in.flight.requests.per.connection等待请求数小于等于5
 *              retries：大于等于0
 *              ACK：-1
 */
properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

生产者事务：开启事务必须先开启幂等性，事务是基于幂等性的。

try {
    // 开启事务
    producer.beginTransaction();
    for (int i=0;i<10;i++){
        ProducerRecord<String, String> message = new ProducerRecord<>("topic-3",UUID.randomUUID().toString(), UUID.randomUUID().toString());
        producer.send(message);
    }
    // 提交事务
    producer.commitTransaction();
}catch (Exception e){
    // 回滚事务，如果数据发送出现异常就会回滚所有发送的数据
    producer.abortTransaction();
    e.printStackTrace();
}finally {
    producer.close();
}

注意：生产者在使用事务前需要指定自定义唯一的 transaction-id，在第一次使用事务会初始化一个 __transaction_state 主题数据，默认有 50 个分区，这里面存放着对事务数据的存放。

数据有序性：

单分区有序性：单分区会根据数据发送的先后顺序进行排序。
多分区有序性：由于是多分区在消费者消费时无法保证取到的数据是有序的，但是可以先把所有数据全部取出来，然后手动进行排序。

单分区有序：由于在 Sender 线程中最多缓存 5 个请求，第一个请求没有应答前可以发送第二个请求，就可能出现第一个请求失败后重试导致数据乱序，但是在 Kafka1.0 之后会缓存生产者发送的最近 5 个请求的元数据，会根据幂等性序列号进行排序然后再进行数据持久化。

/**
 * 保证单分区数据有序的前提条件：(也可以将最大请求数设置为1，只有一个请求缓存)
 */
properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
// sender 线程缓存最大请求数
properties.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

4、Kafka 之 Broker

Zookeeper 模式：在 kafka2.8.0 之前是需要依赖 zookeeper 组件，由 zookeeper 负责集群元数据管理、控制器的选举等。

KRaft 模式：在 KRaft 中，一部分 broker 节点被指定为控制器，这些 Controller 提供 Zookeeper 的共识服务，集群的所有元数据以主题方式存储在 kafka 中。

注意：每一个 Broker 节点既可以充当 Broker，也可以充当 Controller 角色，两者也可以同时充当。

4.1、Broker 节点上下线

新建节点：修改新增节点配置，启动节点服务到集群中。

# -t 的参数就是生成的唯一集群id，每个节点都要根据这个id去执行命令
./bin/kafka-storage.sh format -t gfCReVjpRqWi3RzL-sg7Lw -c ./config/kraft/server.properties
# 启动 kafka 服务
./bin/kafka-server-start.sh -daemon ./config/kraft/server.properties

新建负载均衡计划：即使新增了节点，但是以前数据依然不会对新节点做负载均衡，需要我们自己去对旧数据做负载均衡。

创建负载均衡计划 json 文件：vim topic-to-move.json

{
    "version": 1, // 版本号固定1
    "topics": [ // 需要做负载均衡的主题名称
        {
            "topic": "topic-1"
        },
        {
            "topic": "topic-2"
        },
        {
            "topic": "topic-3"
        }
    ]
}

生成计划：

# --bootstrap-server：连接服务，--topic-to-move-json-file：指定负载均衡计划文件，--broker-list "0,1,2,3"：指定负载均衡的broker的id
./bin/kafka-reassign-partitions.sh --bootstrap-server 192.168.32.135:9092 --topics-to-move-json-file ./json/topic-to-move.json --broker-list "0,1,2,3" --generate

新计划 json 文件：vim increase-replication-factor.json，将生成的计划复制进去，然后执行计划。

# 执行计划命令
./bin/kafka-reassign-partitions.sh --bootstrap-server 192.168.32.135:9092 --reassignment-json-file ./json/increase-replication-factor.json --execute
# 验证是否执行成功
./bin/kafka-reassign-partitions.sh --bootstrap-server 192.168.32.135:9092 --reassignment-json-file ./json/increase-replication-factor.json --verify

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HIqaBd3l-1692610909244)(https://jx-image-storage.oss-cn-hangzhou.aliyuncs.com/image/image-20230816113532281.png)]

节点下线：节点下线只需要将在均衡计划中主题对应的节点去掉需要下线节点 id，然后执行对应计划就可以，然后将节点关机。

4.2、Broker 副本

Kafka 副本：用于提高数据的可靠性，默认副本 1 个，生产环境一般配置 2 个。太多副本会增加磁盘存储空间也会增加网络上数据传输，降低效率。Kafka 中副本分为 Leader 副本和 Follower 副本，但是生产者和消费者都只会去操作 Leader 副本，Follower 副本只是用于存放备份数据。AR = ISR + OSR

AR：Kafka 分区中所有副本。
ISR：与 Leader 保持数据同步的 Follower 副本集合，如果 Follower 默认 30s 未向 Leader 副本同步数据，则会被踢出集合。
OSR：表示 Follower 与 Leader 副本同步超过延迟时间的副本。

Leader 选举：当 Leader 宕机以后，Follower 会根据一定规则选举出新的 Leader，在集群中由某个 Controller 节点用于选举新的 Leader。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-dAYoxwlS-1692610909244)(https://jx-image-storage.oss-cn-hangzhou.aliyuncs.com/image/image-20230816175330559.png)]

Broker 故障：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-RlGuLYGG-1692610909244)(https://jx-image-storage.oss-cn-hangzhou.aliyuncs.com/image/image-20230817093910619.png)]

LEO：每个副本的最后一个 offset，LEO 是每个副本最新的 offset + 1。
HW(高水位)：所有副本中最小的 LEO。

Follower 故障：首先会被踢出 ISR 队列，其它正常的 Broker 继续同步数据。当故障 Follower 重新上线后，它会读取磁盘记录的上次 HW 记录，并将 log 数据高于 HW 的数据截取掉，然后从 HW 部分开始向后继续从 Leader 同步数据，当数据同步到所有副本的 HW 水平就可以重新加入 ISR 队列。

Leader 故障：首先会被踢出 ISR 队列，然后选举出新的 Leader，为保证数据一致性，其它的 Follower 会将高于 HW 的数据裁剪掉，然后和新的 Leader 进行同步。

注意：这只能保证副之间数据一致性，但是不能保证数据不丢失或者不重复(旧 Leader 可能存在还未同步的数据)。

手动调整 Broker 副本：kafka 默认副本是均分分配在每个 Broker 上，可能出现指定副本需求。

创建副本分配文件：vim increase-replication-factor.json，将 topic-1 主题的副本放在 0、1 节点上。

{
    "version":1,
    "partitions":[
        {
            "topic":"topic-1",
            "partition":0,
            "replicas":[
                0,
                1
            ]
        },
        {
            "topic":"topic-1",
            "partition":1,
            "replicas":[
                0,
                1
            ]
        },
        {
            "topic":"topic-1",
            "partition":2,
            "replicas":[
                1,
                0
            ]
        },
        {
            "topic":"topic-1",
            "partition":3,
            "replicas":[
                1,
                0
            ]
        }
    ]
}

执行副本分配计划：

# 执行
./bin/kafka-reassign-partitions.sh --bootstrap-server 192.168.32.135:9092 --reassignment-json-file ./json/increase-replication-factor.json --execute
# 验证
./bin/kafka-reassign-partitions.sh --bootstrap-server 192.168.32.135:9092 --reassignment-json-file ./json/increase-replication-factor.json --verify

Leader Partition 自动平衡：正常情况 kafka 会将分区均匀分配到每一个 Broker 上。当某个 Leader 宕机，新 Leader 可能会集中在其它几台 Broker 上，这可能造成负载不均衡的情况，但是生产中一般会关闭这个功能，因为触发自动平衡很耗性能。

auto.leader.rebalance.enable：是否开启分区自动平衡，默认开启。
leader.imbalance.per.broker.percentage：默认值是10%，broker 中允许 Leader 不平衡比例，如果操作这个比例就会触发自动平衡。
leader.imbalance.check.interval.seconds：默认值 300s，检查 Leader 是否平衡的间隔时间。

增加分区副本：创建副本分配文件：vim increase-replication-factor.json，将 topic-1 主题的副本进行重新规划，然后执行计划。

4.3、Broker 文件存储机制

Broker 数据存储：一个 topic 可以分在多个 partition 上进行存储，一个 topic 下的分区有一个topic名-partition号的 log 文件夹，在这个文件夹下存储着生产者产生的数据，生产者生产的数据会不断追加在 log 文件末尾。

为了防止 log 文件过大导致数据定位效率低下，kafka 采取分片、索引机制。将 log 分片成一个个 Segment，每个Segment 默认大小是 1GB，每个 Segment 由 .index、.log、.timeindex 以及其它文件组成。(文件名称以当前 Segment 的第一条消息的 offset 命名)

.index：作为稀疏索引，每往 log 文件中写入 4KB 数据，就会向 index 文件中添加一条索引。
.log：存放数据文件。
.timeindex：时间戳索引文件，默认 kafka 数据保留7天，会根据这个文件去清除数据。

数据删除策略：kafka 默认数据保留7天，可以设置对应参数修改数据删除时间。

log.retention.hours：单位小时，默认168小时(7天)，优先级最低。
log.retention.minutes：单位分钟，如果设置这个值，小时单位就失效。
log.retention.ms：单位毫秒，如果设置这个值，分钟单位失效。
log.retention.check.interval.ms：设置检查周期，默认5分钟检查数据是否过期。
log.cleanup.policy=delete/compact：设置数据的删除策略。

delete：将过期数据删除。

基于时间(默认)：以 Segment 中所有记录的最大时间戳作为该文件时间戳，到了时间就把整个 Segment 文件删除。
基于大小：当数据超过存储容量，就会删除最早的 Segment 文件。

compact：数据压缩，将相同 key 的数据只保留最后一个版本数据，压缩后的 offset 不是连续的，如果不存在对应 offset 的数据就会拿去下一个 offset 的数据。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-N0W88klu-1692610909246)(https://jx-image-storage.oss-cn-hangzhou.aliyuncs.com/image/image-20230817161834876.png)]

Kafka 高效读写：

kafka 本身是一个分布式集群，并且采用分区技术，对于生产者和消费者在操作数据提高了并行度。
读数据采用稀疏索引，在 .index 文件中存放了数据索引，可以快速定位数据。
采用顺序读写磁盘，在 .log 文件写入数据是追加数据到文件末端，顺序写数据速度快。
kafka 采用也缓存技术和零拷贝技术，kafka 应用层不关心存储的数据，不会对数据进行处理，所以保存数据时 kafka 会把数据交给操作系统的页缓存，再由操作系统完成数据持久化。零拷贝指消费者再消费数据时，先会查看页缓存中是否有数据，如果没有数据操作系统会从磁盘中读取数据到页缓存，然后操作系统直接通过网卡发送给消费者，并没有将数据加载到 kafka 的应用内存中。

5、Kafka 之消费者

消费方式：

pull 拉模式：消费者主动从 broker 拉取数据，kafka 采用该方式可以根据消费者的消费能力自定义数据拉取速度，但是存在 broker 没有数据，导致消费者循环拉取数据为空。
push 推模式：broker 主动向消费者推送消息，但是每个消费者消费速率不一样，可能出现消息来不及处理。

消费者工作流程：

offset：每个消费者对于每个分区都有一个消息偏移量，记录消费者消费到哪个位置了，这个 offset 数据会被持久化到 kafka 的 __consumer_offsets 这个主题中，即使消费者重启，也会从下一个消息进行消费。

5.1、消费者组

消费者组：由多个 consumer 组成，当消费者的 groupId 相同时这些消费者就属于同一个消费者组。

消费者组中的消费者负责消费不同分区数据，一个分区只能由一个组内的一个消费者消费。
消费者组之间相互不干扰，组之间可以消费同一个分区。
如果消费者数量多于分区数量，则多出来的消费者就会闲置。

消费者组初始化：每个 broker 节点都有一个 coordinator 协调器组件，辅助实现消费者组的初始化和分区分配，指定消费者组中消费者应该消费哪个分区。

coordinator 的选择：由消费者组 groupId 决定，（groupId 的 hash 值）% 50，（50 是消费者 __consumer_offsets 主题的分区数），找到 __consumer_offsets 在哪个分区上，就由这个分区上的 coordinator 协调器进行负责消费者组消费。
Consumer Leader 选择：由 coordinator 协调器从消费组中随机选择一个消费者作为 Leader，coordinator 会把消费的 topic 信息发送给Leader，再由 Leader 分配消费任务，具体哪个消费者消费哪个分区，然后将任务发送给 coordinator。
coordinator 分配任务：coordinator 会把 Leader 的分配任务发送给消费组中的所有消费者，按照规则进行消费。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-N5lznpgr-1692610909246)(https://jx-image-storage.oss-cn-hangzhou.aliyuncs.com/image/image-20230818104642229.png)]

注意：所有消费者都和协调器保存 3s 的心跳包，也会有一个连接超时时间，默认 45 s 超过 45s 没有心跳包，那么这个消费者就会被移除，就会触发自动平衡重新分配任务。如果某个消费者某次处理数据时间超过 5 分钟，也会触发自动平衡，将任务交给其它消费者。

消费者消费流程：

fetch.min.bytes：每批次拉取最小大小，默认 1KB，如果不满足这个大小即使有消息也不会拉取。
fetch.max.wait.ms：一批数据未到达超时时间，默认500ms，超过这个时间就会拉取一次数据，即使没有达到最小拉取大小。
fetch.max.bytes：每批次最大抓取大小，默认50MB。
max.poll.records：一次拉取数据返回消息的最大条数，默认 500 条。

5.2、Java 消费者 API

消费者消费一个主题：

public static void main(String[] args) {
    Properties properties = new Properties();
    // 连接集群
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.32.135:9092,192.168.32.136:9092");
    // 设置消费者的消费组id
    properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test2");
    // 设置key和value的反序列化
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
    ArrayList<String> topics = new ArrayList<>();
    topics.add("topic-1");
    // 设置订阅的主题
    consumer.subscribe(topics);

    // 进行拉取数据
    while (true){
        // 间隔多少秒拉取一次数据
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
        // 拉取的数据
        for (ConsumerRecord<String, String> record : records) {
            System.out.println(record);
        }
    }
}

消费某一个分区：

// 消费某个主题下的某个分区
List<TopicPartition> topicPartitions = new ArrayList<>();
// 指定主题和分区
TopicPartition partition = new TopicPartition("topic-1", 0);
topicPartitions.add(partition);
consumer.assign(topicPartitions);

// 执行消费
while (true){
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println(record);
    }
}

5.3、消费者分区分配

消费分区策略：在消费者 Leader 分配消费任务时，会根据对应的分配策略分配任务。kafka 中主要分配策略：Range、RoundRobin、Sticky、CooperativeSticky，默认使用 Range + CooperativeSticky，可以使用组合分配策略。

// 设置消费分区策略
properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RangeAssignor.class.getName() + "," + CooperativeStickyAssignor.class.getName());

Range：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-NM0FNFcW-1692610909247)(https://jx-image-storage.oss-cn-hangzhou.aliyuncs.com/image/image-20230818145123076.png)]

注意：如果消费者组去消费多个主题，就可能存在数据倾斜问题，每个主题多出来的分区就会全部由前面的消费者进行消费。

RoundRobin：轮询消费，所有主题的所有分区和所有消费者进行排序，然后针对分区轮询指定消费者进行消费。

Sticky：粘性分配，尽量均衡分配分区，与 Range 相似，但是不是按照顺序进行分配分区，而是随机将分区分配给消费者。

5.4、消费者 offset 维护

消费者 offset ：表示消费者消费分区已经消费的位置，0.9版本之前，offset 是存放在 Zookeeper 中，o.9 版本之后是存放在 kafka 的 __consumer_offset 这个主题下的。主题的 key：groupId + tpoic + 分区号，value：offset 值，每隔一段时间就会将这个 topic 进行 compact 数据压缩。

自动 offset 维护：kafka 提供了自动提交 offset 功能，每当消费者消费数据，消费者可以自动向 __consumer_offset 主题提交 offset 数据。

enbale.auto.commit：是否开启自动提交 offset 功能，默认是 true。
auto.commit.interval.ms：自动提交 offset 的时间间隔，默认是 5s，单位是毫秒。

// 设置是否自动提交 offset
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, true);
// 设置自动提交的时间
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);

手动提交 offset 维护：自动提交 offset 不能掌握提交的时间，有时候需要手动去提交 offset。

同步提交：需要将最新一批消息提交完成才会继续拉取数据，提交 offset 并且会自动失败重试。
异步提交：处理数据完成后，发出提交 offset 请求后，就继续拉取数据，不会等 offset 提交是是否成功。

// 关闭自动提交
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, false);

// 进行拉取数据
while (true){
    // 间隔多少秒拉取一次数据
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
    // 拉取的数据
    for (ConsumerRecord<String, String> record : records) {
        System.out.println(record);
    }
    // 提交 offset
    consumer.commitSync(); // 同步提交
    consumer.commitAsync(); // 异步提交，可以指定异步提交后的回调方法
}

指定 offset 消费：

earliest：对于同一个消费组，如果从未提交过 offset，自动将偏移量重置为最早偏移量，从头开始消费。但是如果这个消费组提交过 offset，那么效果和 lastest 效果一样。
latest（默认值）：如果没有提交过 offset，只能消费最新的消息，对于历史消息不能消费；如果提交过 offset，那么就从 offset 位置继续消费。
none：如果消费者组从未提交过 offset，那么就向消费者推送错误，如果有就继续按照 offset 消费数据。

指定 offset 进行消费：直接指定 offset 不行，需要等消费者分区完成后再指定 offset 才会生效。

// 设置订阅的主题，设置消费者分区事件
consumer.subscribe(topics, new ConsumerRebalanceListener() {
    // 消费者分区前，例如提交偏移量、释放资源
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {

    }
    // 设置偏移量、初始化资源
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 指定消费的偏移量
        for (TopicPartition partition : partitions) {
            // 手动指定消费者从分区哪个 offset 开始消费
            consumer.seek(partition, 100);
        }
    }
});
// 进行拉取数据
while (true){
    // 间隔多少秒拉取一次数据
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
    // 拉取的数据
    for (ConsumerRecord<String, String> record : records) {
        System.out.println(record);
    }
}

按照时间消费：指定开始消费的时间，可以根据时间去获取 offset 值。

Map<TopicPartition, Long> timeForOffset = new HashMap<>();
// 设置消费者分区事件
consumer.subscribe(topics, new ConsumerRebalanceListener() {
    // 消费者分区前，例如提交偏移量、释放资源
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {

    }
    // 设置偏移量、初始化资源
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 指定消费的偏移量
        for (TopicPartition partition : partitions) {
            // 设置对应消费的时间戳，key：指定分区，value：消费开始位置的时间戳，从当前时间前一天的消息进行消费
            timeForOffset.put(partition, System.currentTimeMillis() - 1000 * 60 * 60 * 24 * 10);
        }
        // 将时间转换成 offset
        Map<TopicPartition, OffsetAndTimestamp> offsets = consumer.offsetsForTimes(timeForOffset);
        // 将转换后的 offset 指定给消费者
        for (TopicPartition partition : partitions) {
            OffsetAndTimestamp offsetAndTimestamp = offsets.get(partition);
            consumer.seek(partition, offsetAndTimestamp.offset());
        }
    }
});

// 进行拉取数据
while (true){
    // 间隔多少秒拉取一次数据
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
    // 拉取的数据
    for (ConsumerRecord<String, String> record : records) {
        System.out.println(record);
    }
}

5.5、消费者常见问题

重复消费：消费者消费消息后但是没有到自动提交时间，这时消费者宕机后面重启后就会从上一次自动提交的位置进行消费，就会出现重复消费。

漏消费：设置为手动提交时，当消费者拉取数据后就手动提交 offset，但是消费者在进行处理数据时出现宕机，并没有正常消费数据，但是已经手动提交了 offset 下一次重启就会跳过没有正常消费的数据。

数据积压：当 kafka 中数据过多，消费者端不能够及时消费，导致数据时间过期会删除数据。例如：kafka 有三天数据需要消费，但是消费者消费这些数据需要4天，有些数据消费不及时就会丢失。

增加 topic 的分区数量，同时增加消费者数量。消费者=分区数，并行消费。
修改 fetch.min.bytes单次拉取大小，提高拉取效率。
修改 max.poll.records 单次最多拉取消息条数，默认 500 条并且对应修改拉取的最大大小。

6、Kafka-Eagle 监控

安装 MySQL 环境。
停止 kafka 集群，并修改 kafka 启动运行内存。

# 修改启动命令
vim ./bin/kafka-server-stop.sh

# 修改对应内存
# 内存参数
export KAFKA_ HEAP_OPTS="-server -Xms2G -Xmx2G --XX:PermSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:ParallelGCThreads=8 -XX:ConcGCThreads=5 -XX:InitiatingHeapOccupancyPercent=70"
# Eagle 监控端口
export JMX_PORT="9999"

下载安装包并解压，官网地址：https://www.kafka-eagle.org/。
配置 Java 环境变量和 EFAK 环境变量。

# java 环境变量
vi /etc/profile
export JAVA_HOME=/usr/java/jdk1.8
export PATH=$PATH:$JAVA_HOME/bin

# EFAK 环境变量
vi /etc/profile
export KE_HOME=/data/soft/new/efak
export PATH=$PATH:$KE_HOME/bin

修改配置，vim ./config/system-config.properties。

# Zookeeper 配置方式
efak.zk.cluster.alias=cluster2
cluster2.zk.list=xdn10:2181,xdn11:2181,xdn12:2181

# 端口
efak.webui.port=8048

######################################
# kafka mysql jdbc driver address
######################################
efak.driver=com.mysql.cj.jdbc.Driver
efak.url=jdbc:mysql://192.168.32.143:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
efak.username=root
efak.password=xxxx

cluster1.efak.offset.storage=kafka

启动 ./bin/ke.sh start，然后通过 ip地址:端口可以直接访问。

注意：kafka-eagle 暂时只支持 kafka 的 Zookeeper 方式，不支持 Kraft 协议的方式。

Docker 安装 kafka-ui：

# 安装命令
docker run -p 9876:8080 \
    --name kafka-ui \
    -e KAFKA_CLUSTERS_0_NAME=kafka-1 \
    -e KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=192.168.32.135:9092,192.168.32.136:9092 \
    -e TZ=Asia/Shanghai \
    -e SERVER_SERVLET_CONTEXT_PATH="/" \
	-e AUTH_TYPE="LOGIN_FORM" \
	-e SPRING_SECURITY_USER_NAME=admin \
	-e SPRING_SECURITY_USER_PASSWORD="admin" \
	-e LANG=C.UTF-8 \
	-d provectuslabs/kafka-ui:latest

7、Spring Boot 整合 Kafka

导入依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
    <version>2.7.13</version>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.8.0</version>
</dependency>

7.1、Kafka 生产者

修改配置：

spring:
  # kafka 相关配置
  kafka:
    bootstrap-servers: 192.168.32.135:9092,192.168.32.136:9092,192.168.32.136:9092
    # 生产者配置
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      retries: 3
      acks: -1
      compression-type: snappy
      buffer-memory: 64MB
      batch-size: 32KB

编写生产者代码：

// 注入kafka
@Resource
private KafkaTemplate<String, String> kafkaTemplate;

@GetMapping("/test1")
public void produce(String msg){
    for (int i=0;i<1000;i++){
        kafkaTemplate.send("topic-boot", UUID.randomUUID().toString() + i, UUID.randomUUID().toString());
    }
}

7.2、Kafka 消费者

修改配置：

spring:
  # kafka 相关配置
  kafka:
    bootstrap-servers: 192.168.32.135:9092,192.168.32.136:9092,192.168.32.137:9092
    # 消费者配置
    consumer:
      group-id: test
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

编写代码：

// 消费者进行消费，id:全局唯一标识
@KafkaListener(id = "consumer1", groupId = "test-1", topics = {"topic-boot","topic-1"})
public void consumer(ConsumerRecord<?, ?> record){
    // msg：接收到的数据
    System.out.println(record);
}