Kafka 使用手册

kafka3.0

1. 什么是kafka？

• Kafka传 统定义：Kafka是一个分布式的基于发布/订阅模式的消息队列（MessageQueue），主要应用于大数据实时处理领域。
• Kafka最 新定义 ： Kafka是 一个开源的 分布式事件流平台 （Event StreamingPlatform），被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用。

在大数据场景主要采用 Kafka 作为消息队列。在 JavaEE 开发中主要采用 ActiveMQ、RabbitMQ、RocketMQ。

1.1 消息队列应用场景

- 缓冲/消峰：有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。
- 解耦：允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束
- 异步通信：允许用户把一个消息放入队列，但并不立即处理它，然后在需要的时候再去处理它们。

1.2 消息队列的两种模式（或者两种模型）

• 点对点模式：消费者主动拉取数据，消息收到后清除消息

• 发布/订阅模式：

  • 可以有多个topic主题（浏览、点赞、收藏、评论等）
  • 消费者消费数据之后，不删除数据
  • 每个消费者相互独立，都可以消费到数据

  ​

2. kafka基础架构

• Producer： 消息生产者，就是向 Kafka broker 发消息的客户端。
• Consumer： 消息消费者，向 Kafka broker 取消息的客户端。
• Consumer Group（CG）： 消费者组，由多个 consumer 组成。消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费；消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
• Broker： 一台 Kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个broker 可以容纳多个 topic
• Topic： 可以理解为一个队列，生产者和消费者面向的都是一个 topic。一个非常大的 topic 可以分布到多个 broker（即服务器）上，
• Partition： 分区。一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列。
• Segment： partition 物理上由多个 segment 组成
• Offset： 每个partition都由一系列有序的、不可变的消息组成，这些消息被连续的追加到partition中。partition中的每个消息都有一个连续的序列号叫做offset,用于partition唯一标识一条消息.
• Replica： 副本。一个 topic 的每个分区都有若干个副本，一个 Leader 和若干个Follower。
• Leader： 每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是 Leader。
• Follower： 每个分区多个副本中的“从”，实时从 Leader 中同步数据，保持和 Leader 数据的同步。Leader 发生故障时，某个 Follower 会成为新的 Leader。

3. kafka集群搭建

1）安装环境准备

上传解压kafka安装包，以下仅演示一台服务器的操作，另外两台也许执行相同步骤

2）修改server.properties

进入到kafka目录，修改配置文件（vim server.properties），重点修改如下三个配置

• broker.id=0（broker 的全局唯一编号，不能重复，只能是数字）
• log.dirs=xx（运行日志(数据)存放的路径）
• zookeeper.connect=10.1.61.121:2181/kafka

#🚩broker 的全局唯一编号，不能重复，只能是数字
broker.id=0
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘 IO 的线程数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#🚩kafka 运行日志(数据)存放的路径，路径不需要提前创建，kafka自动帮你创建，可以配置多个磁盘路径，路径与路径之间可以用"，"分隔
log.dirs=/opt/module/kafka/datas
#topic 在当前 broker 上的分区个数
num.partitions=1
#用来恢复和清理 data 下数据的线程数量
num.recovery.threads.per.data.dir=1
# 每个 topic 创建分区时的副本数，默认时 1 个副本
offsets.topic.replication.factor=1
#segment 文件保留的最长时间，超时将被删除
log.retention.hours=168
#每个 segment 文件的大小，默认最大 1G
log.segment.bytes=1073741824
# 检查过期数据的时间，默认 5 分钟检查一次是否数据过期
log.retention.check.interval.ms=300000
#🚩配置连接 Zookeeper 集群地址（在 zk 根目录下创建/kafka，方便管理）
zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181/ka
fka

注：在其他服务器分别执行上述 操作，修改server.properties中的broker.id=1、broker.id=2（注：broker.id 不得重复，整个集群中唯一）

3）配置环境变量

#在/etc/profile.d/my_env.sh 文件中增加 kafka 环境变量配置
sudo vim /etc/profile.d/my_env.sh


-------------- 增加以下内容 ------------------
#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin
---------------------------------------------


#刷新一下环境变量 
source /etc/profile

4）zookeeper 集群启停脚本

#!/bin/bash

case $1 in
"start"){
	for i in hadoop102 hadoop103 hadoop104
	do
		echo  ------------- zookeeper $i 启动 ------------
		ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh start"
	done
}
;;
"stop"){
	for i in hadoop102 hadoop103 hadoop104
	do
		echo  ------------- zookeeper $i 停止 ------------
		ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh stop"
	done
}
;;
"status"){
	for i in hadoop102 hadoop103 hadoop104
	do
		echo  ------------- zookeeper $i 状态 ------------
		ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh status"
	done
}
;;
esac

5）kafka 集群启停脚本

• Kafka启动命令： ./kafka-server-start.sh -daemon …/config/server.properties
• Kafka关闭命令： ./kafka-server-stop.sh

编写kf启停脚本：

linux查看服务器主机名命令：hostname。需要提前配置所有服务器的免密登录

#! /bin/bash
case $1 in
"start"){
 for i in node1 node2
 do
 echo " --------启动 $i Kafka-------"
 ssh -p 19222 $i "/home/crbt/local/kafka_2.13-2.7.1/bin/kafka-server-start.sh -daemon /home/crbt/local/kafka_2.13-2.7.1/config/server.properties"
 done
};;
"stop"){
 for i in node1 node2
 do
 echo " --------停止 $i Kafka-------"
 ssh -p 19222  $i "/home/crbt/local/kafka_2.13-2.7.1/bin/kafka-server-stop.sh "
 done
};;
esac

分配权限：chmod +x kf.sh

注意：停止 Kafka 集群时，一定要等 Kafka 所有节点进程全部停止后再停止 Zookeeper集群。因为 Zookeeper 集群当中记录着 Kafka 集群相关信息，Zookeeper 集群一旦先停止，Kafka 集群就没有办法再获取停止进程的信息，只能手动杀死 Kafka 进程了。

4. kafka命令行操作

kafka启停命令

• Kafka启动命令：./kafka-server-start.sh -daemon …/config/server.properties
• Kafka关闭命令： ./kafka-server-stop.sh

（注意：启动时配置文件的路径要能够到 server.properties。先启动 Zookeeper 集群，然后启动 Kafka）

主题命令行【topic】

bin/kafka-topics.sh

# 查看指定kafka上的所有主题
./kafka-topics.sh --bootstrap-server localhost:9092,10.1.61.122:9092 --list


# 创建first topic
./kafka-topics.sh --bootstrap-server localhost:9092 --create --partitions 2 --replication-factor 2 --topic first


