花生_TL007

这个屌丝很懒，什么也没留下！

热门标签

kafka框架从入门到精通非常全_卡夫卡框架

作者：花生_TL007 | 2024-04-07 09:23:37

踩

卡夫卡框架

1、kafka定义

kafka传统定义：kafka是一个分布式的基于发布/订阅模式的消息队列，主要应用于大数据实时处理场景

Kafka最新定义：主要倾向于分布式事件流平台，被数千家公司用于高性能数据管道，流分析，数据集成和关键人物应用

1.1 消息队列

目前企业中比较常见的消息队列产品主要有Kafka、ActiveMQ、RabbitMQ、RocketMQ等。

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

1.1.1 应用场景

解耦：比如订单对接库存系统，订单只要消息发往消息队列，等待库存系统消费就可以了
异步：重要业务同步完成返回，将一些非必要耗时业务发给消息队列异步处理
削峰（缓存）：秒杀活动，一般会因为流量过大，导致流量暴增，应用挂掉。

1.1.2 消息队列的两个模式

点对点：
- 消费者主动拉取数据，消费者收到消息后删除消息
发布/订阅模式
- 可以有多个topic,根据业务场景不通，可以有多个topic
- 消费者消费数据后，不删除（kafka默认保留7天）
- 每个消费者相互独立，都可以消费到数据

1.3 kafka基础架构

为方便扩展，提高吞吐量，一个topic可以有多个partion
配合分区的设计，提出消费者组的概念，组内的每个消费者并行消费
为提高可用性，每个partion有若干个副本
zk中记录谁是leader,kafka2.8.0以后可以配置不使用zk
1. Early access of replace ZooKeeper with a self-managed quorum----来自官网

图解：

Producer :生产者，就是像kafka Broker发消息的客户端
Consumer:生产者，向kafka broker取消息的客户端
Consumer group:消费者组，由多个consumer组成。消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费；消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
Broker ：一台Kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。
topic：可以理解为一个队列，生产者和消费者面向的都是一个 topic。
Partition：为了实现扩展性，一个非常大的topic可以分布到多个broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。
Replica：副本。一个topic的每个分区都有若干个副本，一个Leader和若干个Follower。
Leader：每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是Leader。
Follower：每个分区多个副本中的“从”，实时从Leader中同步数据，保持和Leader数据的同步。Leader发生故障时，某个Follower会成为新的Leader。

2、安装

2.1 安装部署

2.1.1 集群规划

node1	node2	node3
zk	zk	zk
kafka	kafka	kafka

2.1.2 集群部署

官方网站下载地址 http://kafka.apache.org/downloads.htm
解压缩包 tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/soft/
修改文件 mv kafka_2.12-3.0.0/ kafka

进入到/opt/soft/kafka/config目录，修改配置文件 vim server.properties

#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=node1:2181,node2:2181,node3:2181/kafka

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

分发安装包

scp -r /opt/soft/kafka root@node2:/opt/soft
scp -r /opt/soft/kafka root@node3:/opt/soft
1
2

分别在node2,node3中修改第四部的server.properties，将broker.id改为1 ，2broker.id不得重复，整个集群中唯一。

启动集群

先启动Zookeeper集群，然后启动Kafka。

启动集群 每个节点都要执行
bin/kafka-server-start.sh -daemon config/server.properties
关闭集群 -每个节点都要执行
bin/kafka-server-stop.sh 
1
2
3
4

集群启动脚本

vim zkStart.sh
1

#/bin/bash
case $1 in 
"start")
	for i in node1 node2 node3
	do 
	echo "$i zk正在启动..."
	ssh $i "/opt/soft/zookeeper/bin/zkServer.sh start"
	done
;;
"stop")
	for i in node1 node2 node3
	do 
	echo "$i 正在关闭...."
	ssh $i "/opt/soft/zookeeper/bin/zkServer.sh stop"
	done

;;
esac
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

vim kafka.sh
1

#/bin/bash
case $1 in 
"start")
	for i in node1 node2 node3 
	do 
	 echo "$i start kafka"
	 ssh $i "/opt/soft/kafka/bin/kafka-server-start.sh -daemon /opt/soft/kafka/config/server.properties"
	done
;;
"stop")
        for i in node1 node2 node3
        do
         echo "$i stop kafka"
         ssh $i "/opt/soft/kafka/bin/kafka-server-stop.sh"
        done

;;
esac
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

添加执行权限

chmod +x kafka.sh  或者 chmod 777 kafka.sh
1

集群启停命令

 #启动
zkStart.sh start
kafka.sh start
#停止 先停kafka 再停止 zk
kafka.sh stop
zkStart.sh stop

1
2
3
4
5
6
7

2.2 命令行操作

2.2.1 主题命令行操作

主题命令行参数

bin/kafka-topics.sh
1

参数	描述
–bootstrap-server <String: server toconnect to>	连接的Kafka Broker主机名称和端口号。
–topic <String: topic>	操作的topic名称。
–create	创建主题。
–delete	删除主题。
–alter	修改主题。
–list	查看所有主题。
–describe	查看主题详细描述。
–partitions <Integer: # of partitions>	设置分区数。
–replication-factor<Integer: replication factor>	设置分区副本。
–config <String: name=value>	更新系统默认的配置。

查看当前服务器中的所有topic

bin/kafka-topics.sh --bootstrap-server node1:9092 --list
1

创建 first topic

bin/kafka-topics.sh --bootstrap-server node2:9092 --create --partitions 1 --replication-factor 3 --topic first

--topic 定义topic名
--replication-factor  定义副本数
--partitions  定义分区数

1
2
3
4
5
6

查看first主题的详情

bin/kafka-topics.sh --bootstrap-server node1:9092 --describe --topic first
1

修改分区数（注意：分区数只能增加，不能减少）

bin/kafka-topics.sh --bootstrap-server node1:9092 --alter --topic first --partitions 3

Topic: first	TopicId: shEIm88JQfasdndSNtHURw	PartitionCount: 3	ReplicationFactor: 3	Configs: segment.bytes=1073741824
	Topic: first	Partition: 0	Leader: 2	Replicas: 2,1,0	Isr: 2,1,0
	Topic: first	Partition: 1	Leader: 0	Replicas: 0,1,2	Isr: 0,1,2
	Topic: first	Partition: 2	Leader: 1	Replicas: 1,2,0	Isr: 1,2,0
1
2
3
4
5
6

删除topic

bin/kafka-topics.sh --bootstrap-server node1:9092 --delete --topic first
1

2.2.2 生产者命令行

查看操作生产者命令参数
```
bin/kafka-console-producer.sh
1
```
参数描述
–bootstrap-server <String: server toconnect to> 连接的Kafka Broker主机名称和端口号。
–topic <String: topic> 操作的topic名称。

发送消息

bin/kafka-console-producer.sh --bootstrap-server node2:9092 --topic first
> hello world
1
2

2.2.3 消费者命令行

查看消费者命令行参数

bin/kafka-console-consumer.sh
1

参数	描述
–bootstrap-server <String: server toconnect to>	连接的Kafka Broker主机名称和端口号。
–topic <String: topic>	操作的topic名称。
–from-beginning	从头开始消费。
–group <String: consumer group id>	指定消费者组名称。

消费消息

消费first主题中的数据

bin/kafka-console-consumer.sh --bootstrap-server node1:9092 --topic first
1

把主题中所有的数据都读取出来（包括历史数据（kafka持久化的数据默认7天））
```
bin/kafka-console-consumer.sh --bootstrap-server node1:9092 --from-beginning --topic first
1
```

3、生产者

3.1 发送原理

在消息发送的过程中，涉及到了两个线程—— main 线程和Sender 线程**。在main线程中创建了**一个双端队列 RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka Broker。
在这里插入图片描述

图解：

batch.size :只有数据积累到batch.size之后，sender才会发送数据默认16K
linger.ms :如果数据迟迟未达到batch.size大小。sender设置的等待时间到了之后就会自动发送数据，单位为ms,默认是0ms,代表没有延迟
应答ACK
- 0：生产者发送过来的数据不需要等数据落盘应答
- 1：生产者发送过来数据之后，leader收到后应答
- -1（all）：生产者发送过来数据后，Leader和ISR队列里面的所有节点收齐数据后应答

