kafka3.0

1. 什么是kafka？

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

在大数据场景主要采用 Kafka 作为消息队列。在 JavaEE 开发中主要采用 ActiveMQ、RabbitMQ、RocketMQ。
kafka官网地址：http://kafka.apache.org/downloads.html

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集群搭建

安装环境准备

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

修改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 不得重复，整个集群中唯一）

配置环境变量

#在/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

zk集群启停脚本

#!/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

kafka启停脚本

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生产者

生产者消息发送流程

发送原理

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

普通的异步发送

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();

生产者重要参数列表

生产者分区

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

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

• 指定partition的，发送消息时直接发送到指定的分区中
• 未指定partition但有key的，对key进行hash，然后对partition数进行取余，得出该发往哪个分区
• 没有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++) {
            // 指定数据发送到 1 号分区，key 为空（IDEA 中 ctrl + p 查看参数）
            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;
        // 判断消息是否包含 atguigu
        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超时，(leader:0, isr:0,1)。

重复数据分析

acks： -1（all）：生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答。

leader收到消息并落盘后，返回adc之前挂掉，此时会重新选选举出新的leader，producer重新发送相同的消息，导致数据重复。

代码配置

 // 设置 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（node3）节点
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）
   {"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. 创建副本存储计划（所有副本存储在 broker0、broker1、broker2、broker3 中）

   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）执行负载均衡操作

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

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（当前分区副本分配情况）
   {"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（建议的分区副本分配情况）
   {"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. 创建副本存储计划（所有副本存储在 broker0、broker1、broker2 中）

   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副本基本信息

1）kafka 副本作用：提高数据可靠性。

2）Kafka 默认副本 1 个，生产环境一般配置为 2 个，保证数据可靠性；太多副本会增加磁盘存储空间，增加网络上数据传输，降低效率

3）Kafka 中副本分为：Leader 和 Follower。Kafka 生产者只会把数据发往 Leader，然后 Follower 找 Leader 进行同步数据

4）Kafka 分区中的所有副本统称为 AR（Assigned Repllicas）

AR = ISR + OSR

• ISR：表示和 Leader 保持同步的 Follower 集合。如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值由 replica.lag.time.max.ms参数设定，默认 30s。Leader 发生故障之后，就会从 ISR 中选举新的 Leader。
• OSR：表示 Follower 与 Leader 副本同步时，延迟过多的副本。

Leader选举流程【测试】

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

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

./kafka-topics.sh --bootstrap-server 10.1.61.121:9092 --create --topic lihw --partitions 4 --replication-factor 4
# 由于我的kafka是三台，所以无法创建多余3的副本数（kafka的副本数量不能大于broker节点数量）
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.


./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

故障处理流程

• 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,1],"log_dirs":["any","any"]},{"topic":"three","partition":1,"replicas":[0,1],"log_dirs":["any","any"]},{"topic":"three","partition":2,"replicas":[1,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的位置

index 文件和 log 文件详解

思考四个问题：

1. topic中partition存储分布
2. partiton中文件存储方式
3. partiton中segment文件存储结构
4. 在partition中如何通过offset查找message

1.topic中partition存储分布：

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

2.partiton中文件存储方式：

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

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

• segment file由segment索引文件、数据文件2部分组成，这两个文件一一对应，后缀是”.index”和“.log”，分别表示为segment索引文件、数据文件
• segment文件命名规则：partion全局的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset值。数值最大为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采取稀疏索引存储方式，不会为每条数据创建索引，大大的减少索了引文件大小。

kafka日志存储参数配置

compact日志删除/压缩

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

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

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

kafka为啥可以高效读写数据

为啥kafka快啊？

1）Kafka 本身是分布式集群，可以采用分区技术，并行度高

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

3）顺序写磁盘：Kafka 的 producer 生产数据，要写入到 log 文件中，写的过程是一直追加到文件末端，为顺序写。

4）页缓存 + 零拷贝技术：

• 零拷贝：Kafka的数据加工处理操作交由Kafka生产者和Kafka消费者处理。Kafka Broker应用层不关心存储的数据，所以就不用走应用层，传输效率高。
• PageCache页缓存：Kafka重度依赖底层操作系统提供的PageCache功能。当上层有写操作时，操作系统只是将数据写PageCache。当读操作发生时，先从PageCache中查找，如果找不到，再去磁盘中读取。实际上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来使用。

8. kafak消费者

消费者总体工作流程

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

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

  

• offset：每个消费者的offset由消费者提交到系统主题保存（__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

• 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, 1000);

手动提交 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：消费者获取服务器端一批消息最大的字节数。默认 Default: 52428800（50 m）
• max.poll.records：一次 poll 拉取数据返回消息的最大条数，默认是 500 条

本栏目推荐文章
• 第二章 Spring Boot 整合 Kafka消息队列 生产者
• ClickHouse(21)ClickHouse集成Kafka表引擎详细解析
• org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
• ogg-kafka
• kafka 服务开机自启动
• Kafka - 面试题
• Kafka 万字精讲｜工作五年这些你都知道吗？
• kafka入门(七)：kafka实现高吞吐量
• 关于函数式接口中常用的Supplier、Consumer、predicate、Function的总结以及其使用场景
• kafka入门(五)：日志分段（LogSegment）

producer consumer Kafka3 broker Kafkaproducer consumer kafka3 broker consumer消费者kafka3 kafka 生产者producer kafka3 kafka producer-consumer producers consumers model and kafka3 partition consumer数量broker kafka3 kafka 分布式producer-consumer producer consumer windows kafka3 kafka cmak