# 查看主题详情
./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic first
Topic: first    PartitionCount: 1     ReplicationFactor: 2    Configs: segment.bytes=1073741824
Topic: first    Partition: 0    Leader: 2     Replicas: 2,1   Isr: 2,1
        
./kafka-topics.sh --zookeeper 10.1.61.121:2181 --describe --topic first
Topic: first    PartitionCount: 2       ReplicationFactor: 2    Configs:
        Topic: first    Partition: 0    Leader: none    Replicas: 2,1   Isr: 1
        Topic: first    Partition: 1    Leader: none    Replicas: 1,2   Isr: 1

        
# 修改主题分区数（注意：分区数只能增加，不能减少）
./kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic first  --partitions 2
./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic first
Topic: first    PartitionCount: 2       ReplicationFactor: 2    Configs: segment.bytes=1073741824
        Topic: first    Partition: 0    Leader: 2       Replicas: 2,1   Isr: 2,1
        Topic: first    Partition: 1    Leader: 1       Replicas: 1,2   Isr: 1,2
       

# 删除主题
./kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic first

生产者命令行【producer】

kafka-console-producer.sh

# 往topic中写入数据
./kafka-console-producer.sh --bootstrap-server localhost:9092  --topic first
>

消费者命令行【consumer】

kafka-console-consumer.sh

# 消费主题中新生成的消息
./kafka-console-consumer.sh --bootstrap-server localhost:9092  --topic first

# 消费主题中存量消息（从最早的消息开始消费）
./kafka-console-consumer.sh --bootstrap-server localhost:9092  --from-beginning --topic first

5. kafka生产者

生产者消息发送流程

Producer 发送原理

​ 在消息发送的过程中，涉及到了两个线程——main 线程和 Sender 线程。在 main 线程中创建了一个双端队列 RecordAccumulator。main 线程将消息发送给 RecordAccumulator，Sender 线程不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker。

RecordAccumulator 双端队列（缓冲区）有两个核心参数:

• batch.size：只有数据积累到这个值的大小后，sender 线程才会来拉取一波数据（默认大小16k），发送到 Broker。
• linger.ms：数据未达到 batch.size 大小时，等待时间超过 linger.ms 设置的值后也会被发送到 sender 线程中发给 Broker。

Acks:

• 0：生产者发来数据，不需要等待消息落盘直接应答。
• 1：生产者发来数据，leader 收到数据后应答。
• -1（all）生产者发来数据后，leader 和 ISR 队列里所有节点都收到消息后才应答。

retries：发送失败后的重试次数，默认时int类型最大值，建议设置小一，还要设置一个间隔时间。

普通的异步发送

package com.lihw.kafka.producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
public class CustomProducer {

    public static void main(String[] args) {

        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();

        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.1.61.121:9092");

        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                   "org.apache.kafka.common.serialization.StringSerializer");

        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        // 4. 调用 send 方法,发送消息
        for (int i = 0; i < 10; i++) {
            kafkaProducer.send(new ProducerRecord<>("first","lihw " + i));
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

消费first主题

crbt@node2:/home/crbt/local/kafka_2.13-2.7.1/bin>./kafka-console-consumer.sh --bootstrap-server localhost:9092  --topic first
lihw 0
lihw 1
lihw 2
lihw 3
lihw 4

带回调函数的异步发送

​ 回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是元数据信息（RecordMetadata）和异常信息（Exception），如果 Exception 为 null，说明消息发送成功，如果 Exception 不为 null，说明消息发送失败。

package com.lihw.kafka.producer;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class CustomProducerCallback {

    public static void main(String[] args){

        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();

        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.1.61.121:9092,10.1.61.122:9092");

        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                       "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                       "org.apache.kafka.common.serialization.StringSerializer");

        // request.timeout.ms = 3000ms
        properties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,3000);

        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

       try{
           // 4. 调用 send 方法,发送消息
           for (int i = 0; i < 10; i++) {
               kafkaProducer.send(new ProducerRecord<>("first", "lihw " + i), new Callback() {
                   @Override
                   public void onCompletion(RecordMetadata metadata, Exception e) {
                       if (e == null) {
                           // 没有异常,输出信息到控制台
                           System.out.println(" 主题： " +
                                   metadata.topic() + " ->" + " 分区：" + metadata.partition());
                       } else {
                           // 出现异常打印
                           e.printStackTrace();
                       }
                   }
               });
               // 延迟一会会看到数据发往不同分区
            Thread.sleep(2);
           }
       }catch (Exception e){
           e.printStackTrace();
       }finally {
           // 5. 关闭资源
           kafkaProducer.close();
       }
    }
}

 主题： first -> 分区：0
 主题： first -> 分区：0
 主题： first -> 分区：0
 主题： first -> 分区：0
 主题： first -> 分区：0
 主题： first -> 分区：1
 主题： first -> 分区：1
 主题： first -> 分区：1
 主题： first -> 分区：1
 主题： first -> 分区：1

注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。

同步发送API

只需在异步发送的基础上，再调用一下 get()方法即可。

kafkaProducer.send(new ProducerRecord<>("first","kafka" + i)).get();

生产者重要参数列表

生产者分区

• 便于合理使用存储资源，一个Topic可以有多个partition（分区），Partition 可以分布在多个 Broker 上存储，可以把海量的数据按照分区切割成一块一块数据存储在多台 Broker 上。合理控制分区的任务，可以实现负载均衡的效果。
• 提高并行度，生产者可以以分区为单位发送数据；消费者可以以分区为单位进行消费数据。

生产者发送消息的分区策略

• 指定partition的，发送消息时直接发送到指定的分区中
• 未指定partition但有key的，对key进行hash，然后对partition数进行取余，得出该发往哪个分区（key相同的会将消息发往同一分区）
• 没有key，也未指定partition的，采用粘性分区器，选择一个分区开始存储，待该分区的batch已满或者已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）

public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers) {
  ...
}

public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value){
  ...
}

public ProducerRecord(String topic, Integer partition, K key, V value, Iterable<Header> headers) {
  ...
}

public ProducerRecord(String topic, Integer partition, K key, V value) {
  ...
}

/*
 * 没有指明partition值但有key的情况下，将key的hash值与topic的partition数进行取余得到partition值；
 * 例如：key1的hash值=5， key2的hash值=6 ，topic的总partition数=2，那么key1 的value1写入1分区，key2的value2写入0分区。
*/
public ProducerRecord(String topic, K key, V value) {
  ...
}