3.2 生产者重要参数列表

参数名称	描述
bootstrap.servers	生产者连接集群所需的broker地址清单。例如node1:9092,node2:9092,node3:9092，可以设置1个或者多个，中间用逗号隔开。注意这里并非需要所有的broker地址，因为生产者从给定的broker里查找到其他broker信息。
key.serializer和value.serializer	指定发送消息的key和value的序列化类型。一定要写全类名。
buffer.memory	RecordAccumulator缓冲区总大小，默认32m。
batch.size	缓冲区一批数据最大值，默认16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。
linger.ms	如果数据迟迟未达到batch.size，sender等待linger.time之后就会发送数据。单位ms，默认值是0ms，表示没有延迟。生产环境建议该值大小为5-100ms之间。
acks	0：生产者发送过来的数据，不需要等数据落盘应答。 1：生产者发送过来的数据，Leader收到数据后应答。 -1（all）：生产者发送过来的数据，Leader和isr队列里面的所有节点收齐数据后应答。默认值是-1，-1和all是等价的。
max.in.flight.requests.per.connection	允许最多没有返回ack的次数，默认为5，开启幂等性要保证该值是 1-5的数字。
retries	当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。默认是int最大值，2147483647。如果设置了重试，还想保证消息的有序性，需要设置 MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1否则在重试此失败消息的时候，其他的消息可能发送成功了。
retry.backoff.ms	两次重试之间的时间间隔，默认是100ms。
enable.idempotence	是否开启幂等性，默认true，开启幂等性。
compression.type	生产者发送的所有数据的压缩方式。默认是none，也就是不压缩。支持压缩类型：none、gzip、snappy、lz4和zstd。

3.3异步发送API

3.3.1 普通异步发送

maven项目导入jar包依赖

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

1
2
3
4
5
6
7
8

