（作者：Angelvn大数据研发工作室）
版本：V5.1
- Kafka概述
- 定义
- 消息队列
目前企业中比较常见的消息队列产品主要有Kafka、ActiveMQ、RabbitMQ、RocketMQ等。
在大数据应用场景中，主要采用Kafka作为消息队列。在JavaEE开发中主要采用ActiveMQ、RabbitMQ、RocketMQ。
- 传统消息队列的应用场景
传统的消息队列的主要应用场景包括：缓存/消峰、解耦和异步通信。
- 消息队列的两种模式
- Kafka基础架构
- Producer：消息生产者，就是向Kafka broker发消息的客户端。
- Consumer：消息消费者，从Kafka broker取消息的客户端。
- Consumer Group（CG）：消费者组，由多个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。
- Consumer：消息消费者，从Kafka broker取消息的客户端。
- Kafka快速入门
- 安装部署
- 集群规划
- 安装部署
| hadoop102 | hadoop103 | hadoop104 |
| zk | zk | zk |
| kafka | kafka | kafka |
- 集群部署官方下载地址：http://kafka.apache.org/downloads.html解压安装包。
[Angelvn@hadoop102 software]$ tar -zxvf kafka_2.12-3.6.1.tgz -C /opt/module/- 进入到/opt/module/kafka_2.12-3.6.1目录，修改配置文件。
[Angelvn@hadoop102 kafka]$ cd config/
[Angelvn@hadoop102 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=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka- 分发安装包。
[Angelvn@hadoop102 module]$ xsync kafka/- 分别在hadoop103和hadoop104上修改配置文件/opt/module/kafka/config/server.properties中的broker.id=1、broker.id=2。
注：broker.id不得重复，整个集群中唯一。
[Angelvn@hadoop103 module]$ vim kafka/config/server.properties修改:
# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1
[Angelvn@hadoop104 module]$ vim kafka/config/server.properties
修改:
# The id of the broker. This must be set to a unique integer for each broker.
broker.id=2- 配置环境变量。在/etc/profile.d/my_env.sh文件中增加kafka环境变量配置。
[Angelvn@hadoop102 module]$ sudo vim /etc/profile.d/my_env.sh
增加如下内容：
#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin- 刷新一下环境变量。
[Angelvn@hadoop102 module]$ source /etc/profile- 分发环境变量文件到其他节点，并source。
[Angelvn@hadoop102 module]$ sudo /home/Angelvn/bin/xsync /etc/profile.d/my_env.sh
[Angelvn@hadoop103 module]$ source /etc/profile
[Angelvn@hadoop104 module]$ source /etc/profile- 启动集群。先启动Zookeeper集群，然后启动Kafka。
[Angelvn@hadoop102 kafka]$ zk.sh start- 依次在hadoop102、hadoop103、hadoop104节点上启动Kafka。
[Angelvn@hadoop102 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties
[Angelvn@hadoop103 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties
[Angelvn@hadoop104 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties注意：配置文件的路径要能够到server.properties。
- 关闭集群。
[Angelvn@hadoop102 kafka]$ bin/kafka-server-stop.sh
[Angelvn@hadoop103 kafka]$ bin/kafka-server-stop.sh
[Angelvn@hadoop104 kafka]$ bin/kafka-server-stop.sh- 集群启停脚本在/home/Angelvn/bin目录下创建文件kf.sh脚本文件。
[Angelvn@hadoop102 bin]$ vim kf.sh脚本如下：
#! /bin/bash
case $1 in
"start"){
for i in hadoop102 hadoop103 hadoop104
do
echo " --------启动 $i Kafka-------"
ssh $i "/opt/module/kafka_2.12-3.6.1/bin/kafka-server-start.sh -daemon /opt/module/kafka_2.12-3.6.1/config/server.properties"
done
};;
"stop"){
for i in hadoop102 hadoop103 hadoop104
do
echo " --------停止 $i Kafka-------"
ssh $i "/opt/module/kafka_2.12-3.6.1/bin/kafka-server-stop.sh "
done
};;
esac- 添加执行权限。
[Angelvn@hadoop102 bin]$ chmod +x kf.sh- 启动集群命令。
[Angelvn@hadoop102 ~]$ kf.sh start- 停止集群命令。
[Angelvn@hadoop102 ~]$ kf.sh stop注意：停止Kafka集群时，一定要等Kafka所有节点进程全部停止后再停止Zookeeper集群。因为Zookeeper集群当中记录着Kafka集群相关信息，Zookeeper集群一旦先停止，Kafka集群就没有办法再获取停止进程的信息，只能手动杀死Kafka进程了。
- Kafka命令行操作
- 主题命令行操作查看操作主题命令参数。
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh| 参数 | 描述 |
| --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。
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --list- 创建first topic。
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 1 --replication-factor 3 --topic first选项说明：
- --topic 定义topic名
- --replication-factor 定义副本数
- --partitions 定义分区数
- 查看first主题的详情。
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first- 修改分区数（注意：分区数只能增加，不能减少）。
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 3- 再次查看first主题的详情。
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first- 删除topic（学生自己演示）。
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --delete --topic first- 生产者命令行操作查看操作生产者命令参数。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-producer.sh| 参数 | 描述 |
| --bootstrap-server <String: server toconnect to> | 连接的Kafka Broker主机名称和端口号。 |
| --topic <String: topic> | 操作的topic名称。 |
- 发送消息。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first
>hello world
>Angelvn Angelvn- 消费者命令行操作查看操作消费者命令参数。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-consumer.sh| 参数 | 描述 |
| --bootstrap-server <String: server toconnect to> | 连接的Kafka Broker主机名称和端口号。 |
| --topic <String: topic> | 操作的topic名称。 |
| --from-beginning | 从头开始消费。 |
| --group <String: consumer group id> | 指定消费者组名称。 |
- 消费消息消费first主题中的数据。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first- 把主题中所有的数据都读取出来（包括历史数据）。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --from-beginning --topic first- Kafka生产者
- 生产者消息发送流程
- 发送原理
- 生产者消息发送流程
在消息发送的过程中，涉及到了两个线程——main线程和Sender线程。在main线程中创建了一个双端队列RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka Broker。
- 生产者重要参数列表
| 参数名称 | 描述 |
| bootstrap.servers | 生产者连接集群所需的broker地址清单。例如hadoop102:9092,hadoop103:9092,hadoop104: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。 |
- 异步发送API普通异步发送需求：创建Kafka生产者，采用异步的方式发送到Kafka Broker。
- 代码编写。创建工程kafka。导入依赖。
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.6.1</version>
</dependency>
</dependencies>- 创建包名：com.Angelvn.kafka.producer。
- 编写不带回调函数的API代码。
package com.Angelvn.kafka.producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
public class CustomProducer {
public static void main(String[] args) throws InterruptedException {
// 1. 创建kafka生产者的配置对象
Properties properties = new Properties();
// 2. 给kafka配置对象添加配置信息：bootstrap.servers
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102: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 < 5; i++) {
kafkaProducer.send(new ProducerRecord<>("first","Angelvn " + i));
}
// 5. 关闭资源
kafkaProducer.close();
}
}测试：
① 在hadoop102上开启Kafka消费者。
[Angelvn@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
② 在IDEA中执行代码，观察hadoop102控制台中是否接收到消息。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
Angelvn 0
Angelvn 1
Angelvn 2
Angelvn 3
Angelvn 4- 带回调函数的异步发送
回调函数会在producer收到ack时调用，为异步调用，该方法有两个参数，分别是元数据信息（RecordMetadata）和异常信息（Exception），如果Exception为null，说明消息发送成功，如果Exception不为null，说明消息发送失败。
注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。
package com.Angelvn.kafka.producer;
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class CustomProducerCallback {
public static void main(String[] args) throws InterruptedException {
// 1. 创建kafka生产者的配置对象
Properties properties = new Properties();
// 2. 给kafka配置对象添加配置信息
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102: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. 创建kafka生产者对象
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
// 4. 调用send方法,发送消息
for (int i = 0; i < 5; i++) {
// 添加回调
kafkaProducer.send(new ProducerRecord<>("first", "Angelvn " + i), new Callback() {
// 该方法在Producer收到ack时调用，为异步调用
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception == null) {
// 没有异常,输出信息到控制台
System.out.println("主题：" + metadata.topic() + "->" + "分区：" + metadata.partition());
} else {
// 出现异常打印
exception.printStackTrace();
}
}
});
// 延迟一会会看到数据发往不同分区
Thread.sleep(2);
}
// 5. 关闭资源
kafkaProducer.close();
}
}测试：
① 在hadoop102上开启Kafka消费者。
[Angelvn@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first② 在IDEA中执行代码，观察hadoop102控制台中是否接收到消息。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
Angelvn 0
Angelvn 1
Angelvn 2
Angelvn 3
Angelvn 4③ 在IDEA控制台观察回调信息。
主题：first->分区：0
主题：first->分区：0
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
- 同步发送API
只需在异步发送的基础上，再调用一下get()方法即可。
package com.Angelvn.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;
import java.util.concurrent.ExecutionException;
public class CustomProducerSync {
public static void main(String[] args) throws InterruptedException, ExecutionException {
// 1. 创建kafka生产者的配置对象
Properties properties = new Properties();
// 2. 给kafka配置对象添加配置信息
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102: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. 创建kafka生产者对象
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
// 4. 调用send方法,发送消息
for (int i = 0; i < 10; i++) {
// 异步发送 默认
// kafkaProducer.send(new ProducerRecord<>("first","kafka" + i));
// 同步发送
kafkaProducer.send(new ProducerRecord<>("first","kafka" + i)).get();
}
// 5. 关闭资源
kafkaProducer.close();
}
}测试：
① 在hadoop102上开启Kafka消费者。
[Angelvn@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first② 在IDEA中执行代码，观察hadoop102控制台中是否接收到消息。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
Angelvn 0
Angelvn 1
Angelvn 2
Angelvn 3
Angelvn 4- 生产者分区分区好处
- 生产者发送消息的分区策略默认的分区器 DefaultPartitioner
在IDEA中ctrl + n，全局查找DefaultPartitioner。
NOTE this partitioner is deprecated and shouldn't be used. To use default partitioning logic remove partitioner.class configuration setting. See KIP-794 for more info. The default partitioning strategy:
If a partition is specified in the record, use it
If no partition is specified but a key is present choose a partition based on a hash of the key
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.
@Deprecated
public class DefaultPartitioner implements Partitioner {
}
- 案例一
将数据发往指定partition，例如，将所有数据发往分区1中。
package com.Angelvn.kafka.producer;
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,"hadoop102: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,"","Angelvn " + 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();
}
}测试：
① 在hadoop102上开启Kafka消费者。
[Angelvn@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first② 在IDEA中执行代码，观察hadoop102控制台中是否接收到消息。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
Angelvn 0
Angelvn 1
Angelvn 2
Angelvn 3
Angelvn 4③ 在IDEA控制台观察回调信息。
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
- 案例二
没有指明partition值但有key的情况下，将key的hash值与topic的partition数进行取余得到partition值。
package com.Angelvn.kafka.producer;
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.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
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","Angelvn " + 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();
}
}测试：
① key="a"时，在控制台查看结果。
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
② key="b"时，在控制台查看结果。
主题：first->分区：2
主题：first->分区：2
主题：first->分区：2
主题：first->分区：2
主题：first->分区：2
③key="f"时，在控制台查看结果。
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
- 自定义分区器
如果研发人员可以根据企业需求，自己重新实现分区器。
- 需求
实现一个分区器，发送过来的数据中如果包含Angelvn，就发往0号分区，不包含Angelvn，就发往1号分区。
- 实现步骤定义类实现Partitioner接口。重写partition()方法。
package com.Angelvn.kafka.producer;
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;
// 判断消息是否包含Angelvn
if (msgValue.contains("Angelvn")){
partition = 0;
}else {
partition = 1;
}
// 返回分区号
return partition;
}
// 关闭资源
@Override
public void close() {
}
// 配置方法
@Override
public void configure(Map<String, ?> configs) {
}
}- 使用分区器的方法，在生产者的配置中添加分区器参数。
package com.Angelvn.kafka.producer;
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,"hadoop102: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,"com.Angelvn.kafka.producer.MyPartitioner");
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
for (int i = 0; i < 5; i++) {
kafkaProducer.send(new ProducerRecord<>("first", "Angelvn " + 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();
}
}- 测试
① 在hadoop102上开启Kafka消费者。
[Angelvn@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first② 在IDEA控制台观察回调信息。
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
- 生产经验——生产者如何提高吞吐量
package com.Angelvn.kafka.producer;
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, "hadoop102: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","Angelvn " + i));
}
// 5. 关闭资源
kafkaProducer.close();
}
}测试：
① 在hadoop102上开启Kafka消费者。
[Angelvn@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first② 在IDEA中执行代码，观察hadoop102控制台中是否接收到消息。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
Angelvn 0
Angelvn 1
Angelvn 2
Angelvn 3
Angelvn 4- 生产经验——数据可靠性回顾发送流程
- ack应答原理
- 代码配置
package com.Angelvn.kafka.producer;
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, "hadoop102: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","Angelvn " + i));
}
// 5. 关闭资源
kafkaProducer.close();
}
}- 生产经验——数据去重数据传递语义
- 幂等性幂等性原理
- 如何使用幂等性
开启参数enable.idempotence 默认为true，false关闭。
- 生产者事务
0.11版本的Kafka同时引入了事务的特性，为了实现跨分区跨会话的事务，需要引入一个全局唯一的Transaction ID，并将Producer获得的PID和Transaction ID绑定。这样当Producer重启后就可以通过正在进行的Transaction ID获得原来的PID。
为了管理Transaction，Kafka引入了一个新的组件Transaction Coordinator。Producer就是通过和Transaction Coordinator交互获得Transaction ID对应的任务状态。Transaction Coordinator还负责将事务所有写入Kafka的一个内部Topic，这样即使整个服务重启，由于事务状态得到保存，进行中的事务状态可以得到恢复，从而继续进行。
注意：提前开启幂等性!!!
- 生产经验——数据有序or怎么解决乱序Kafka 最多只保证单分区内的消息是有序的，所以如果要保证业务全局严格有序，就要设置 Topic 为单分区。
- 如何保证单分区内数据有序or怎么解决乱序？
- Kafka Broker
- Kafka Broker工作流程
- Zookeeper存储的Kafka信息启动Zookeeper客户端。
- Kafka Broker工作流程
[Angelvn@hadoop102 zookeeper-3.5.7]$ bin/zkCli.sh- 通过ls命令可以查看kafka相关信息。
[zk: localhost:2181(CONNECTED) 2] ls /kafka
- Kafka Broker总体工作流程
模拟Kafka上下线，Zookeeper中数据变化
- 查看/kafka/brokers/ids路径上的节点。
[zk: localhost:2181(CONNECTED) 2] ls /kafka/brokers/ids
[0, 1, 2]- 查看/kafka/controller路径上的数据。
[zk: localhost:2181(CONNECTED) 15] get /kafka/controller
{"version":1,"brokerid":0,"timestamp":"1637292471777"}- 查看/kafka/brokers/topics/first/partitions/0/state路径上的数据。
[zk: localhost:2181(CONNECTED) 16] get /kafka/brokers/topics/first/partitions/0/state
{"controller_epoch":24,"leader":0,"version":1,"leader_epoch":18,"isr":[0,1,2]}- 停止hadoop104上的kafka。
[Angelvn@hadoop104 kafka]$ bin/kafka-server-stop.sh- 再次查看/kafka/brokers/ids路径上的节点。
[zk: localhost:2181(CONNECTED) 3] ls /kafka/brokers/ids
[0, 1]- 再次查看/kafka/controller路径上的数据。
[zk: localhost:2181(CONNECTED) 15] get /kafka/controller
{"version":1,"brokerid":0,"timestamp":"1637292471777"}- 再次查看/kafka/brokers/topics/first/partitions/0/state路径上的数据。
[zk: localhost:2181(CONNECTED) 16] get /kafka/brokers/topics/first/partitions/0/state
{"controller_epoch":24,"leader":0,"version":1,"leader_epoch":18,"isr":[0,1]}- 启动hadoop104上的kafka。
[Angelvn@hadoop104 kafka]$ bin/kafka-server-start.sh -daemon ./config/server.properties
- 再次观察（1）、（2）、（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。一般不建议修改，交给系统自己管理。 |
- Kafka 副本副本基本信息Kafka副本作用：提高数据可靠性。Kafka默认副本 ，生产环境一般配置为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副本同步时，延迟过多的副本。
- Leader选举流程
Kafka集群中有一个broker的Controller会被选举为Controller Leader，负责管理集群broker的上下线，所有topic的分区副本分配和Leader选举等工作。
Controller的信息同步工作是依赖于Zookeeper的。
- 创建一个新的topic，4个分区，4个副本
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic Angelvn1 --partitions 4 --replication-factor 4
Created topic Angelvn1.- 查看Leader分布情况
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic Angelvn1
Topic: Angelvn1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824
Topic: Angelvn1 Partition: 0 Leader: 3 Replicas: 3,0,2,1 Isr: 3,0,2,1
Topic: Angelvn1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,2,3,0
Topic: Angelvn1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,3,1,2
Topic: Angelvn1 Partition: 3 Leader: 2 Replicas: 2,1,0,3 Isr: 2,1,0,3- 停止掉hadoop105的kafka进程，并查看Leader分区情况
[Angelvn@hadoop105 kafka]$ bin/kafka-server-stop.sh
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic Angelvn1
Topic: Angelvn1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824
Topic: Angelvn1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,2,1
Topic: Angelvn1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,2,0
Topic: Angelvn1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,1,2
Topic: Angelvn1 Partition: 3 Leader: 2 Replicas: 2,1,0,3 Isr: 2,1,0- 停止掉hadoop104的kafka进程，并查看Leader分区情况
[Angelvn@hadoop104 kafka]$ bin/kafka-server-stop.sh
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic Angelvn1
Topic: Angelvn1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824
Topic: Angelvn1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,1
Topic: Angelvn1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,0
Topic: Angelvn1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,1
Topic: Angelvn1 Partition: 3 Leader: 1 Replicas: 2,1,0,3 Isr: 1,0- 启动hadoop105的kafka进程，并查看Leader分区情况
[Angelvn@hadoop105 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic Angelvn1
Topic: Angelvn1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824
Topic: Angelvn1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,1,3
Topic: Angelvn1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,0,3
Topic: Angelvn1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,1,3
Topic: Angelvn1 Partition: 3 Leader: 1 Replicas: 2,1,0,3 Isr: 1,0,3- 启动hadoop104的kafka进程，并查看Leader分区情况
[Angelvn@hadoop104 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic Angelvn1
Topic: Angelvn1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824
Topic: Angelvn1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,1,3,2
Topic: Angelvn1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,0,3,2
Topic: Angelvn1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,1,3,2
Topic: Angelvn1 Partition: 3 Leader: 1 Replicas: 2,1,0,3 Isr: 1,0,3,2- 停止掉hadoop103的kafka进程，并查看Leader分区情况
[Angelvn@hadoop103 kafka]$ bin/kafka-server-stop.sh
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic Angelvn1
Topic: Angelvn1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824
Topic: Angelvn1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,3,2
Topic: Angelvn1 Partition: 1 Leader: 2 Replicas: 1,2,3,0 Isr: 0,3,2
Topic: Angelvn1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,3,2
Topic: Angelvn1 Partition: 3 Leader: 2 Replicas: 2,1,0,3 Isr: 0,3,2- Leader和Follower故障处理细节
- 文件存储文件存储机制Topic数据的存储机制
- 思考：Topic数据到底存储在什么位置？启动生产者，并发送消息。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first
>hello world- 查看hadoop102（或者hadoop103、hadoop104）的/opt/module/kafka/datas/first-1（first-0、first-2）路径上的文件。
[Angelvn@hadoop104 first-1]$ ls
00000000000000000092.index
00000000000000000092.log
00000000000000000092.snapshot
00000000000000000092.timeindex
leader-epoch-checkpoint
partition.metadata- 直接查看log日志，发现是乱码。
[Angelvn@hadoop104 first-1]$ cat 00000000000000000092.log
\CYnF|©|©ÿÿÿÿÿÿÿÿÿÿÿÿÿÿ"hello world- 通过工具查看index和log信息。
[Angelvn@hadoop104 first-1]$ kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.index
Dumping ./00000000000000000000.index
offset: 3 position: 152// 查询kafka具体消息的解析方式
kafka-run-class.sh kafka.tools.DumpLogSegments --print-data-log --files 00000000000000000000.log
[Angelvn@hadoop104 first-1]$ kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.log
Dumping datas/first-0/00000000000000000000.log
Starting offset: 0
baseOffset: 0 lastOffset: 1 count: 2 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 0 CreateTime: 1636338440962 size: 75 magic: 2 compresscodec: none crc: 2745337109 isvalid: true
baseOffset: 2 lastOffset: 2 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 75 CreateTime: 1636351749089 size: 77 magic: 2 compresscodec: none crc: 273943004 isvalid: true
baseOffset: 3 lastOffset: 3 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 152 CreateTime: 1636351749119 size: 77 magic: 2 compresscodec: none crc: 106207379 isvalid: true
baseOffset: 4 lastOffset: 8 count: 5 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 229 CreateTime: 1636353061435 size: 141 magic: 2 compresscodec: none crc: 157376877 isvalid: true
baseOffset: 9 lastOffset: 13 count: 5 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 370 CreateTime: 1636353204051 size: 146 magic: 2 compresscodec: none crc: 4058582827 isvalid: true- index文件和log文件详解
说明：日志存储参数配置
| 参数 | 描述 |
| log.segment.bytes | Kafka中log日志是分成一块块存储的，此配置是指log日志划分 成块的大小，默认值1G。 |
| log.index.interval.bytes | 默认4kb，kafka里面每当写入了4kb大小的日志（.log），然后就往index文件里面记录一个索引。 稀疏索引。 |
- 文件清理策略
Kafka中默认的日志保存时间为7天，可以通过调整如下参数修改保存时间。
- log.retention.hours，最低优先级小时，默认7天。
- log.retention.minutes，分钟。
- log.retention.ms，最高优先级毫秒。
- log.retention.check.interval.ms，负责设置检查周期，默认5分钟。
那么日志一旦超过了设置的时间，怎么处理呢？
Kafka中提供的日志清理策略有delete和compact两种。
- delete日志删除：将过期数据删除
log.cleanup.policy = delete 所有数据启用删除策略
- 基于时间：默认打开。以segment中所有记录中的最大时间戳作为该文件时间戳。
- 基于大小：默认关闭。超过设置的所有日志总大小，删除最早的segment。
log.retention.bytes，默认等于-1，表示无穷大。
思考：如果一个segment中有一部分数据过期，一部分没有过期，怎么处理？
- compact日志压缩
- 高效读写数据Kafka本身是分布式集群，可以采用分区技术，并行度高读数据采用稀疏索引，可以快速定位要消费的数据顺序写磁盘
Kafka的producer生产数据，要写入到log文件中，写的过程是一直追加到文件末端，为顺序写。官网有数据表明，同样的磁盘，顺序写能到600M/s，而随机写只有100K/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。
- 页缓存 + 零拷贝技术
| 参数 | 描述 |
| log.flush.interval.messages | 强制页缓存刷写到磁盘的条数，默认是long的最大值，9223372036854775807。一般不建议修改，交给系统自己管理。 |
| log.flush.interval.ms | 每隔多久，刷数据到磁盘，默认是null。一般不建议修改，交给系统自己管理。 |
- Kafka消费者
- Kafka消费方式
- Kafka消费者工作流程消费者总体工作流程
- 消费者组原理
- 消费者重要参数
| 参数名称 | 描述 |
| 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条。 |
- 消费者API独立消费者案例（订阅主题）需求：
创建一个独立消费者，消费first主题中数据。
注意：在消费者API代码中必须配置消费者组id。命令行启动消费者不填写消费者组id会被自动填写随机的消费者组id。
- 实现步骤创建包名：com.Angelvn.kafka.consumer编写代码
package com.Angelvn.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 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, "hadoop102:9092");
// 配置序列化 必须
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 配置消费者组（组名任意起名） 必须
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
// 创建消费者对象
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
// 注册要消费的主题（可以消费多个主题）
ArrayList<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);
}
}
}
}- 测试在IDEA中执行消费者程序。在Kafka集群控制台，创建Kafka生产者，并输入数据。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first
>hello- 在IDEA控制台观察接收到的数据。
ConsumerRecord(topic = first, partition = 1, leaderEpoch = 3, offset = 0, CreateTime = 1629160841112, serialized key size = -1, serialized value size = 5, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello)- 消费者组案例需求：测试同一个主题的分区数据，只能由一个消费者组中的一个消费。
- 案例实操复制一份基础消费者的代码，在IDEA中同时启动，即可启动同一个消费者组中的两个消费者。
package com.Angelvn.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 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, "hadoop102:9092");
// 配置序列化 必须
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 配置消费者组 必须
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
// 创建消费者对象
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
// 注册主题
ArrayList<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);
}
}
}
}- 启动代码中的生产者发送消息，在IDEA控制台即可看到两个消费者在消费不同分区的数据（如果只发生到一个分区，可以在发送时增加延迟代码Thread.sleep(2);）。
ConsumerRecord(topic = first, partition = 0, leaderEpoch = 2, offset = 3, CreateTime = 1629169606820, serialized key size = -1, serialized value size = 8, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello1)
ConsumerRecord(topic = first, partition = 1, leaderEpoch = 3, offset = 2, CreateTime = 1629169609524, serialized key size = -1, serialized value size = 6, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello2)
ConsumerRecord(topic = first, partition = 2, leaderEpoch = 3, offset = 21, CreateTime = 1629169611884, serialized key size = -1, serialized value size = 6, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello3)- 重新发送到一个全新的主题中，由于默认创建的主题分区数为1，可以看到只能有一个消费者消费到数据。
- 生产经验——分区的分配以及再平衡
| 参数名称 | 描述 |
| 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 |
- Range以及再平衡Range分区策略原理
- Range分区分配策略案例修改主题first为7个分区。
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 7注意：分区数可以增加，但是不能减少。
- 复制CustomConsumer类，创建CustomConsumer2。这样可以由三个消费者CustomConsumer、CustomConsumer1、CustomConsumer2组成消费者组，组名都为“test”，同时启动3个消费者。
- 启动CustomProducer生产者，发送500条消息，随机发送到不同的分区。
package com.Angelvn.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) throws InterruptedException {
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
for (int i = 0; i < 7; i++) {
kafkaProducer.send(new ProducerRecord<>("first", i, "test", "Angelvn"));
}
kafkaProducer.close();
}
}说明：Kafka默认的分区分配策略就是Range + CooperativeSticky，所以不需要修改策略。
- 观看3个消费者分别消费哪些分区的数据。
- 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方式分配。
- RoundRobin以及再平衡RoundRobin分区策略原理
- RoundRobin分区分配策略案例依次在CustomConsumer、CustomConsumer1、CustomConsumer2三个消费者代码中修改分区分配策略为RoundRobin。
// 修改分区分配策略
properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RoundRobinAssignor");- 重启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方式分配。
- Sticky以及再平衡
粘性分区定义：可以理解为分配的结果带有“粘性的”。即在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可以节省大量的开销。
粘性分区是Kafka从0.11.x版本开始引入这种分配策略，首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。
- 需求
设置主题为first，7个分区；准备3个消费者，采用粘性分区策略，并进行消费，观察消费分配情况。然后再停止其中一个消费者，再次观察消费分配情况。
- 步骤修改分区分配策略为粘性。
注意：3个消费者都应该注释掉，之后重启3个消费者，如果出现报错，全部停止等会再重启，或者修改为全新的消费者组。
// 修改分区分配策略
ArrayList<String> startegys = new ArrayList<>();
startegys.add("org.apache.kafka.clients.consumer.StickyAssignor");
properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, startegys);- 使用同样的生产者发送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已经被踢出消费者组，所以重新按照粘性方式分配。
- offset位移offset的默认维护位置
__consumer_offsets主题里面采用key和value的方式存储数据。key是group.id+topic+分区号，value就是当前offset的值。每隔一段时间，kafka内部会对这个topic进行compact，也就是每个group.id+topic+分区号就保留最新数据。
消费offset案例
- 思想：__consumer_offsets为Kafka中的topic，那就可以通过消费者进行消费。
- 在配置文件config/consumer.properties中添加配置exclude.internal.topics=false，默认是true，表示不能消费系统主题。为了查看该系统主题数据，所以该参数修改为false。
- 采用命令行方式，创建一个新的topic。
- 在配置文件config/consumer.properties中添加配置exclude.internal.topics=false，默认是true，表示不能消费系统主题。为了查看该系统主题数据，所以该参数修改为false。
[Angelvn@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic Angelvn --partitions 2 --replication-factor 2- 启动生产者往Angelvn生产数据。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-producer.sh --topic Angelvn --bootstrap-server hadoop102:9092- 启动消费者消费Angelvn数据。
[Angelvn@hadoop104 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic Angelvn --group test注意：指定消费者组名称，更好观察数据存储位置（key是group.id+topic+分区号）。
- 查看消费者消费主题__consumer_offsets。
[Angelvn@hadoop102 kafka]$ bin/kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server hadoop102:9092 --consumer.config config/consumer.properties --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning
[offset,Angelvn,1]::OffsetAndMetadata(offset=7, leaderEpoch=Optional[0], metadata=, commitTimestamp=1622442520203, expireTimestamp=None)
[offset,Angelvn,0]::OffsetAndMetadata(offset=8, leaderEpoch=Optional[0], metadata=, commitTimestamp=1622442520203, expireTimestamp=None)- 自动提交offset
| 参数名称 | 描述 |
| enable.auto.commit | 默认值为true，消费者会自动周期性地向服务器提交偏移量。 |
| auto.commit.interval.ms | 如果设置了 enable.auto.commit 的值为true， 则该值定义了消费者偏移量向Kafka提交的频率，默认5s。 |
消费者自动提交offset
package com.Angelvn.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 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, "hadoop102: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());
}
}
}
}- 手动提交offset
- 同步提交offset
由于同步提交offset有失败重试机制，故更加可靠，但是由于一直等待提交结果，提交的效率比较低。以下为同步提交offset的示例。
package com.Angelvn.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 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, "hadoop102: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更可靠一些，但是由于其会阻塞当前线程，直到提交成功。因此吞吐量会受到很大的影响。因此更多的情况下，会选用异步提交offset的方式。
以下为异步提交offset的示例：
package com.Angelvn.kafka.consumer;
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, "hadoop102: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中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量时（例如该数据已被删除），该怎么办？
- earliest：自动将偏移量重置为最早的偏移量，--from-beginning。
- latest（默认值）：自动将偏移量重置为最新偏移量。
- none：如果未找到消费者组的先前偏移量，则向消费者抛出异常。
- 漏消费和重复消费
重复消费：已经消费了数据，但是offset没提交。
漏消费：先提交offset后消费，有可能会造成数据的漏消费。
思考：怎么能做到既不漏消费也不重复消费呢？详看消费者事务。
- 生产经验——消费者事务
- 生产经验——数据积压（消费者如何提高吞吐量）
| 参数名称 | 描述 |
| fetch.max.bytes | 默认Default: 52428800（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受message.max.bytes （broker config）or max.message.bytes （topic config）影响。 |
| max.poll.records | 一次poll拉取数据返回消息的最大条数，默认是500条 |
- Kafka-Eagle监控
Kafka-Eagle框架可以监控Kafka集群的整体运行情况，在生产环境中经常使用。
- MySQL环境准备
Kafka-Eagle的安装依赖于MySQL，MySQL主要用来存储可视化展示的数据。如果集群中之前安装过MySQL可以跨过该步。
- Kafka环境准备关闭Kafka集群
[Angelvn@hadoop102 kafka]$ kf.sh stop- 修改/opt/module/kafka/bin/kafka-server-start.sh命令中
[Angelvn@hadoop102 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注意：修改之后在启动Kafka之前要分发之其他节点。
[Angelvn@hadoop102 bin]$ xsync kafka-server-start.sh- Kafka-Eagle安装官网：https://www.kafka-eagle.org/上传压缩包kafka-eagle-bin-2.0.8.tar.gz到集群/opt/software目录解压到本地
[Angelvn@hadoop102 software]$ tar -zxvf kafka-eagle-bin-2.0.8.tar.gz- 进入刚才解压的目录
[Angelvn@hadoop102 kafka-eagle-bin-2.0.8]$ ll
总用量 79164
-rw-rw-r--. 1 Angelvn Angelvn 81062577 10月 13 00:00 efak-web-2.0.8-bin.tar.gz- 将efak-web-2.0.8-bin.tar.gz解压至/opt/module
[Angelvn@hadoop102 kafka-eagle-bin-2.0.8]$ tar -zxvf efak-web-2.0.8-bin.tar.gz -C /opt/module/- 修改名称
[Angelvn@hadoop102 module]$ mv efak-web-2.0.8/ efak- 修改配置文件 /opt/module/efak/conf/system-config.properties
[Angelvn@hadoop102 conf]$ 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=hadoop102:2181,hadoop103:2181,hadoop104: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://hadoop102: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- 添加环境变量
[Angelvn@hadoop102 conf]$ sudo vim /etc/profile.d/my_env.sh
# kafkaEFAK
export KE_HOME=/opt/module/efak
export PATH=$PATH:$KE_HOME/bin注意：source /etc/profile
[Angelvn@hadoop102 conf]$ source /etc/profile- 启动注意：启动之前需要先启动ZK以及KAFKA。
[Angelvn@hadoop102 kafka]$ kf.sh start- 启动efak
[Angelvn@hadoop102 efak]$ bin/ke.sh start
Version 2.0.8 -- Copyright 2016-2021
*****************************************************************
* EFAK Service has started success.
* Welcome, Now you can visit 'http://192.168.10.102:8048'
* Account:admin ,Password:123456
*****************************************************************
* <Usage> ke.sh [start|status|stop|restart|stats] </Usage>
* <Usage> https://www.kafka-eagle.org/ </Usage>
*****************************************************************说明：如果停止efak，执行命令。
[Angelvn@hadoop102 efak]$ bin/ke.sh stop
- Kafka-Eagle页面操作
登录页面查看监控数据
- Kafka-Kraft模式
- Kafka-Kraft架构
左图为Kafka现有架构，元数据在zookeeper中，运行时动态选举controller，由controller进行Kafka集群管理。右图为kraft模式架构（实验性），不再依赖zookeeper集群，而是用三台controller节点代替zookeeper，元数据保存在controller中，由controller直接进行Kafka集群管理。
这样做的好处有以下几个：
- Kafka不再依赖外部框架，而是能够独立运行；
- controller管理集群时，不再需要从zookeeper中先读取数据，集群性能上升；
- 由于不依赖zookeeper，集群扩展时不再受到zookeeper读写能力限制；
- controller不再动态选举，而是由配置文件规定。这样我们可以有针对性的加强controller节点的配置，而不是像以前一样对随机controller节点的高负载束手无策。
- 7.2 Kafka-Kraft集群部署再次解压一份kafka安装包
[Angelvn@hadoop102 software]$ tar -zxvf kafka_2.12-3.6.1.tgz -C /opt/module/- 重命名为kafka2
[Angelvn@hadoop102 module]$ mv kafka_2.12-3.6.1/ kafka2- 在hadoop102上修改/opt/module/kafka2/config/kraft/server.properties配置文件
[Angelvn@hadoop102 kraft]$ vim server.properties
#kafka的角色（controller相当于主机、broker节点相当于从机，主机类似zk功能）
process.roles=broker, controller
#节点ID
node.id=2
#controller服务协议别名
controller.listener.names=CONTROLLER
#全Controller列表
controller.quorum.voters=2@hadoop102:9093,3@hadoop103:9093,4@hadoop104:9093
#不同服务器绑定的端口
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
#broker服务协议别名
inter.broker.listener.name=PLAINTEXT
#broker对外暴露的地址
advertised.Listeners=PLAINTEXT://hadoop102:9092
#协议别名到安全协议的映射
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
#kafka数据存储目录
log.dirs=/opt/module/kafka/kafka_2.12-3.6.1/datas- 分发kafka2
[Angelvn@hadoop102 module]$ xsync kafka2/在hadoop103和hadoop104上需要对node.id相应改变，值需要和controller.quorum.voters对应。
在hadoop103和hadoop104上需要根据各自的主机名称，修改相应的advertised.Listeners地址。
- 初始化集群数据目录首先生成存储目录唯一ID。
[Angelvn@hadoop102 kafka2]$ bin/kafka-storage.sh random-uuid
J7s9e8PPTKOO47PxzI39VA- 用该ID格式化kafka存储目录（三台节点）。
[Angelvn@hadoop102 kafka2]$ bin/kafka-storage.sh format -t J7s9e8PPTKOO47PxzI39VA -c /opt/module/kafka2/config/kraft/server.properties
[Angelvn@hadoop103 kafka2]$ bin/kafka-storage.sh format -t J7s9e8PPTKOO47PxzI39VA -c /opt/module/kafka2/config/kraft/server.properties
[Angelvn@hadoop104 kafka2]$ bin/kafka-storage.sh format -t J7s9e8PPTKOO47PxzI39VA -c /opt/module/kafka2/config/kraft/server.properties- 启动kafka集群
[Angelvn@hadoop102 kafka2]$ bin/kafka-server-start.sh -daemon config/kraft/server.properties
[Angelvn@hadoop103 kafka2]$ bin/kafka-server-start.sh -daemon config/kraft/server.properties
[Angelvn@hadoop104 kafka2]$ bin/kafka-server-start.sh -daemon config/kraft/server.properties- 停止kafka集群
[Angelvn@hadoop102 kafka2]$ bin/kafka-server-stop.sh
[Angelvn@hadoop103 kafka2]$ bin/kafka-server-stop.sh
[Angelvn@hadoop104 kafka2]$ bin/kafka-server-stop.sh- Kafka-Kraft集群启动停止脚本在/home/Angelvn/bin目录下创建文件kf2.sh脚本文件
[Angelvn@hadoop102 bin]$ vim kf2.sh脚本如下：
#! /bin/bash
case $1 in
"start"){
for i in hadoop102 hadoop103 hadoop104
do
echo " --------启动 $i Kafka2-------"
ssh $i "/opt/module/kafka2/bin/ .sh -daemon /opt/module/kafka2/config/kraft/server.properties"
done
};;
"stop"){
for i in hadoop102 hadoop103 hadoop104
do
echo " --------停止 $i Kafka2-------"
ssh $i "/opt/module/kafka2/bin/kafka-server-stop.sh "
done
};;
esac- 添加执行权限
[Angelvn@hadoop102 bin]$ chmod +x kf2.sh- 启动集群命令
[Angelvn@hadoop102 ~]$ kf2.sh start- 停止集群命令
[Angelvn@hadoop102 ~]$ kf2.sh stop
📮评论