/**
既没有partition值又没有key值的情况下，Kafka采用Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，待该分区的batch已满或者已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）
例如：第一次随机选择0号分区，等0号分区当前批次满了（默认16k）或者linger.ms设置的时间到， Kafka再随机一个分区进行使用（如果还是0会继续随机）
 */
public ProducerRecord(String topic, V value) {
  ...
}

案例一：指定分区发送消息

package com.lihw.kafka.producer;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class CustomProducerCallbackPartitions {

    public static void main(String[] args) {

        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();

        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.1.61.121:9092,10.1.61.122:9092");

        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        // request.timeout.ms = 3000ms
        properties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,3000);

        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        for (int i = 0; i < 5; i++) {
            // 指定数据发送到 0 号分区，key 为空（IDEA 中 ctrl + p 查看参数）
            // 参数1：主题  参数2：分区  参数3：key  参数4：value
            kafkaProducer.send(new ProducerRecord<>("first",0,"","NBA " + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata,Exception e) {
                    if (e == null){
                        System.out.println(" 主题： " +
                                metadata.topic() + " -> " + "分区：" + metadata.partition()
                        );
                    }else {
                        e.printStackTrace();
                    }
                }
            });
        }
        kafkaProducer.close();
    }
}

 主题： first -> 分区：0
 主题： first -> 分区：0
 主题： first -> 分区：0
 主题： first -> 分区：0
 主题： first -> 分区：0

案例二：自定义分区器

自定义分区器，实现Partitioner接口，重写partition方法，方法返回的是分区数

package com.lihw.kafka.producer;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;

public class MyPartitioner 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) {

        // 获取消息
        String msgValue = value.toString();
        // 创建 partition
        int partition;
        // 判断消息是否包含 lihw
        if (msgValue.contains("lihw")){
            partition = 0;
        }else {
            partition = 1;
        }
        // 返回分区号
        return partition;
    }

    @Override
    public void close() {

    }

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

    }
}

发送消息测试，最终消息全部打在0号分区

package com.lihw.kafka.producer;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class CustomProducerCallbackPartitions {

    public static void main(String[] args) {

        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();

        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.1.61.121:9092,10.1.61.122:9092");

        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        // request.timeout.ms = 3000ms
        properties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,3000);

        // 3.添加自定义分区器🚩
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.lihw.kafka.producer.MyPartitioner");

        // 4. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        for (int i = 0; i < 10; i++) {
            // 指定数据发送到 0 号分区，key 为空,根据自定义分区器进行分区
            kafkaProducer.send(new ProducerRecord<>("first","lihw " + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata,Exception e) {
                    if (e == null){
                        System.out.println(" 主题： " +
                                metadata.topic() + " -> " + "分区：" + metadata.partition()
                        );
                    }else {
                        e.printStackTrace();
                    }
                }
            });
        }
        kafkaProducer.close();
    }
}

结果：消息全部发到了0号分区（121节点）

crbt@node1:/home/crbt/local/kafka_2.13-2.7.1/bin>./kafka-console-consumer.sh --bootstrap-server localhost:9092  --topic first
lihw 0
lihw 1
lihw 2
lihw 3
lihw 4
lihw 5
lihw 6
lihw 7
lihw 8
lihw 9

生产者如何提高吞吐量

• batch.size：批次大小，默认16k
• linger.ms：等待时间，修改为5-100ms
• compression.type：压缩snappy
• RecordAccumulator：缓冲区大小，修改为64m

package com.lihw.kafka.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.Properties;

public class CustomProducerParameters {

    public static void main(String[] args) {
        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.1.61.121:9092,10.1.61.122:9092");

        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 🚩提高吞吐量的参数
        // batch.size：批次大小，默认 16K
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // linger.ms：等待时间，默认 0
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        // RecordAccumulator：缓冲区大小，默认 32M：buffer.memory
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // compression.type：压缩，默认 none，可配置值 gzip、snappy、lz4 和 zstd
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        // 4. 调用 send 方法,发送消息
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("first",1,"","lihewei777 " + i));
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

数据可靠性【ACK】

ACK应答级别

• 0：生产者发送过来的数据，不需要等数据落盘应答。生产者发送过来数据就不管了，可靠性差，效率高；
• 1：生产者发送过来的数据，Leader收到数据后应答。可靠性中等，效率中等；
• -1（all）：生产者发送过来的数据，Leader 和 ISR 队列里面的所有节点收齐数据后应答。可靠性高，效率低；

ISR队列

​ Leader 维护了一个动态的 in-sync replica set（ISR），意为和 Leader 保持同步的 Follower + Leader 集合(leader：0，isr:0,1,2)。如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值由 replica.lag.time.max.ms 参数设定，默认30s。例如2号副本超时，ISR队列则 (leader:0, isr:0,1)。

重复数据分析

acks： -1（all）：生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答。leader 收到消息并落盘后，返回ack之前挂掉，此时会重新选选举出新的 leader，producer 重新发送相同的消息，导致数据重复。

设置ACK操作案例

 // 设置 acks
 properties.put(ProducerConfig.ACKS_CONFIG, "all");

 // 重试次数 retries，默认是 int 最大值，2147483647
 properties.put(ProducerConfig.RETRIES_CONFIG, 3);

package com.lihw.kafka.producer;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class CustomProducerAck {

    public static void main(String[] args) {
        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();

        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.1.61.121:9092,10.1.61.122:9092");

        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        // 3. 设置 acks
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        // 4. 重试次数 retries，默认是 int 最大值，2147483647
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);

        // 4. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        for (int i = 0; i < 10; i++) {
            // 指定数据发送到 0 号分区，key 为空,根据自定义分区器进行分区
            kafkaProducer.send(new ProducerRecord<>("first","lili " + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata,Exception e) {
                    if (e == null){
                        System.out.println(" 主题： " + metadata.topic() + " -> " + "分区：" + metadata.partition()
                        );
                    }else {
                        e.printStackTrace();
                    }
                }
            });
        }
        kafkaProducer.close();
    }
}

数据传递的意义

• 至少一次（At Least Once）= ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2，不能保证数据不重复
• 最多一次（At Most Once）= ACK级别设置为0，不能保证数据不丢失
• 精确一次（Exactly Once）：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。Kafka 0.11版本以后，引入了一项重大特性：幂等性和事务。

Kafka 0.11版本以后，引入了一项重大特性：幂等性和事务。

幂等性

幂等性就是指 Producer 不论向 Broker 发送多少次重复数据，Broker 端都只会持久化一条，保证了不重复。 🚩🚩🚩