package org.example.kafka;

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 CustomProducer {
    public static void main(String[] args) {
        //1、kafka生产者配置对象
        Properties config = new Properties();
        // 2、添加配置信息
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //3、创建kafka生产者对象
        KafkaProducer producer = new KafkaProducer(config);
        //4、发送数据
        for (int i = 0; i < 5; i++) {
            producer.send(new ProducerRecord("first", "hello" + i));
        }
        //5、关闭
        producer.close();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

3.3.2 带回调函数的异步发送

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

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

package org.example.kafka;

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

import java.util.Objects;
import java.util.Properties;

public class CustomProducerCallback {
    public static void main(String[] args) {
        // 1、创建配置信息
        Properties config = new Properties();
        //2、添加配置信息
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //3 创建生产者对象
        KafkaProducer producer = new KafkaProducer(config);
        //4 发送数据 添加回调
        for (int i = 0; i < 5; i++) {
            producer.send(new ProducerRecord("first", "hello callback" + i), new Callback() {
                // 在producer收到ack时3调用，为异步调用
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    // 没有异常发送成功 打印信息
                    if (Objects.isNull(exception)) {
                        System.out.println("topic:" + metadata.topic() + "分区：" + metadata.partition());
                    } else {
                        exception.printStackTrace();
                    }
                }
            });
        }
        producer.close();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

3.4 同步API

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

package org.example.kafka;

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 CustomProducer {
    public static void main(String[] args) {
        //1、kafka生产者配置对象
        Properties config = new Properties();
        // 2、添加配置信息
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //3、创建kafka生产者对象
        KafkaProducer producer = new KafkaProducer(config);
        //4、发送数据
        for (int i = 0; i < 5; i++) {
            producer.send(new ProducerRecord("first", "hello" + i)).get();
        }
        //5、关闭
        producer.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

3.4 生产者分区

3.4.1 分区的好处

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

3.4.2 分区策略

默认的分区器 DefaultPartitioner

/**
 * The default partitioning strategy:
 * <ul>
 * <li>If a partition is specified in the record, use it
 * <li>If no partition is specified but a key is present choose a partition based on a hash of the key
 * <li>If no partition or key is present choose the sticky partition that changes when the batch is full.
 * 
 * See KIP-480 for details about sticky partitioning.
 */
public class DefaultPartitioner implements Partitioner {
...
}
1
2
3
4
5
6
7
8
9
10
11
12

ProdecerRecord构造

上述都为指定partition的情况，发送到指定的partition

/**
* 没有指定partition但是有key的情况，将key的hash值与topic的partition数进行取余，得到partition值
*/
public ProducerRecord(String topic, Integer partition, K key, V value) {
        this(topic, partition, null, key, value, null);
    }
1
2
3
4
5
6

/**
既没有partition也没有key的情况下，kafka采用sticky Partition (粘性分区器)，会随机选择一个分区，并尽可能使用该分区。等待该分区的batch已满或者已完成，Kafka再随机选择一个分区进行使用（和上一次的分区不同）
*/
public ProducerRecord(String topic, K key, V value) {
        this(topic, null, null, key, value, null);
    }

1
2
3
4
5
6
7

3.4.2.1 case1 将数据发送指定的partition的情况—>将所有数据发往分区1中

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class CustomProducerCallbackPartitions {

    public static void main(String[] args) {

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

        // 2. 给kafka配置对象添加配置信息
      properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1: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());

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        for (int i = 0; i < 5; i++) {
            // 指定数据发送到1号分区，key为空（IDEA中ctrl + p查看参数）
            kafkaProducer.send(new ProducerRecord<>("first", 1,"","hello world " + 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();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

3.4.2.2 没有指明partition值但是有key的情况，将key的hash值与partition总数进行取余，得到具体的partition值

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class CustomProducerCallback {

    public static void main(String[] args) {

        Properties properties = new Properties();
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        for (int i = 0; i < 5; i++) {
            // 依次指定key值为a,b,f ，数据key的hash值与3个分区求余，分别发往1、2、0
            kafkaProducer.send(new ProducerRecord<>("first", "a","hello world " + 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();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

3.4.3自定义分区

实现步骤

定义类实现Partitioner接口
重写partition方法

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

/**
 * 1. 实现接口Partitioner
 * 2. 实现3个方法:partition,close,configure
 * 3. 编写partition方法,返回分区号
 */
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;

        // 根据具体的业务逻辑 返回具体的partition

        // 返回分区号
        return partition;
    }

    // 关闭资源
    @Override
    public void close() {

    }

    // 配置方法
    @Override
    public void configure(Map<String, ?> configs) {

    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

使用分区器的方法，在生产者的配置中添加分区器参数。

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class CustomProducerCallbackPartitions {

    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
   properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 添加自定义分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,MyPartitioner.class.getName());

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        for (int i = 0; i < 5; i++) {
            
            kafkaProducer.send(new ProducerRecord<>("first", "hello " + 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();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

3.4.4 生产经验-如何提高吞吐量

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

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class CustomProducerParameters {

    public static void main(String[] args) throws InterruptedException {

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

        // 2. 给kafka配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1: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");

        // 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","hello" + i));

        }

        // 5. 关闭资源
        kafkaProducer.close();
    }
} 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

3.4.5 生产经验-数据可靠性

ACK应答等级（acks）

0 : 生产者发送过来数据，不需要等待数据落盘应答 ----- 数据可靠性分析：丢失
1：生产者发送过来数据，Leader收到数据后应答 ----- 数据可靠性分析：丢失
-1（all）: 生产者发送过来数据，Leader和ISR队列里面的所有节点收齐数据后应答

思考：

leader收到数据后。所有follower都开始同步数据，但有一follwer,因为某种故障，迟迟不能与leader进行同步，这个问题怎么解决？

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,
这样就不用等待联系不上的follower或者已经故障的follower
1
2
3

数据完全可靠性条件： ACK设施为-1，分区副本大于等于2并且ISR队列里面的副本数量大于等于2

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class CustomProducerAck {

    public static void main(String[] args) throws InterruptedException {

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

        // 2. 给kafka配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1: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());

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

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

        // 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","hello " + i));

        }

        // 5. 关闭资源
        kafkaProducer.close();
    }
} 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

3.4.6 生产经验-数据去重

3.4.6.1 数据传递语义

至少一次（at least once） : ack设置为-1，分区副本数>=2 ,且ISR最小副本数量>=2，
- At Least once :可以保证数据不丢失，但是不能保证数据不重复
最多一次(at most once) ： ack设置为0
- At Most once ：可以保证数据不重复，但是不能保证数据不丢失
精确一次（Exactly once）：对与一些非常重要的信息，比如和钱相关的信息，就必须达到：既不丢失数据，也不重复数据

kafka0.11版本之后引入了 幂等性和事务

3.4.6.2 幂等性

幂等性原理

幂等性就是值不论Producer向broker发送多少重复的数据，broker只会持久化一条，保证不重复

精确一次（Exactly once）: 幂等性 + 至少一次（ack=-1 分区副本数>=2,ISR最小副本数 >=2）

重复数据判断标准：具有<PID，partition,seqNumber>相同主键的消息提交时，Broker只会持久化一条，其中PID是kafka每次重启都会重新生成一个新的。Partition分区号，SeqNumber是单调递增的

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

3.4.6.3 生产者事务

kafka事务原理

开启事务必须开启幂等性

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;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

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

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class CustomProducerTransactions {

    public static void main(String[] args) throws InterruptedException {

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

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

        // key,value序列化
        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 < 5; i++) {
                // 发送消息
                kafkaProducer.send(new ProducerRecord<>("first", "hello " + i));
            }

//            int i = 1 / 0;

            // 提交事务
            kafkaProducer.commitTransaction();

        } catch (Exception e) {
            // 终止事务
            kafkaProducer.abortTransaction();
        } finally {
            // 5. 关闭资源
            kafkaProducer.close();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

3.4.7 生产经验-数据有序性

单个分区内，有序
多个分区，分区与分区间无序

3.4.8 生产经验-数据乱序

kafka在1.x之前保证数据单分区内有序，条件如下:max.in.flight.requests.per.connection=1
1.x之后单分区内有序
1. 未开启幂等 max.in.flight.requests.per.connection=1
2. 开启幂等性 max.in.flight.requests.per.connection需要设置小于等于5
  
  因为在开启幂等性后，kafka服务端会缓存producer发送过来的最近5个request的元数据
  
  所以无论如何，都可以保证最近5个request的数据是有序的

4、Kafka Broker

4.1 kafka broker 工作流程

4.1.1 zookeeper 存储的kafka信息

在这里插入图片描述

#记录谁是leader 有哪些服务器可以使用
/kafka/brokers/topics/first/partitions/0/state
{"controller_epoch":2,"leader":1,"version":1,"leader_epoch":3,"isr":[1,2,0]}
#辅助选举 保存控制器（broker的leader， 这里的leader要和副本的leader区分开，这里的leader是kafka集群中所有broker的leader）对应的brokerId信息等
/kafka/controller
{"version":1,"brokerid":1,"timestamp":"1681261027316"}
/kafka/consumers#0.9版本之前用于保存offset信息，0.9版本之后offset存储在kafka主题中。
/controller_epoch #这里用来解决脑裂问题，存放的是一个整形值（纪元编号，也称为隔离令牌）
/config/topics#存储动态修改主题级别的配置信息
/config/clients#存储动态修改客户端级别的配置信息
/config/changes#动态修改配置时存储相应的信息
/admin/delete_topics#在对主题进行删除操作时保存待删除主题的信息
/isr_change_notification#保存Kafka副本ISR列表发生变化时通知的相应路径
1
2
3
4
5
6
7
8
9
10
11
12
13

4.1.2 kafka broke工作流程

4.1.2.1 leader选举的种类

这里需要先明确一个概念leader选举，因为kafka中涉及多处选举机制，容易搞混，kafka由三个方面会涉及到选举：

broker（控制器）选leader
分区多副本选leader
消费者选leader

4.1.2.2 Broker选举过程

在kafka集群中由很多的broker（也叫做控制器），但是他们之间需要选举出一个leader，其他的都是follower。broker的leader有很重要的作用，诸如：创建、删除主题、增加分区并分配leader分区；集群broker管理，包括新增、关闭和故障处理；分区重分配（auto.leader.rebalance.enable=true，后面会介绍）；分区leader选举

下面看一下Broker的leader选举过程和故障处理

在这里插入图片描述

集群中第一个启动的broker会在zookeeper中创建临时节点/controller来让自己成为控制器，当其他的broker启动也会在zookeeper中创建临时节点/controller，但是会发现节点已经存在，此时会受到一个异常，此时就会在zookeeper中创建一个watch对象，方便这些broker接受leader的变更消息；
如果主leader因为网络问题与zookeeper断开连接或者发生异常退出了，其他的broker就可以通watch接收到控制器的变更通知，开始尝试去创建临时节点/controller，如果有一个broker创建成功了，就和上面说的一样，其他的broker也会收到异常通知，此时就说明集群中broker的leader已经确定，其他的broker只能创建watch对象了
集群中broker的leader发生异常退出，在选举出新的leader之后，会检测这个异常的broker上面是否有分区副本的leader，如果有就是发起分区的leader选举，选出新额分区leader，然后更新ISR队列数据

kafka解决脑裂问题

为了解决Controller脑裂问题，zookeeper中有一个持久节点/controller_epoch，存放的是一个整形值的epoch number（纪元编号，也称为隔离令牌），集群中每选举一次控制器，就会通过Zookeeper创建一个数值更大的epoch number，如果有broker收到比这个epoch数值小的数据，就会忽略消息
1

4.1.3 broker重要参数

参数名称	描述
replica.lag.time.max.ms	ISR中，如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。该时间阈值，默认30s。
auto.leader.rebalance.enable	默认是true。自动Leader Partition 平衡。
leader.imbalance.per.broker.percentage	默认是10%。每个broker允许的不平衡的leader的比率。如果每个broker超过了这个值，控制器会触发leader的平衡。
leader.imbalance.check.interval.seconds	默认值300秒。检查leader负载是否平衡的间隔时间。
log.segment.bytes	Kafka中log日志是分成一块块存储的，此配置是指log日志划分成块的大小，默认值1G。
log.index.interval.bytes	默认4kb，kafka里面每当写入了4kb大小的日志（.log），然后就往index文件里面记录一个索引。
log.retention.hours	Kafka中数据保存的时间，默认7天。
log.retention.minutes	Kafka中数据保存的时间，分钟级别，默认关闭。
log.retention.ms	Kafka中数据保存的时间，毫秒级别，默认关闭。
log.retention.check.interval.ms	检查数据是否保存超时的间隔，默认是5分钟。
log.retention.bytes	默认等于-1，表示无穷大。超过设置的所有日志总大小，删除最早的segment。
log.cleanup.policy	默认是delete，表示所有数据启用删除策略；如果设置值为compact，表示所有数据启用压缩策略。
num.io.threads	默认是8。负责写磁盘的线程数。整个参数值要占总核数的50%。
num.replica.fetchers	副本拉取线程数，这个参数占总核数的50%的1/3
num.network.threads	默认是3。数据传输线程数，这个参数占总核数的50%的2/3 。
log.flush.interval.messages	强制页缓存刷写到磁盘的条数，默认是long的最大值，9223372036854775807。一般不建议修改，交给系统自己管理。
log.flush.interval.ms	每隔多久，刷数据到磁盘，默认是null。一般不建议修改，交给系统自己管理。

4.2 节点服役和退役

4.2.1 节点服役

新增节点

关闭node3，并克隆一台命名为node4

开启node4,更改地址

# vim /etc/sysconfig/network-scripts/ifcfg-ens33

TYPE="Ethernet"
BOOTPROTO="static"
DEFROUTE="yes"
PEERDNS="yes"
PEERROUTES="yes"
IPV4_FAILURE_FATAL="no"
IPV6INIT="yes"
IPV6_AUTOCONF="yes"
IPV6_DEFROUTE="yes"
IPV6_PEERDNS="yes"
IPV6_PEERROUTES="yes"
IPV6_FAILURE_FATAL="no"
IPV6_ADDR_GEN_MODE="stable-privacy"
NAME="ens33"
UUID="48cbb1b5-14a8-4576-984e-1afedb15bbf0"
DEVICE="ens33"
ONBOOT="yes"
IPADDR=192.168.245.134
NETMASK=255.255.255.0
GATEWAY=192.168.245.2
DNS1=8.8.8.8
DNS2=221.7.128.68
DNS3=221.7.136.68

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

修改主机名
```
vim /etc/hostname
node4
1
2
```

配置hosts(每台机器)

vim /etc/hosts
192.168.245.131 node1
192.168.245.132 node2
192.168.245.133 node3
192.168.245.134 node4
1
2
3
4
5

在node上修改broker.id为3
删除ndoe4下kafka下的kafka-logs 以及logs
重启node3,node4
启动node1，node2,node3原集群
单独启动node4中的kafka

执行负载均衡操作

创建一个要均衡的主题

vim topics-to-move.json 

{
    "topics": [

        {"topic": "first"}

    ],

    "version": 1
}

1
2
3
4
5
6
7
8
9
10
11
12

生成一个负载均衡的计划

bin/kafka-reassign-partitions.sh --bootstrap-server node2:9092  --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate

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

Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,3,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,1,2],"log_dirs":["any","any","any"

1
2
3
4
5
6
7
8

创建副本存储计划（所有副本存储在broker0、broker1、broker2、broker3中）。

vim increase-replication-factor.json
#输入上面generate生成的内容
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,3,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,1,2],"log_dirs":["any","any","any"]}]}

1
2
3
4

执行副本执行计划

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

验证副本存储计划

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

Status of partition reassignment:
Reassignment of partition first-0 is complete.
Reassignment of partition first-1 is complete.
Reassignment of partition first-2 is complete.

Clearing broker-level throttles on brokers 0,1,2,3
Clearing topic-level throttles on topic first

1
2
3
4
5
6
7
8
9
10

4.2.2 节点退役

执行负载均衡操作

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

创建一个要均衡的主题

vim topics-to-move.json 

{
    "topics": [

        {"topic": "first"}

    ],

    "version": 1
}
1
2
3
4
5
6
7
8
9
10
11

创建执行计划

bin/kafka-reassign-partitions.sh --bootstrap-server node2:9092  --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2" --generate

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

Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,2,0],"log_dirs":["any","any","any"]}]}

1
2
3
4
5
6
7
8

创建副本存储计划（所有副本存储在broker0、broker1、broker2中）。

vim increase-replication-factor.json

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

执行副本存储计划。

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

验证副本存储计划验证副本存储计划

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

Status of partition reassignment:
Reassignment of partition first-0 is complete.
Reassignment of partition first-1 is complete.
Reassignment of partition first-2 is complete.

Clearing broker-level throttles on brokers 0,1,2,3
Clearing topic-level throttles on topic first

1
2
3
4
5
6
7
8
9
10

执行停止命令

在node4上执行  bin/kafka-server-stop.sh 
1

4.3 kafka副本信息

4.3.1副本基本信息

Kafka副本作用：提高数据可靠性。
Kafka默认副本1个，生产环境一般配置为2个，保证数据可靠性；太多副本会增加磁盘存储空间，增加网络上数据传输，降低效率
Kafka中副本分为：Leader和Follower。Kafka生产者只会把数据发往Leader，然后Follower找Leader进行同步数据。
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副本同步时，延迟过多的副本

4.3.2Leader选举流程

Kafka集群中有一个broker的Controller会被选举为Controller Leader，负责管理集群broker的上下线，所有topic的分区副本分配和Leader选举等工作。

Controller的信息同步工作是依赖于Zookeeper的。
在这里插入图片描述

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

bin/kafka-topics.sh --bootstrap-server node2:9092 --create --topic second --partitions 4 --replication-factor 4
Created topic second.

1
2
3

查看Leader分布情况

bin/kafka-topics.sh --bootstrap-server node2:9092 --describe --topic second
Topic: second	TopicId: awpgX_7WR-OX3Vl6HE8sVg	PartitionCount: 4	ReplicationFactor: 4	Configs: segment.bytes=1073741824
	Topic: second	Partition: 0	Leader: 3	Replicas: 3,0,2,1	Isr: 3,0,2,1
	Topic: second	Partition: 1	Leader: 1	Replicas: 1,2,3,0	Isr: 1,2,3,0
	Topic: second	Partition: 2	Leader: 0	Replicas: 0,3,1,2	Isr: 0,3,1,2
	Topic: second	Partition: 3	Leader: 2	Replicas: 2,1,0,3	Isr: 2,1,0,3

1
2
3
4
5
6
7

停止掉node4的kafka进程，并查看Leader分区情况

bin/kafka-server-stop.sh

 $ bin/kafka-topics.sh --bootstrap-server node2:9092 --describe --topic second 
Topic: second	TopicId: awpgX_7WR-OX3Vl6HE8sVg	PartitionCount: 4	ReplicationFactor: 4	Configs: segment.bytes=1073741824
	Topic: second	Partition: 0	Leader: 0	Replicas: 3,0,2,1	Isr: 0,2,1
	Topic: second	Partition: 1	Leader: 1	Replicas: 1,2,3,0	Isr: 1,2,0
	Topic: second	Partition: 2	Leader: 0	Replicas: 0,3,1,2	Isr: 0,1,2
	Topic: second	Partition: 3	Leader: 2	Replicas: 2,1,0,3	Isr: 2,1,0

1
2
3
4
5
6
7
8
9

停止掉node3的kafka进程，并查看Leader分区情况

second bin/kafka-server-stop.sh

  bin/kafka-topics.sh --bootstrap-server node2:9092 --describe --topic second 
Topic: second	TopicId: awpgX_7WR-OX3Vl6HE8sVg	PartitionCount: 4	ReplicationFactor: 4	Configs: segment.bytes=1073741824
	Topic: second	Partition: 0	Leader: 0	Replicas: 3,0,2,1	Isr: 0,1
	Topic: second	Partition: 1	Leader: 1	Replicas: 1,2,3,0	Isr: 1,0
	Topic: second	Partition: 2	Leader: 0	Replicas: 0,3,1,2	Isr: 0,1
	Topic: second	Partition: 3	Leader: 1	Replicas: 2,1,0,3	Isr: 1,0

1
2
3
4
5
6
7
8
9

启动node4的kafka进程，并查看Leader分区情况

$ bin/kafka-server-start.sh -daemon config/server.properties

 $ bin/kafka-topics.sh --bootstrap-server ndoe2:9092 --describe --topic second 
Topic: second	TopicId: awpgX_7WR-OX3Vl6HE8sVg	PartitionCount: 4	ReplicationFactor: 4	Configs: segment.bytes=1073741824
	Topic: second	Partition: 0	Leader: 0	Replicas: 3,0,2,1	Isr: 0,1,3
	Topic: second	Partition: 1	Leader: 1	Replicas: 1,2,3,0	Isr: 1,0,3
	Topic: second	Partition: 2	Leader: 0	Replicas: 0,3,1,2	Isr: 0,1,3
	Topic: second	Partition: 3	Leader: 1	Replicas: 2,1,0,3	Isr: 1,0,3

1
2
3
4
5
6
7
8
9

启动node3的kafka进程，并查看Leader分区情况

$ bin/kafka-server-start.sh -daemon config/server.properties

$ bin/kafka-topics.sh --bootstrap-server ndoe2:9092 --describe --topic second 
Topic: second	TopicId: awpgX_7WR-OX3Vl6HE8sVg	PartitionCount: 4	ReplicationFactor: 4	Configs: segment.bytes=1073741824
	Topic: second	Partition: 0	Leader: 0	Replicas: 3,0,2,1	Isr: 0,1,3,2
	Topic: second	Partition: 1	Leader: 1	Replicas: 1,2,3,0	Isr: 1,0,3,2
	Topic: second	Partition: 2	Leader: 0	Replicas: 0,3,1,2	Isr: 0,1,3,2
	Topic: second	Partition: 3	Leader: 1	Replicas: 2,1,0,3	Isr: 1,0,3,2

1
2
3
4
5
6
7
8
9

停止掉node2的kafka进程，并查看Leader分区情况

$ bin/kafka-server-stop.sh

$ bin/kafka-topics.sh --bootstrap-server ndoe2:9092 --describe --topic second 
Topic: second	TopicId: awpgX_7WR-OX3Vl6HE8sVg	PartitionCount: 4	ReplicationFactor: 4	Configs: segment.bytes=1073741824
	Topic: second	Partition: 0	Leader: 0	Replicas: 3,0,2,1	Isr: 0,3,2
	Topic: second	Partition: 1	Leader: 2	Replicas: 1,2,3,0	Isr: 0,3,2
	Topic: second	Partition: 2	Leader: 0	Replicas: 0,3,1,2	Isr: 0,3,2
	Topic: second	Partition: 3	Leader: 2	Replicas: 2,1,0,3	Isr: 0,3,2

1
2
3
4
5
6
7
8
9

4.3.3Follower节点故障

此时假如Broker1上的follower发生故障会出现什么情况呢？首先Broker1上的follower会被踢出ISR队列中，但是leader和其他的follower都还是会继续接受数据，并不会收到影响，对应的LEO和HW都会往后移动；如果此时发生故障的Broker1上的follower恢复后，此时Broker1上的follower会读取本地磁盘记录的上次HW位置，并将log文件中高于HW的部分截取掉，从HW开始向Leader进行同步；直到Broker1上的follower的LEO大于等于该分区的HW，此时说明这个follower追上了leader，就会将其重新加入ISR队列中。

在这里插入图片描述

在这里插入图片描述 ###### 4.3.4 Leader节点故障

上面了解了follower故障的情况，那么如果leader发生故障呢？接着上面的图片来看，首先如果Broker0上的leader发生故障之后，也是一样会先从ISR队列中被踢出，然后从ISR中选出一个新的Leader来；此时为了保证多个副本之间的数据一致性，其他的follower会先将各自的log文件中高于HW的部分截取掉，然后从新的leader同步数据（由此可知这只能保证副本之间数据一致性，并不能保证数据不丢失或者不重复）。
在这里插入图片描述

4.3.5 分区副本分配

如果kafka服务器只有4个节点，那么设置kafka的分区数大于服务器台数，在kafka底层如何分配存储副本呢？

创建16分区，3个副本

创建一个新的topic，名称为second

bin/kafka-topics.sh --bootstrap-server node2:9092 --create --partitions 16 --replication-factor 3 --topic second
1

查看分区和副本情况

[root@node1 kafka]#  bin/kafka-topics.sh --bootstrap-server node2:9092 --describe --topic second
Topic: second	Partition: 0	Leader: 0	Replicas: 0,1,2	Isr: 0,1,2
Topic: second	Partition: 1	Leader: 1	Replicas: 1,2,3	Isr: 1,2,3
Topic: second	Partition: 2	Leader: 2	Replicas: 2,3,0	Isr: 2,3,0
Topic: second	Partition: 3	Leader: 3	Replicas: 3,0,1	Isr: 3,0,1

Topic: second	Partition: 4	Leader: 0	Replicas: 0,2,3	Isr: 0,2,3
Topic: second	Partition: 5	Leader: 1	Replicas: 1,3,0	Isr: 1,3,0
Topic: second	Partition: 6	Leader: 2	Replicas: 2,0,1	Isr: 2,0,1
Topic: second	Partition: 7	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2

Topic: second	Partition: 8	Leader: 0	Replicas: 0,3,1	Isr: 0,3,1
Topic: second	Partition: 9	Leader: 1	Replicas: 1,0,2	Isr: 1,0,2
Topic: second	Partition: 10	Leader: 2	Replicas: 2,1,3	Isr: 2,1,3
Topic: second	Partition: 11	Leader: 3	Replicas: 3,2,0	Isr: 3,2,0

Topic: second	Partition: 12	Leader: 0	Replicas: 0,1,2	Isr: 0,1,2
Topic: second	Partition: 13	Leader: 1	Replicas: 1,2,3	Isr: 1,2,3
Topic: second	Partition: 14	Leader: 2	Replicas: 2,3,0	Isr: 2,3,0
Topic: second	Partition: 15	Leader: 3	Replicas: 3,0,1	Isr: 3,0,1

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

在这里插入图片描述

4.3.6 生产经验-手动调整分区副本存储

在生产环境中，每台服务器的配置和性能不一样，但是kafka只会根据自己的规则创建分区副本，就会导致个别服务器面临存储的压力。所以需要手动调整分区副本的存储
在这里插入图片描述

手动调整分区副本存储的步骤如下：

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

bin/kafka-topics.sh --bootstrap-server node2:9092 --create --partitions 4 --replication-factor 2 --topic three
1

查看分区副本存储情况

bin/kafka-topics.sh --bootstrap-server node2:9092 --describe --topic three
1

创建副本存储计划（所有副本都指定存储在broker0、broker1中）

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

1
2
3
4
5
6
7
8
9

执行副本存储计划

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

验证副本存储计划

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

查看分区副本存储情况

bin/kafka-topics.sh --bootstrap-server node2:9092 --describe --topic three
1

4.3.7 生产经验-leader partition 负载均衡

正常情况下，kafka会自动把leader partition均衡分散在各个机器上，来保证每台机器的读写吞吐量都是均衡的，但是如果某些broker宕机，会导致leader partition过去集中在其他几台少部分机器上，从而导致这几台broker读写请求过高。其它宕机的broker重启之后都是follower partition,读写请求很低，造成集群负载不均衡

参数名称	描述
auto.leader.rebalance.enable	默认是true。自动Leader Partition 平衡。生产环境中，leader重选举的代价比较大，可能会带来性能影响，建议设置为false关闭。
leader.imbalance.per.broker.percentage	默认是10%。每个broker允许的不平衡的leader的比率。如果每个broker超过了这个值，控制器会触发leader的平衡。
leader.imbalance.check.interval.seconds	默认值300秒。检查leader负载是否平衡的间隔时间。

4.3.8生产经验——增加副本因子

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

创建topic

bin/kafka-topics.sh --bootstrap-server ndoe2:9092 --create --partitions 3 --replication-factor 1 --topic four
1

手动增加副本存储

创建副本存储计划（所有副本都指定存储在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]}]}
1
2

执行副本存储计划。

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

4.4 文件存储

4.4.1 文件存储机制

topic数据存储机制

topic 是逻辑上的概念，而partition是物理上的概念，每个partition对应与一个log文件，该log文件中存储的就是Producer生产的数据。producer生产的数据会不断的追加到该log文件末端，为防止log文件过大导致数据定位效率低下，kafka采用了分片和索引机制，将每个partition分为多个segment，每个segment包括：index文件、log文件、timeindex文件，这些文件在同一个文件夹下，该文件夹的命名规则为：topic名称+分区序号比如：first-0
在这里插入图片描述

4.4.2 topic数据存储位置

cd /opt/soft/kafka/kafka-logs/first-0
[root@node1 first-0]# ll
总用量 20
-rw-r--r--. 1 root root 10485760 4月  12 08:57 00000000000000000000.index
-rw-r--r--. 1 root root      210 4月  12 09:13 00000000000000000000.log
-rw-r--r--. 1 root root 10485756 4月  12 08:57 00000000000000000000.timeindex
-rw-r--r--. 1 root root       10 4月  12 08:57 00000000000000000001.snapshot
-rw-r--r--. 1 root root       12 4月  12 09:04 leader-epoch-checkpoint
-rw-r--r--. 1 root root       43 4月   7 15:57 partition.metadata
1
2
3
4
5
6
7
8
9

分段文件名规则：分区的第一个segment是从0开始的，后续每个segment文件名为上一个segment文件最后一条消息的offset，ofsset的数值最大为64位（long类型），20位数字字符长度，没有数字用0填充

log文件默认写满1G后，会进行log rolling形成一个新的分段（segment）来记录消息，这里面的分段大小取决于：log.segment.bytes参数决定。

index和timeindex文件在刚使用时会分配10M的大小，当进行log rolling后，它会修剪为实际的大小，所以看到前几个索引文件的大小，只有几百K

查看index文件

bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files kafka-logs/first-0/00000000000000000000.index
Dumping kafka-logs/first-0/00000000000000000000.index
offset: 0 position: 0
1
2
3

查看log文件

bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files kafka-logs/first-0/00000000000000000000.log
Dumping kafka-logs/first-0/00000000000000000000.log
Starting offset: 0
baseOffset: 0 lastOffset: 0 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 0 CreateTime: 1680854454623 size: 70 magic: 2 compresscodec: none crc: 2149647709 isvalid: true
baseOffset: 1 lastOffset: 1 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 3 isTransactional: false isControl: false position: 70 CreateTime: 1681261467334 size: 69 magic: 2 compresscodec: none crc: 3985046740 isvalid: true
baseOffset: 2 lastOffset: 2 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 3 isTransactional: false isControl: false position: 139 CreateTime: 1681262035105 size: 71 magic: 2 compresscodec: none crc: 961893559 isvalid: true
1
2
3
4
5
6

4.4.3 index和log文件详解

当log文件写入4k（这里可以通过log.index.interval.bytes设置）数据，就会写入一条索引信息到index文件中，这样的index索引文件就是一个稀疏索引，它并不会每条日志都建立索引信息。

log日志文件是顺序写入，大体上由message+实际offset+position组成，而索引文件的数据结构则是由相对offset（4byte）+position（4byte）组成。

当kafka查询一条offset对应实际消息时，可以通过index进行二分查找，获取最近的低位offset，然后从低位offset对应的position开始，从实际的log文件中开始往后查找对应的消息

时间戳索引文件，它的作用是可以查询某一个时间段内的消息，它的数据结构是：时间戳（8byte）+ 相对offset（4byte），如果要使用这个索引文件，先要通过时间范围找到对应的offset，然后再去找对应的index文件找到position信息，最后在遍历log文件，这个过程也是需要用到index索引文件的
在这里插入图片描述

参数	描述
log.segment.bytes	Kafka中log日志是分成一块块存储的，此配置是指log日志划分成块的大小，默认值1G。
log.index.interval.bytes	默认4kb，kafka里面每当写入了4kb大小的日志（.log），然后就往index文件里面记录一个索引。稀疏索引。

4.4.4 文件清洗策略

Kafka中默认的日志保存时间为7天，可以通过调整如下参数修改保存时间

log.retention.hours，最低优先级小时，默认7天。
log.retention.minutes，分钟。
log.retention.ms，最高优先级毫秒。
log.retention.check.interval.ms，负责设置检查周期，默认5分钟。
file.delete.delay.ms：延迟执行删除时间
log.retention.bytes：当设置为-1时表示运行保留日志最大值（相当于关闭）；当设置为1G时，表示日志文件最大值

那么日志一旦超过了设置的时间，怎么处理呢？

Kafka中提供的日志清理策略有delete和compact两种

delete日志删除：将过期数据删除
1. log.cleanup.policy = delete 所有数据启用删除策略
2. 基于时间：默认打开。以segment中所有记录中的最大时间戳作为该文件时间戳
  
  日志删除任务会周期检查当前日志文件中是否有保留时间超过设定的阈值来寻找可删除的日志段文件集合；这里需要注意log.retention参数的优先级：log.retention.ms > log.retention.minutes > log.retention.hours，默认只会配置log.retention.hours参数，值为168即为7天
  
  删除过期的日志段文件，并不是简单的根据日志段文件的修改时间计算，而是要根据该日志段中最大的时间戳来计算的，首先要查询该日志分段所对应的时间戳索引文件，查找该时间戳索引文件的最后一条索引数据，如果时间戳大于0就取值，否则才会使用最近修改时间。
  在删除的时候先从Log对象所维护的日志段的跳跃表中移除要删除的日志段，用来确保已经没有线程来读取这些日志段；接着将日志段所对应的所有文件，包括索引文件都添加上.deleted的后缀；最后交给一个以delete-file命名的延迟任务来删除这些以.deleted为后缀的文件，默认是1分钟执行一次，可以通过file.delete.delay.ms来配置。
3. 基于大小：默认关闭。超过设置的所有日志总大小，删除最早的segment, log.retention.bytes，默认等于-1，表示无穷大
4. 基于日志起始偏移量
  
  该策略判断依据是日志段的下一个日志段的起始偏移量 baseOffset是否小于等于 logStartOffset，如果是，则可以删除此日志分段。
  这里说一下logStartOffset，一般情况下，日志文件的起始偏移量 logStartOffset等于第一个日志分段的 baseOffset，但这并不是绝对的，logStartOffset的值可以通过 DeleteRecordsRequest请求、使用 kafka-delete-records.sh 脚本、日志的清理和截断等操作进行修改。
如果一个segment中有一部分数据过期，一部分没有过期，怎么处理？

保留等到下次清理，一起清理
compact日志压缩

4.5高效读写

kafka是分布式集群，可以采用分区技术，并行度高
读数据采用稀疏索引，可以快速定位要消费的数据
顺写写磁盘

Kafka的producer生产数据，要写入到log文件中，写的过程是一直追加到文件末端，为顺序写。官网有数据表明，同样的磁盘，顺序写能到600M/s，而随机写只有100K/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。

页缓存+零拷贝

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

非零拷贝
在这里插入图片描述

零拷贝
在这里插入图片描述

参数	描述
log.flush.interval.messages	强制页缓存刷写到磁盘的条数，默认是long的最大值，9223372036854775807。一般不建议修改，交给系统自己管理。
log.flush.interval.ms	每隔多久，刷数据到磁盘，默认是null。一般不建议修改，交给系统自己管理。

5、kafka消费者

5.1消费方式

pull（拉）模式:消费者采用从broker中主动拉取数据 kafka使用这种方式
push(推)模式：kafka没有使用这种方式，因为由broker决定消息的发送速率，很难适应所有的消费者的消费速度

pull的不足之处：如果kafka没有数据，消费者可能会陷入循环中，一直返回空数据

5.2消费这工作流程

5.2.1总体工作流程

消费mer组成，形成一个消费者的条件，是所有的消费者的groupId相同，

消费者组

concumer group :消费者组，由多个consumer组成，形成一个消费者的条件，是所有的消费者的groupId相同，

消费者组内的每个消费者负责消费不通分区的数据，一个分区只能由一个组内的一个消费者消费
消费者组之间互不影响，所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者

在这里插入图片描述

5.3消费者重要参数

参数名称	描述
bootstrap.servers	向Kafka集群建立初始连接用到的host/port列表。
key.deserializer和value.deserializer	指定接收消息的key和value的反序列化类型。一定要写全类名。
group.id	标记消费者所属的消费者组。
enable.auto.commit	默认值为true，消费者会自动周期性地向服务器提交偏移量。
auto.commit.interval.ms	如果设置了 enable.auto.commit 的值为true，则该值定义了消费者偏移量向Kafka提交的频率，默认5s。
auto.offset.reset	当Kafka中没有初始偏移量或当前偏移量在服务器中不存在（如，数据被删除了），该如何处理？ earliest：自动重置偏移量到最早的偏移量。 latest：默认，自动重置偏移量为最新的偏移量。 none：如果消费组原来的（previous）偏移量不存在，则向消费者抛异常。 anything：向消费者抛异常。
offsets.topic.num.partitions	__consumer_offsets的分区数，默认是50个分区。
heartbeat.interval.ms	Kafka消费者和coordinator之间的心跳时间，默认3s。该条目的值必须小于 session.timeout.ms ，也不应该高于 session.timeout.ms 的1/3。
session.timeout.ms	Kafka消费者和coordinator之间连接超时时间，默认45s。超过该值，该消费者被移除，消费者组执行再平衡。
max.poll.interval.ms	消费者处理消息的最大时长，默认是5分钟。超过该值，该消费者被移除，消费者组执行再平衡。
fetch.min.bytes	默认1个字节。消费者获取服务器端一批消息最小的字节数。
fetch.max.wait.ms	默认500ms。如果没有从服务器端获取到一批数据的最小字节数。该时间到，仍然会返回数据。
fetch.max.bytes	默认Default: 52428800（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受message.max.bytes （broker config）or max.message.bytes （topic config）影响。
max.poll.records	一次poll拉取数据返回消息的最大条数，默认是500条。

5.4消费者API

5.4.1订阅主题

在消费者API代码中必须配置消费者组id。命令行启动消费者不填写消费者组id会被自动填写随机的消费者组id

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 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, "node1: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, "group-test");

        // 创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);

        // 注册要消费的主题（可以消费多个主题）
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

        // 拉取数据打印
        while (true) {
            // 设置1s中消费一批数据
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            // 打印消费到的数据
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

5.4.2 订阅分区

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 java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Properties;

public class CustomConsumerPartition {

    public static void main(String[] args) {

        Properties properties = new Properties();

        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1: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,"group-test");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

        // 消费某个主题的某个分区数据
        ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
        topicPartitions.add(new TopicPartition("first", 0));
        kafkaConsumer.assign(topicPartitions);

        while (true){

            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

5.4.3消费者

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

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 java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

public class CustomConsumer1 {

    public static void main(String[] args) {

        // 1.创建消费者的配置对象
        Properties properties = new Properties();

        // 2.给消费者配置对象添加参数
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1: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, "group-test");

        // 创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);

        // 注册主题
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

        // 拉取数据打印
        while (true) {
            // 设置1s中消费一批数据
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            // 打印消费到的数据
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

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

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

5.5生产经验-分区的分配以及在均衡

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

在这里插入图片描述

参数名称	描述
heartbeat.interval.ms	Kafka消费者和coordinator之间的心跳时间，默认3s。该条目的值必须小于 session.timeout.ms，也不应该高于 session.timeout.ms 的1/3。
session.timeout.ms	Kafka消费者和coordinator之间连接超时时间，默认45s。超过该值，该消费者被移除，消费者组执行再平衡。
max.poll.interval.ms	消费者处理消息的最大时长，默认是5分钟。超过该值，该消费者被移除，消费者组执行再平衡。
partition.assignment.strategy	消费者分区分配策略，默认策略是Range + CooperativeSticky。Kafka可以同时使用多个分区分配策略。可以选择的策略包括：Range、RoundRobin、Sticky、CooperativeSticky

5.5.1Range以及在均衡

range是对没个topic而言的

对同一个topic，分区按照顺序号进行排序，并对消费者按照字母顺序进行排序

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

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

例如：7/3=2余1 除不尽，那么消费者C0便会多消费1个分区，8/3=2余2，除不尽，那么C0和C1分别多消费1个

如果只是针对1个topic而言，C0消费者多消费1个分区影响不是很大，但是如果有N个topic，那么针对每个topic,消费者C0都将多消费1个分区，topic约多，c0消费的分区会比其他消费者明显多消费N个分区，容易产生数据倾斜

Kafka默认的分区分配策略就是Range + CooperativeSticky，所以不需要修改策略

5.5.1.1 range分区分配再均衡案例

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

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

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

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

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

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

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

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

5.5.2RoundRobin以及再均衡

RoundRobin针对集群中所有Topic而言

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

5.5.2.1 RoundRobin分区分配策略案例

依次在CustomConsumer、CustomConsumer1、CustomConsumer2三个消费者代码中修改分区分配策略为RoundRobin

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

1
2
3

重启3个消费者，重复发送消息的步骤，观看分区结果

RoundRobin 分区分配再平衡案例

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

1号消费者：消费到2、5号分区数据

2号消费者：消费到4、1号分区数据

0号消费者的任务会按照RoundRobin的方式，把数据轮询分成0 、6和3号分区数据，分别由1号消费者或者2号消费者消费。

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

1号消费者：消费到0、2、4、6号分区数据

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

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

5.5.3 Sticky以及再平衡

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

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

1）需求

设置主题为first，7个分区；准备3个消费者，采用粘性分区策略，并进行消费，观察消费分配情况。然后再停止其中一个消费者，再次观察消费分配情况。

2）步骤

（1）修改分区分配策略为粘性。

注意：3个消费者都应该注释掉，之后重启3个消费者，如果出现报错，全部停止等会再重启，或者修改为全新的消费者组。

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

（2）使用同样的生产者发送500条消息。

可以看到会尽量保持分区的个数近似划分分区。

Sticky分区分配再平衡案例

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

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

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

0号消费者的任务会按照粘性规则，尽可能均衡的随机分成0和1号分区数据，分别由1号消费者或者2号消费者消费。

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

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

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

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

5.6 offset位移

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

从0.9版本开始consumer默认将offset保存在kafka一个内置的topic中，该topic为__consumer_offsets

在0.9之前版本是存在方zookeeper中的

消费offset 案例

思想：__consumer_offsets为Kafka中的topic，那就可以通过消费者进行消费。
在配置文件config/consumer.properties中添加配置exclude.internal.topics=false，默认是true，表示不能消费系统主题。为了查看该系统主题数据，所以该参数修改为false。
采用命令行方式，创建一个新的topic。

$ bin/kafka-topics.sh --bootstrap-server node1:9092 --create --topic four --partitions 2 --replication-factor 2
1

启动生产者往four生产数据。

bin/kafka-console-producer.sh --topic four --bootstrap-server node1:9092
1

启动消费者消费four数据。

bin/kafka-console-consumer.sh --bootstrap-server node1:9092 --topic four --group test
1

注意：指定消费者组名称，更好观察数据存储位置（key是group.id+topic+分区号）。

查看消费者消费主题__consumer_offsets。

$ bin/kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server node1:9092 --consumer.config config/consumer.properties --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning
1

5.6.1自动提交Offset

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

自动提交offset的参数

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

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 java.util.Arrays;
import java.util.Properties;

public class CustomConsumerAutoOffset {

    public static void main(String[] args) {

        // 1. 创建kafka消费者配置类
        Properties properties = new Properties();

        // 2. 添加配置参数
        // 添加连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1: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, true);
        // 提交offset的时间周期1000ms，默认5s
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

5.6.2 手动提交offset

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

相同点：都会将本次提交的一批数据最高的偏移量提交

不同点：同步提交阻塞当前线程。一直到提交成功。并且会自动失败重试（由不可抗因素，也会提交失败）。异步提交没有重试，有可能失败

同步提交

/**由于同步提交offset有失败重试机制，故更加可靠，但是由于一直等待提交结果，提交的效率比较低。以下为同步提交offset的示例。
*/
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 java.util.Arrays;
import java.util.Properties;

public class CustomConsumerByHandSync {

    public static void main(String[] args) {
        // 1. 创建kafka消费者配置类
        Properties properties = new Properties();
        // 2. 添加配置参数
        // 添加连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1: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();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

异步提交offset

/**
虽然同步提交offset更可靠一些，但是由于其会阻塞当前线程，直到提交成功。因此吞吐量会受到很大的影响。因此更多的情况下，会选用异步提交offset的方式。
以下为异步提交offset的示例：

*/
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.Arrays;
import java.util.Map;
import java.util.Properties;

public class CustomConsumerByHandAsync {

    public static void main(String[] args) {

        // 1. 创建kafka消费者配置类
        Properties properties = new Properties();

        // 2. 添加配置参数
        // 添加连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1: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();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

5…6.3 指定offset消费

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

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

earliest：自动将偏移量重置为最早的偏移量，–from-beginning。
latest（默认值）：自动将偏移量重置为最新偏移量。
none：如果未找到消费者组的先前偏移量，则向消费者抛出异常。

指定任意offset消费

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.HashSet;
import java.util.Properties;
import java.util.Set;

public class CustomConsumerSeek {

    public static void main(String[] args) {

        // 0  配置信息
        Properties properties = new Properties();

        // 连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092");

        // key value反序列化
        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");

        // 1 创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

        // 2 订阅一个主题
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

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

        // 3 消费该主题数据
        while (true) {

            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

注意：每次执行完，需要修改消费者组名；

5.6.4 指定时间消费

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

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

public class CustomConsumerForTime {

    public static void main(String[] args) {

        // 0  配置信息
        Properties properties = new Properties();

        // 连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092");

        // key value反序列化
        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");

        // 1 创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

        // 2 订阅一个主题
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

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

        // 3 消费该主题数据
        while (true) {

            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

5.5.6 漏消费和重复消费

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

**漏消费：**先提交offset后消费，有可能会造成数据的漏消费。
在这里插入图片描述

5.7 生产经验-消费者事务

如果想完成Consumer端的精准一次消费，那么需要kafka消费端将消费过程与提交offset做原子绑定。

5.8 消费者如何提高吞吐量

在这里插入图片描述

参数名称	描述
fetch.max.bytes	默认Default: 52428800（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受message.max.bytes （broker config）or max.message.bytes （topic config）影响。
max.poll.records	一次poll拉取数据返回消息的最大条数，默认是500条

6、Kafka-Eagle监控

Kafka-Eagle框架可以监控Kafka集群的整体运行情况，在生产环境中经常使用

需要安装mysql

关闭Kafka集群

修改启动脚本

vim bin/kafka-server-start.sh
修改如下参数值：
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
fi
为
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-server -Xms2G -Xmx2G -XX:PermSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:ParallelGCThreads=8 -XX:ConcGCThreads=5 -XX:InitiatingHeapOccupancyPercent=70"
    export JMX_PORT="9999"
    #export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
fi

1
2
3
4
5
6
7
8
9
10
11
12

注意：修改之后在启动Kafka之前要分发之其他节点

kafka-Eagle安装

https://www.kafka-eagle.org/
上传压缩包kafka-eagle-bin-2.0.8.tar.gz到集群/opt/soft目录
解压 tar -zxvf kafka-eagle-bin-2.0.8.tar.gz
cd到刚刚解压过的文件夹
将efak-web-2.0.8-bin.tar.gz解压至/opt/soft tar -zxvf efak-web-2.0.8-bin.tar.gz -C /opt/soft/
修改 mv efak-web-2.0.8/ efak

修改配置文件 /opt/soft/efak/conf/system-config.properties

vim system-config.properties

######################################
# multi zookeeper & kafka cluster list
# Settings prefixed with 'kafka.eagle.' will be deprecated, use 'efak.' instead
######################################
efak.zk.cluster.alias=cluster1
cluster1.zk.list=node1:2181,node2:2181,node3:2181/kafka

######################################
# zookeeper enable acl
######################################
cluster1.zk.acl.enable=false
cluster1.zk.acl.schema=digest
cluster1.zk.acl.username=test
cluster1.zk.acl.password=test123

######################################
# broker size online list
######################################
cluster1.efak.broker.size=20

######################################
# zk client thread limit
######################################
kafka.zk.limit.size=32
######################################
# EFAK webui port
######################################
efak.webui.port=8048

######################################
# kafka jmx acl and ssl authenticate
######################################
cluster1.efak.jmx.acl=false
cluster1.efak.jmx.user=keadmin
cluster1.efak.jmx.password=keadmin123
cluster1.efak.jmx.ssl=false
cluster1.efak.jmx.truststore.location=/data/ssl/certificates/kafka.truststore
cluster1.efak.jmx.truststore.password=ke123456

######################################
# kafka offset storage
######################################
# offset保存在kafka
cluster1.efak.offset.storage=kafka


######################################
# kafka jmx uri
######################################
cluster1.efak.jmx.uri=service:jmx:rmi:///jndi/rmi://%s/jmxrmi

######################################
# kafka metrics, 15 days by default
######################################
efak.metrics.charts=true
efak.metrics.retain=15

######################################
# kafka sql topic records max
######################################
efak.sql.topic.records.max=5000
efak.sql.topic.preview.records.max=10

######################################
# delete kafka topic token
######################################
efak.topic.token=keadmin

######################################
# kafka sasl authenticate
######################################
cluster1.efak.sasl.enable=false
cluster1.efak.sasl.protocol=SASL_PLAINTEXT
cluster1.efak.sasl.mechanism=SCRAM-SHA-256
cluster1.efak.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="kafka" password="kafka-eagle";
cluster1.efak.sasl.client.id=
cluster1.efak.blacklist.topics=
cluster1.efak.sasl.cgroup.enable=false
cluster1.efak.sasl.cgroup.topics=
cluster2.efak.sasl.enable=false
cluster2.efak.sasl.protocol=SASL_PLAINTEXT
cluster2.efak.sasl.mechanism=PLAIN
cluster2.efak.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="kafka-eagle";
cluster2.efak.sasl.client.id=
cluster2.efak.blacklist.topics=
cluster2.efak.sasl.cgroup.enable=false
cluster2.efak.sasl.cgroup.topics=

######################################
# kafka ssl authenticate
######################################
cluster3.efak.ssl.enable=false
cluster3.efak.ssl.protocol=SSL
cluster3.efak.ssl.truststore.location=
cluster3.efak.ssl.truststore.password=
cluster3.efak.ssl.keystore.location=
cluster3.efak.ssl.keystore.password=
cluster3.efak.ssl.key.password=
cluster3.efak.ssl.endpoint.identification.algorithm=https
cluster3.efak.blacklist.topics=
cluster3.efak.ssl.cgroup.enable=false
cluster3.efak.ssl.cgroup.topics=

######################################
# kafka sqlite jdbc driver address
######################################
# 配置mysql连接
efak.driver=com.mysql.jdbc.Driver
efak.url=jdbc:mysql://node1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
efak.username=root
efak.password=000000

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

添加环境变量

# kafka EFAK
export KE_HOME=/opt/module/efak
export PATH=$PATH:$KE_HOME/bin

 source /etc/profile
1
2
3
4
5

启动

先启动zk
再启动kafka

最后启动efak

#启动
bin/ke.sh start  
#停止
bin/ke.sh stop
1
2
3
4

##################################
cluster1.efak.sasl.enable=false
cluster1.efak.sasl.protocol=SASL_PLAINTEXT
cluster1.efak.sasl.mechanism=SCRAM-SHA-256
cluster1.efak.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=“kafka” password=“kafka-eagle”;
cluster1.efak.sasl.client.id=
cluster1.efak.blacklist.topics=
cluster1.efak.sasl.cgroup.enable=false
cluster1.efak.sasl.cgroup.topics=
cluster2.efak.sasl.enable=false
cluster2.efak.sasl.protocol=SASL_PLAINTEXT
cluster2.efak.sasl.mechanism=PLAIN
cluster2.efak.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=“kafka” password=“kafka-eagle”;
cluster2.efak.sasl.client.id=
cluster2.efak.blacklist.topics=
cluster2.efak.sasl.cgroup.enable=false
cluster2.efak.sasl.cgroup.topics=

  ######################################
  # kafka ssl authenticate
  ######################################
  cluster3.efak.ssl.enable=false
  cluster3.efak.ssl.protocol=SSL
  cluster3.efak.ssl.truststore.location=
  cluster3.efak.ssl.truststore.password=
  cluster3.efak.ssl.keystore.location=
  cluster3.efak.ssl.keystore.password=
  cluster3.efak.ssl.key.password=
  cluster3.efak.ssl.endpoint.identification.algorithm=https
  cluster3.efak.blacklist.topics=
  cluster3.efak.ssl.cgroup.enable=false
  cluster3.efak.ssl.cgroup.topics=
  
  ######################################
  # kafka sqlite jdbc driver address
  ######################################
  # 配置mysql连接
  efak.driver=com.mysql.jdbc.Driver
  efak.url=jdbc:mysql://node1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
  efak.username=root
  efak.password=000000
  
  ######################################
  # kafka mysql jdbc driver address
  ######################################
  #efak.driver=com.mysql.cj.jdbc.Driver
  #efak.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
  #efak.username=root
  #efak.password=123456
  
  ```
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

添加环境变量

# kafka EFAK
export KE_HOME=/opt/module/efak
export PATH=$PATH:$KE_HOME/bin

 source /etc/profile
1
2
3
4
5

启动

先启动zk
再启动kafka

最后启动efak

#启动
bin/ke.sh start  
#停止
bin/ke.sh stop
1
2
3
4

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/花生_TL007/article/detail/377736