如何保证消息可以发送时精确一次（Exactly Once）= 幂等性开启 + 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数量>=2）

重复数据的判断标准：具有 PID, Partition, SeqNumber 相同主键的消息提交时，Broker 只会持久化一条。其中

• PID 是 Kafka 每次重启都会分配一个新的
• Partition 表示分区号
• Sequence Number 是单调自增的

所以幂等性只能保证的是在单分区单会话内不重复。幂等性开启参数 enable.idempotence 默认为 true，false 关闭。

kafka事务

开启事务，必须开启幂等性，Producer 在使用事务功能前，必须先自定义一个唯一的 transactional.id。有了 transactional.id，即使客户端挂掉了，它重启后也能继续处理未完成的事务。

默认有50个分区，每个分区负责一部分事务。事务划分是根据transactional.id的hashcode值%50，计算出该事务属于哪个分区。该分区Leader副本所在的broker节点即为这个transactional.id对应的Transaction Coordinator节点。

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;

单个 Producer，使用事务保证消息的仅一次发送

package com.lihw.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducerTransactions {

    public static void main(String[] args) {
        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();

        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.1.61.121:9092,10.1.61.122:9092");

        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        // 设置事务 id（必须），事务 id 任意起名🚩
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction_id_0");

        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        // 初始化事务
        kafkaProducer.initTransactions();
        // 开启事务
        kafkaProducer.beginTransaction();
      
        try {
            // 4. 调用 send 方法,发送消息
            for (int i = 0; i < 10; i++) {
                // 发送消息
                kafkaProducer.send(new ProducerRecord<>("first","transaction " + i), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception e) {
                        if (e == null){
                            System.out.println(" 主题： " +
                                    metadata.topic() + " -> " + "分区：" + metadata.partition()
                            );
                        }else {
                            e.printStackTrace();
                        }
                    }
                });
            }
            int i = 1 / 0;
            // 提交事务
            kafkaProducer.commitTransaction();
        } catch (Exception e) {
            // 终止事务
            kafkaProducer.abortTransaction();
        } finally {
            // 5. 关闭资源
            kafkaProducer.close();
        }
    }
}

如果发生异常，则终止事务

org.apache.kafka.common.errors.TransactionAbortedException: Failing batch since transaction was aborted
org.apache.kafka.common.errors.TransactionAbortedException: Failing batch since transaction was aborted
org.apache.kafka.common.errors.TransactionAbortedException: Failing batch since transaction was aborted
org.apache.kafka.common.errors.TransactionAbortedException: Failing batch since transaction was aborted
org.apache.kafka.common.errors.TransactionAbortedException: Failing batch since transaction was aborted
org.apache.kafka.common.errors.TransactionAbortedException: Failing batch since transaction was aborted
org.apache.kafka.common.errors.TransactionAbortedException: Failing batch since transaction was aborted
org.apache.kafka.common.errors.TransactionAbortedException: Failing batch since transaction was aborted
org.apache.kafka.common.errors.TransactionAbortedException: Failing batch since transaction was aborted
org.apache.kafka.common.errors.TransactionAbortedException: Failing batch since transaction was aborted

保证消息的有序性

1）kafka在1.x版本之前保证数据单分区有序，条件如下：max.in.flight.requests.per.connection=1（不需要考虑是否开启幂等性）。

2）kafka在1.x及以后版本保证数据单分区有序，条件如下：

• 开启幂等性max.in.flight.requests.per.connection需要设置小于等于5
• 未开启幂等性max.in.flight.requests.per.connection需要设置为1。

原因说明：因为在kafka1.x以后，启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，故无论如何，都可以保证最近5个request的数据都是有序的

6. kafka-broker

zookeeper存储的kafka信息

• brokers/topics/主题/partitions/…
• brokers/ids（当前存活的主机）
• controller（负责选举副本的Leader）

Kafka Broker总体工作流程

• AR：kafka分区中的所有副本统称（AR = ISR + ASR）
• ISR：记录存活的leader和follower的set集合
• ASR：表示 Follower 与 Leader 副本同步时，延迟过多的副本。

Leader选举流程

1. broker启动后在zk中注册（/brokers/ids/ [0,1,2]）
2. controller在zk中注册，谁先注册谁说了算（老大），负责监听brokers节点变化，决定broker中leader的选举，将节点信息上传到zk的ISR中[“lerder:0”,“isr”:“0,1,2”]，其他controller从zk同步相关信息
3. 假设broker1中leader挂了，controller中的老大监听到节点变化，获取ISR，选举新的Leader（在ISR中存活为前提，按照AR顺序排在前面的优先），更新Leader和ISR队列

broker重要参数

服役新节点和负载均衡

kafka/bin/kafka-reassign-partitions.sh

节点服役：

1. 准备新的kafka节点（node0，假设已有node1，node2）
2. 修改server.properties配置文件中的broker.id、log.dirs、zookeeper.connect
3. 重启kafka集群，然后启动新节点node3
4. 负载均衡👇

服役新节点执行负载均衡操作：

1. 创建一个要均衡的主题。

    vim topics-to-move.json
    
    {
    	"topics": [
    		{"topic": "first"}
    	],
    	"version": 1
   }
   

2. 生成一个负载均衡的计划。（kafka-reassign-partitions.sh）

   crbt@eb61119:/home/crbt/local/kafka_2.13-2.7.1/bin>./kafka-reassign-partitions.sh --bootstrap-server localhost:9093 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2," --generate
   
   Current partition replica assignment（当前分区副本分配情况，只使用1，2，没有0节点）
   {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first","partition":1,"replicas":[2,1],"log_dirs":["any","any"]}]}
   
   Proposed partition reassignment configuration（加入了新的节点broker.id=0,建议的分区副本分配情况）
   {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0],"log_dirs":["any","any"]},{"topic":"first","partition":1,"replicas":[0,1],"log_dirs":["any","any"]}]}
   

3. 创建副本存储计划，将上面生产的负载均衡计划粘贴到此json文件中（所有副本存储在 broker0、broker1、broker2中）

   vim increase-replication-factor.json
   
   {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0],"log_dirs":["any","any"]},{"topic":"first","partition":1,"replicas":[0,1],"log_dirs":["any","any"]}]}
   

4. 执行副本存储计划 kafka-reassign-partitions.sh

   ./kafka-reassign-partitions.sh --bootstrap-server localhost:9093 --reassignment-json-file increase-replication-factor.json --execute
   

5. 验证副本存储计划

   ./kafka-reassign-partitions.sh --bootstrap-server localhost:9093 --reassignment-json-file increase-replication-factor.json --verify
   
   or
   
   ./kafka-topics.sh --bootstrap-server localhost:9093 --describe --topic first
   

退役旧节点和负载均衡

先按照退役一台节点，生成执行计划，然后按照服役时操作流程执行负载均衡。

1. 创建一个要均衡的主题。

   vim topics-to-move.json
   
   {
    	"topics": [
    		{"topic": "first"}
    	],
    	"version": 1
   }
   

2. 创建执行计划

   ./kafka-reassign-partitions.sh --bootstrap-server 10.1.61.119:9093 --topics-to-move-json-file topics-to-move.json --broker-list "0,1" --generate
   
   Current partition replica assignment（当前分区副本分配情况，目前数据分布在0，1，2节点上）
   {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first","partition":1,"replicas":[1,0],"log_dirs":["any","any"]},{"topic":"first","partition":2,"replicas":[1,2],"log_dirs":["any","any"]}]}
   
   Proposed partition reassignment configuration（建议的分区副本分配情况，数据分布在0，1节点上）
   {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[1,0],"log_dirs":["any","any"]},{"topic":"first","partition":1,"replicas":[0,1],"log_dirs":["any","any"]},{"topic":"first","partition":2,"replicas":[1,0],"log_dirs":["any","any"]}]}
   
   

3. 创建副本存储计划，将上方生产的负载均衡策略粘贴到此json文件中（所有副本存储在 broker0、broker1中）

   vim increase-replication-factor.json
   
   （建议的分区副本分配情况）
   {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[1,0],"log_dirs":["any","any"]},{"topic":"first","partition":1,"replicas":[0,1],"log_dirs":["any","any"]},{"topic":"first","partition":2,"replicas":[1,0],"log_dirs":["any","any"]}]}
   

4. 执行副本存储计划

   bin/kafka-reassign-partitions.sh --bootstrap-server 10.1.61.119:9093 --reassignment-json-file increase-replication-factor.json --execute
   

5. 验证副本存储计划

   bin/kafka-reassign-partitions.sh --bootstrap-server 10.1.61.119:9093 --reassignment-json-file increase-replication-factor.json --verify
   
   or
   
   ./kafka-topics.sh --bootstrap-server 10.1.61.119:9093 --describe --topic first
   

7. kafka-副本

kafka副本基本信息

• Kafka 副本作用：提高数据可靠性。
• Kafka 默认副本 1 个，生产环境一般配置为 2 个，保证数据可靠性；太多副本会增加磁盘存储空间，增加网络上数据传输，降低效率
• Kafka 中副本分为：Leader 和 Follower。Kafka 生产者只会把数据发往 Leader，然后 Follower 找 Leader 进行同步数据
• Kafka 分区中的所有副本统称为 AR（Assigned Repllicas）
  • AR = ISR + OSR （AR是指为每个分区分配的副本集合，包括leader副本和follower副本）
    • ISR：表示和 Leader 副本保持同步的 Follower 副本的集合。如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值由 replica.lag.time.max.ms参数设定，默认 30s。Leader 发生故障之后，就会从 ISR 中选举新的 Leader。（已同步的Follower副本）
    • OSR：表示 Follower 与 Leader 副本同步时，延迟过多的副本。（掉队的Follower副本）

Leader选举流程

​ Kafka 集群中有一个 broker 的 Controller 会被选举为 Controller Leader，负责管理集群 broker 的上下线，所有 topic 的分区副本分配和 Leader 选举等工作。Controller 的信息同步工作是依赖于 Zookeeper 的。

测试：创建一个新的 topic，3 个分区，3 个副本

# 错误示范：由于我的kafka是三台，所以无法创建多余3的副本数（kafka的副本数量不能大于broker节点数量）
./kafka-topics.sh --bootstrap-server 10.1.61.121:9092 --create --topic lihw --partitions 4 --replication-factor 4
Error while executing topic command : Replication factor: 4 larger than available brokers: 3.[2023-08-24 11:18:29,468] ERROR org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 4 larger than available brokers: 3.

# 创建一个新的 topic，3 个分区，3 个副本
./kafka-topics.sh --bootstrap-server 10.1.61.121:9092 --create --topic lihw --partitions 3 --replication-factor 3
Created topic lihw.

查看主题分区部署情况

./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic lihw

Topic: lihw     PartitionCount: 3       ReplicationFactor: 3    Configs: segment.bytes=1073741824
Topic: lihw     Partition: 0    Leader: 1       Replicas: 1,2,0 Isr: 1,2,0
Topic: lihw     Partition: 1    Leader: 0       Replicas: 0,1,2 Isr: 0,1,2
Topic: lihw     Partition: 2    Leader: 2       Replicas: 2,0,1 Isr: 2,0,1

停掉broker=0的节点，查看分区部署情况

./kafka-server-stop.sh 

./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic lihw
Topic: lihw     PartitionCount: 3       ReplicationFactor: 3    Configs: segment.bytes=1073741824
Topic: lihw     Partition: 0    Leader: 1       Replicas: 1,2,0 Isr: 1,2
Topic: lihw     Partition: 1    Leader: 1       Replicas: 0,1,2 Isr: 1,2
Topic: lihw     Partition: 2    Leader: 2       Replicas: 2,0,1 Isr: 2,1

# 启动broker=0的kafka后重新查看主题分区部署情况
Topic: lihw     PartitionCount: 3       ReplicationFactor: 3    Configs: segment.bytes=1073741824
Topic: lihw     Partition: 0    Leader: 1       Replicas: 1,2,0 Isr: 1,2,0
Topic: lihw     Partition: 1    Leader: 1       Replicas: 0,1,2 Isr: 1,2,0
Topic: lihw     Partition: 2    Leader: 2       Replicas: 2,0,1 Isr: 2,1,0

Follower故障处理流程

• offset：副本中最后一个节点
• LEO：每个副本的最后一个offset，LEO其实就是最新的offset + 1。
• HW：所有副本中最小LEO

Follower故障

1）Follower故障时会被临时提出ISR队列

2）这个期间其他Leader和Follower会继续接受数据

3）待该Follower恢复后，Follower会读取本地磁盘记录上次的HW，将log文件高于HW的部分截取掉，从HW像Leader进行同步。

4）直到达到高水位线时HW，即Follower追上Leader之后，就可以重新加入ISR了。

Leader故障处理流程

1）leader故障时，首先踢出ISR队列

2）重新选举新的leader，Follower会截取掉比HW高的数据重新从Leader中拉取

注意：只能保证数据一致性，不能保证数据不丢失或者不重复

手动调整分区副本存储

在生产环境中，每台服务器的配置和性能不一致，但是Kafka只会根据自己的代码规则创建对应的分区副本，就会导致个别服务器存储压力较大。所有需要手动调整分区副本的存储。

需求：创建一个新的topic，4个分区，两个副本，名称为three。将该topic的所有副本都存储到broker0和broker1两台服务器上。

创建一个新的 topic，名称为 three。

./kafka-topics.sh --bootstrap-server localhost:9092 --create --partitions 3 --replication-factor 2 --topic three

查看分区副本存储情况。

./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic three

Topic: three    PartitionCount: 3       ReplicationFactor: 2    Configs: segment.bytes=1073741824
Topic: three    Partition: 0    Leader: 0       Replicas: 0,1   Isr: 0,1
Topic: three    Partition: 1    Leader: 2       Replicas: 2,0   Isr: 2,0
Topic: three    Partition: 2    Leader: 1       Replicas: 1,2   Isr: 1,2

🚩创建副本存储计划，将所有副本都指定存储在 broker0中【与服役新节点，退役旧节点类似，手动分配分区和副本的存储情况】

vim increase-replication-factor.json

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

执行副本存储计划。

./kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file increase-replication-factor.json --execute

Current partition replica assignment
{"version":1,"partitions":[
	{"topic":"three","partition":0,"replicas":[0],"log_dirs":["any","any"]},
	{"topic":"three","partition":1,"replicas":[0],"log_dirs":["any","any"]},
	{"topic":"three","partition":2,"replicas":[0],"log_dirs":["any","any"]}
]}

Save this to use as the --reassignment-json-file option during rollback
Successfully started partition reassignments for three-0,three-1,three-2

验证副本存储计划

./kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file increase-replication-factor.json --verify

./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic three

Topic: three    PartitionCount: 3       ReplicationFactor: 1    Configs: segment.bytes=1073741824
Topic: three    Partition: 0    Leader: 0       Replicas: 0     Isr: 0
Topic: three    Partition: 1    Leader: 0       Replicas: 0     Isr: 0
Topic: three    Partition: 2    Leader: 0       Replicas: 0     Isr: 0

增加副本因子

在生产环境当中，由于某个主题的重要等级需要提升，我们考虑增加副本。副本数的增加需要先制定计划，然后根据计划执行。

注意：kafka中不能通过命令行的方式增加副本（可以通过命令行的方式增加分区数，只能增加不能减少）

#创建副本存储计划（所有副本都指定存储在 broker0、broker1、broker2 中）。
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]}
]}


#执行副本存储计划
./kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute

kafka文件存储机制

​ Topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是Producer生产的数据。Producer生产的数据会被不断追加到该log文件末端，为防止log文件过大导致数据定位效率低下（不需要先查出来数据，直接往最后追加，也是kafka可以高效读写的原因之一），Kafka采取了分片和索引机制，将每个partition分为多个segment。（segment默认大小为1GB）每个segment包括：“.index”文件、“.log”文件和.timeindex等文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号，例如：first-0。

Topic 数据存储在server.properties配置文件中设置的logs的位置

#kafka 运行日志(数据)存放的路径，路径不需要提前创建，kafka自动帮你创建，可以配置多个磁盘路径，路径与路径之间可以用"，"分隔
log.dirs=/opt/module/kafka/datas

segment 中 index 文件和 log 文件详解

思考四个问题：

1.topic中partition存储分布：

• 在Kafka文件存储中，同一个topic下有多个不同partition，每个partition为一个目录，partiton命名规则为topic名称+有序序号，第一个partiton序号从0开始，序号最大值为partitions数量减1。（例如：first-0、first-1、first-2）
• 每个partition下面有多个segment。

2.partiton中文件存储方式：

• 每个partition(目录)相当于一个巨型文件被平均分配到多个大小相等segment(段)数据文件中。但每个段segment file消息数量不一定相等，这种特性方便old segment file快速被删除。
• 每个partition只需要支持顺序读写就行了，segment文件生命周期由服务端配置参数决定。

3.partiton中segment文件存储结构：

• segment file由 segment索引文件、数据文件两部分组成，这两个文件一一对应，后缀是”.index”和“.log”，分别表示为segment索引文件、数据文件。
• segment文件命名规则：partition全局的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset值 +1。数值最大为64位long大小，19位数字字符长度，没有数字用0填充。

4.在partition中如何通过offset查找message:

• segment的索引文件命令规则：起始偏移量(offset)为0.后续每个segment文件名为上一个segment文件最后一条消息的offset值，所以，第二个文件00000000000000000522.index的文件名是上一个log中最大偏移量+1（521+1=522），其他后续文件依次类推，只要根据offset 二分查找 文件列表，就可以快速定位到具体文件。 当offset=600时定位到00000000000000000522.index|log，用index文件名上的数字+相对offset计算log文件中数据存在的位置，522+65=587，522+117=639，587 < 600 < 639，所以Offset=600的数据在position=6410的位置往下顺扫。
• segment index file采取稀疏索引存储方式，不会为每条数据创建索引，大大的减少索了引文件大小。

在partition中如何通过offset查找message 总结：

1. 根据目标Offset定位Segment文件
2. 找到小于等于目标Offset的最大Offset对应的索引项
3. 根据索引定位到log文件
4. log文件中从当前索引位置往后顺扫，找到对应的记录

kafka日志存储参数配置

日志存储配置（删除/压缩）

• **删除：**如果一个 segment 中有一部分数据过期，一部分没有过期，未消费完的segment会等最大offset消费完成后在删除

• **压缩：**compact 日志压缩：compact日志压缩：对于相同key的不同value值，只保留最后一个版本。

注意：这种策略只适合特殊场景，比如消息的key是用户ID，value是用户的资料，通过这种压缩策略，整个消息集里就保存了所有用户最新的资料。

kafka为啥可以高效读写数据

为啥kafka快啊？

1）Kafka 本身是分布式集群，采用分区技术，并行度高，还可以添加Broker来扩展集群的处理能力。

2）读数据采用稀疏索引，可以快速定位要消费的数据。

3）批量处理：Kafka生产者和消费者都支持批处理操作，减少了网络通信的开销。

4）顺序写磁盘：producer 生产数据写入到 log 文件中，写的过程是一直追加到文件末端，为顺序写，免了随机写入的开销。另外还支持数据分区与数据压缩等技术，进一步提高了磁盘存储的效率。

5）页缓存 + 零拷贝技术：Kafka的Broker结合了页缓存和零拷贝技术，将要读取或写入的数据缓存到PageCache中，避免了频繁的磁盘IO操作，同时采用零拷贝技术，避免了数据在用户空间和内核空间之间的多次内存拷贝。

• PageCache页缓存：Kafka重度依赖底层操作系统提供的PageCache功能。当上层有读写操作时，操作系统会优先从PageCache中进行读写，如果找不到再去磁盘中读取。实际上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来使用。
• 零拷贝：Kafka的数据加工处理操作交由Kafka生产者和Kafka消费者处理。Kafka Broker应用层不关心存储的数据，所以就不用走应用层，传输效率高。传统上，数据从应用程序缓冲区传输到网络或磁盘之前，需要进行多次内存拷贝。采用零拷贝后，Kafka可以直接从页缓存中读取或写入数据，避免了数据在用户空间和内核空间之间的IO操作，减少了CPU和内存的开销。

8. kafak消费者

消费者总体工作流程

• kafka消费方式： consumer采用从broker中主动拉取的方式去消费数据（pull）。为什么不是broker主动推送呢（push），是因为由broker决定发送速率很难适应所有消费者。pull模式的缺点是kafka没有数据时消费者可能会陷入循环中，一直返回空数据。

• 消费者组： Consumer Group（CG），消费者组，由多个consumer组成，所有消费者的groupid相同。消费者组内每个消费者消费不同的分区，一个分区只能由一个组内的一个消费者消费。消费者组之间互不影响。

• offset： 每个消费者的offset由消费者提交到系统主题(Topic)保存（__consumer_offsets，该主题默认有50个分区）

  

消费者组的初始化流程

• coordinator： 辅助实现消费者组的初始化和分区的分配。

• coordinator节点的选择： groupid的hashcode值 % 50（ consumer_offsets的分区数量）的值就是系统主题consumer_offsets主题所在的分区号。选择该broker上的coordinator作为这个消费者组的老大。消费者组下的所有的消费者提交offset的时候就往这个分区去提交offset。

• 消费者组初始化流程：

  

  1. groupid的hashcode值 % 50 = 分区号，选择该分区所在的broker上的coordinator作为消费者的老大。
  2. 每个consumer都发送JoinGroup请求到coordinator，coordinator选出一个consumer作为leader，把要消费的topic情况发送给leader消费者。
  3. leader负责制定消费方案，把消费方案发送给coordinator，coordinator把消费方案下发给各个consumer。
  4. 消费者和coordinator的心跳检测默认3s，一旦超过（session.timeout.ms=45s）则该消费者会被移除。或者消费者处理消息的时间过长于（max.poll.interval.ms=5min）都会触发再平衡。

消费者重要参数

消费者API

订阅主题

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

package com.lihw.kafka.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;

public class CustomConsumer {

    public static void main(String[] args) {
        // 1.创建消费者的配置对象
        Properties properties = new Properties();
        // 2.给消费者配置对象添加参数
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.1.61.121:9092,10.1.61.122: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");

        // 创建消费者对象
        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);
            }
        }
    }
}

订阅分区

创建一个独立消费者，消费 first 主题 0 号分区的数据。

package com.lihw.kafka.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.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

public class CustomConsumerPartition {

    public static void main(String[] args) {
        // 1.创建消费者的配置对象
        Properties properties = new Properties();
        // 2.给消费者配置对象添加参数
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.1.61.121:9092,10.1.61.122: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");

        // 创建消费者对象
        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) {
            // 设置 1s 消费一批数据
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
            // 打印消费到的数据
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}

消费者组案例

测试同一个主题的分区数据，生产者发送到三个分区的消费情况：

• 如果只有一个消费者，会消费三个分区的数据
• 如果有两个消费者，其中一个consumer会消费两个分区
• 三个消费者，一人一个分区进行消费

分区的分配以及再平衡

1. 一个consumer group中有多个consumer组成，一个 topic有多个partition组成，现在的问题是，到底由哪个consumer来消费哪个partition的数据。
2. Kafka有四种主流的分区分配策略： Range、RoundRobin、Sticky、CooperativeSticky。
3. 可以通过配置参数partition.assignment.strategy，修改分区的分配策略。默认策略是Range + CooperativeSticky。Kafka可以同时使用多个分区分配策略。

Range

分区数 / 消费者数，进行平均分，多余的分给第一个consumer。

• **Range分区分配策略：**分区数 / 消费者数 = 每个消费者应该消费几个分区，如果除不尽，前面的消费者会多消费一个分区。

  例如：7 / 3 = 2 余 1，那么每个消费者消费两个分区，多余的一个给第一个消费者。8 / 3 = 2 余 2，每个小给这消费2个分区，多余的两个分给前两个消费者

• **Range再平衡案例：**消费者组需要按照超时时间 45s 来判断它是否退出，所以需要等待，时间到了 45s 后，判断它真的退出就会把（1）任务分配给其他 broker 执行。（挂掉的broker全部打到某一个消费者上面）

  0消费者：0 、1、2

  1消费者：3、4

  2消费者：5、6

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

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

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

注意：该分区策略在消费的主题多的情况下，容易造成数据倾斜

RoundRobin（轮询）

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

• RoundRobin再平衡： 45s后宕掉的消费者的任务会按照 ’轮询‘ 的方式，把数据轮询分给其他消费者进行消费。

• 修改分区策略：

  // 修改分区分配策略
  properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");
  

Sticky（粘性）

• Sticky分区策略： 可以理解为分配的结果带有“粘性的”。即在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可以节省大量的开销。粘性分区是 Kafka 从 0.11.x 版本开始引入这种分配策略，首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。

• Sticky再平衡： 0 号消费者的任务会按照粘性规则，尽可能均衡的随机分配，交由其他消费者消费。45s以后会重新分配

• 修改分区策略：

  // 修改分区分配策略
  ArrayList<String> startegys = new ArrayList<>();
  startegys.add("org.apache.kafka.clients.consumer.StickyAssignor");
  properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, startegys);
  

offset偏移量

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

自动提交 offset

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

自动提交offset的相关参数：

• enable.auto.commit：是否开启自动提交offset功能，默认是true
• auto.commit.interval.ms：自动提交offset的时间间隔，默认是5s

// 是否自动提交 offset
 properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
 // 提交 offset 的时间周期 1000ms，默认 5s
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);

手动提交 offset

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

手动提交offset的方法有两种：

• commitSync（同步提交）：必须等待offset提交完毕，再去消费下一批数据。
• commitAsync（异步提交） ：发送完提交offset请求后，就开始消费下一批数据了。

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

同步提交offset： 由于同步提交 offset 有失败重试机制，故更加可靠，但是由于一直等待提交结果，提交的效率比较低

public class CustomConsumerByHandSync {
   public static void main(String[] args) {
   // 1. 创建 kafka 消费者配置类
   Properties properties = new Properties();
   // 2. 添加配置参数
   // 添加连接
   properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   // 配置序列化 必须
   properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                  "org.apache.kafka.common.serialization.StringDeserializer");
   properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                   "org.apache.kafka.common.serialization.StringDeserializer");
     
   // 配置消费者组 
   properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
     
   // 是否自动提交 offset
   properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
     
   //3. 创建 kafka 消费者
   KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
     
   //4. 设置消费主题 形参是列表
   consumer.subscribe(Arrays.asList("first"));
     
     //5. 消费数据
	 while (true){
 		// 读取消息
 	    ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
       // 输出消息
 			for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
 				System.out.println(consumerRecord.value());
 			}
 			// 同步提交 offset
 			consumer.commitSync();
 	}
     
  }
}   

异步提交offset： 虽然同步提交 offset 更可靠一些，但是由于其会阻塞当前线程，直到提交成功。因此吞吐量会受到很大的影响

public class CustomConsumerByHandSync {
   public static void main(String[] args) {
   // 1. 创建 kafka 消费者配置类
   Properties properties = new Properties();
   // 2. 添加配置参数
   // 添加连接
   properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   // 配置序列化 必须
   properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                  "org.apache.kafka.common.serialization.StringDeserializer");
   properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                   "org.apache.kafka.common.serialization.StringDeserializer");
     
   // 配置消费者组 
   properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
     
   // 是否自动提交 offset
   properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
     
   //3. 创建 kafka 消费者
   KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
     
   //4. 设置消费主题 形参是列表
   consumer.subscribe(Arrays.asList("first"));
     
     //5. 消费数据
	 while (true){
 		// 读取消息
 	    ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
       // 输出消息
 			for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
 				System.out.println(consumerRecord.value());
 			}
 			// 异步提交 offset
 			consumer.commitAsync();
 	}
     
  }
}   

指定 Offset 消费

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

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

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

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

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

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

任意指定 offset 位移开始消费

public class CustomConsumerByHandSync {
   public static void main(String[] args) {
   // 1. 创建 kafka 消费者配置类
   Properties properties = new Properties();
   // 2. 添加配置参数
   // 添加连接
   properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   // 配置序列化 必须
   properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                  "org.apache.kafka.common.serialization.StringDeserializer");
   properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                   "org.apache.kafka.common.serialization.StringDeserializer");
     
   // 配置消费者组 
   properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
     
   // 是否自动提交 offset
   properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
     
   //3. 创建 kafka 消费者
   KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
     
   // 2 订阅一个主题
   ArrayList<String> topics = new ArrayList<>();
   topics.add("first");
   kafkaConsumer.subscribe(topics);
   
   //🚩 获取消费者分区信息，并指定offset进行消费
   Set<TopicPartition> assignment= new HashSet<>();
 	  while (assignment.size() == 0) {
 		 kafkaConsumer.poll(Duration.ofSeconds(1));
 		 // 获取消费者分区分配信息（有了分区分配信息才能开始消费）
 		 assignment = kafkaConsumer.assignment();
       }
 	    
      // 遍历所有分区，并指定 offset 从 1700 的位置开始消费
	  for (TopicPartition tp: assignment) {
	  	kafkaConsumer.seek(tp, 1700);
	  }
     
     // 消费数据
	 while (true){
 		// 读取消息
 	    ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
       		// 输出消息
 			for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
 				System.out.println(consumerRecord.value());
 			}
 	}
     
  }
}   

指定时间进行消费

在生产环境中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。例如要求按照时间消费前一天的数据，怎么处理？

public class CustomConsumerByHandSync {
   public static void main(String[] args) {
   // 1. 创建 kafka 消费者配置类
   Properties properties = new Properties();
   // 2. 添加配置参数
   // 添加连接
   properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   // 配置序列化 必须
   properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                  "org.apache.kafka.common.serialization.StringDeserializer");
   properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                   "org.apache.kafka.common.serialization.StringDeserializer");
     
   // 配置消费者组 
   properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
     
   // 是否自动提交 offset
   properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
     
   //3. 创建 kafka 消费者
   KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
     
   // 2 订阅一个主题
   ArrayList<String> topics = new ArrayList<>();
   topics.add("first");
   kafkaConsumer.subscribe(topics);
   
   //🚩 获取消费者分区信息，并指定offset进行消费
   Set<TopicPartition> assignment= new HashSet<>();
   while (assignment.size() == 0) {
 	  kafkaConsumer.poll(Duration.ofSeconds(1));
 	  // 获取消费者分区分配信息（有了分区分配信息才能开始消费）
 	  assignment = kafkaConsumer.assignment();
   }
     
   HashMap<TopicPartition, Long> timestampToSearch = new HashMap<>();	    
   // 封装集合存储，每个分区对应一天前的数据
   for (TopicPartition topicPartition : assignment) {
       timestampToSearch.put(topicPartition, System.currentTimeMillis() - 1 * 24 * 3600 * 1000);
   }
    // 获取从 1 天前开始消费的每个分区的 offset
 	Map<TopicPartition, OffsetAndTimestamp> offsets = kafkaConsumer.offsetsForTimes(timestampToSearch);
    // 遍历每个分区，对每个分区设置消费时间。
 	for (TopicPartition topicPartition : assignment) {
 		OffsetAndTimestamp offsetAndTimestamp = offsets.get(topicPartition);
 		// 根据时间指定开始消费的位置
 		if (offsetAndTimestamp != null){
 			kafkaConsumer.seek(topicPartition, offsetAndTimestamp.offset());
 		}
 	}
     
     // 消费数据
	 while (true){
 		// 读取消息
 	    ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
       		// 输出消息
 			for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
 				System.out.println(consumerRecord.value());
 			}
 	}
     
  }
}   

漏消费和重复消费

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

  场景1：重复消费。自动提交offset引起。默认5s一次

  

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

  场景2：消费者还没落盘，导致消费者漏消费

  

消费者事务

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

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

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

2）如果是下游的数据处理不及时：提高每批次拉取的数量。批次拉取数据过少（拉取数据/处理时间 < 生产速度），使处理的数据小于生产的数据，也会造成数据积压。

• fetch.max.bytes：消费者一次 poll 抓取一批消息的最大的字节数。默认 Default: 52428800（50 m）
• max.poll.records：消费者一次 poll 抓取数据的最大条数，默认 500 条