一、Kafka概述和入门

1、Kafka概述

1.1 定义

Kafka是 一个开源的 分布式事件流平台 （Event StreamingPlatform），被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用。发布/订阅：消息的发布者不会将消息直接发送给特定的订阅者，而是将发布的消息分为不同的类别，订阅者只接收感兴趣的消息。

1.2 消息队列

传统的消息队列的主要应用场景包括：缓存/消峰、解耦和异步通信。消息队列的两种模式：

• 点对点模式，消费者主动拉取数据，消息收到后清除消息
• 发布/订阅模式，可以有多个topic主题（浏览、点赞、收藏、评论等）；消费者消费数据之后，不删除数据；每个消费者相互独立，都可以消费到数据

1.3 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

2、Kafka 快速入门

2.1 安装部署

官网地址：https://kafka.apache.org/downloads.html

# 实验配置环境是三台机器分别是hadoop102,103和104，每台机器需要安装zk和kafka,zk和xsync脚本参考之前笔记# 这个是scala2.12，然后kafka是3.0版本
wget https://archive.apache.org/dist/kafka/3.0.0/kafka_2.12-3.0.0.tgztar -zxvf kafka_2.12-3.0.0.tgz -C /opt/module/
cd /opt/module
mv kafka_2.12-3.0.0/ kafka
# 进入到/opt/module/kafka 目录，修改配置文件
cd config/
vim server.properties# 输入以下内容，三个地方需要修改，id，log地址和zk地址
#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# 然后分发安装包，在module路径
xsync kafka/
# 分别在 hadoop103 和 hadoop104 上修改配置文件/opt/module/kafka/config/server.properties中的 broker.id=1、broker.id=2# 最后配置环境变量
sudo vim /etc/profile.d/my_env.sh
#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin# 刷新一下环境变量
source /etc/profile# 分发环境变量文件到其他节点，并 source
sudo /home/atguigu/bin/xsync /etc/profile.d/my_env.sh# 启动集群
# 先启动 Zookeeper 集群，然后启动 Kafka，zk脚本zk学习笔记有记录
zk.sh start
# 依次在 hadoop102、hadoop103、hadoop104 节点上启动 Kafka
# 注意：配置文件的路径要能够到 server.properties
bin/kafka-server-start.sh -daemon config/server.properties
# 关闭集群
bin/kafka-server-stop.sh

2.2 集群启停脚本

在/home/atguigu/bin 目录下创建文件 kf.sh 脚本文件

#! /bin/bash
case $1 in
"start"){for i in hadoop102 hadoop103 hadoop104doecho " --------启动 $i Kafka-------"ssh $i "/opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties"done
};;
"stop"){for i in hadoop102 hadoop103 hadoop104doecho " --------停止 $i Kafka-------"ssh $i "/opt/module/kafka/bin/kafka-server-stop.sh "done
};;
esac

# 添加执行权限
chmod +x kf.sh
# 启动集群命令
kf.sh start
# 停止集群命令
kf.sh stop# 注意：停止 Kafka 集群时，一定要等 Kafka 所有节点进程全部停止后再停止 Zookeeper集群。
# 因为 Zookeeper 集群当中记录着 Kafka 集群相关信息，Zookeeper 集群一旦先停止，Kafka 集群就没有办法再获取停止进程的信息，只能手动杀死 Kafka 进程了

3、Kafka 命令行操作

3.1 Topic命令行操作

参数	描述
–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>	更新系统默认的配置

bin/kafka-topics.sh
# 查看当前服务器中的所有 topic
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --list
# 创建 first topic
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 1 --replication-factor 3 --topic first
# 查看 first 主题的详情
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first
# 修改分区数（注意：分区数只能增加，不能减少）
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 3
# 再次查看 first 主题的详情
# 删除 topic
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --delete --topic first

3.2 生产者命令行操作

参数	描述
–bootstrap-server <String: server toconnect to>	连接的 Kafka Broker 主机名称和端口号
–topic <String: topic>	操作的 topic 名称

bin/kafka-console-producer.sh
# 发送消息
bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first

3.3 消费者命令行操作

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

bin/kafka-console-consumer.sh
# 消费 first 主题中的数据
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
# 把主题中所有的数据都读取出来（包括历史数据）
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --from-beginning --topic first

二、Kafka核心概念详解

中文文档：https://kafka.apachecn.org/documentation.html
kafka参数参考：Kafka配置参数详解

1、Kafka 生产者

1.1 生产者消息发送流程

在消息发送的过程中，涉及到了两个线程——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

1.2 Kafka操作 API

首先导入依赖

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

普通异步发送

public static void main(String[] args) {// 0 配置Properties properties = new Properties();// 连接集群 bootstrap.serversproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 指定对应的key和value的序列化类型 key.serializerproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());// 1 创建kafka生产者对象// "" helloKafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);// 2 发送数据for (int i = 0; i < 5; i++) {kafkaProducer.send(new ProducerRecord<>("first","atguigu"+i));}// 3 关闭资源kafkaProducer.close();
}// 在 hadoop102 上开启 Kafka 消费者
// bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

带回调函数的异步发送

/**
回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是元
数据信息（RecordMetadata）和异常信息（Exception），如果 Exception 为 null，说明消息发
送成功，如果 Exception 不为 null，说明消息发送失败
*/public static void main(String[] args) throws InterruptedException {// 0 配置Properties properties = new Properties();// 连接集群 bootstrap.serversproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 指定对应的key和value的序列化类型 key.serializer
//        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());// 1 创建kafka生产者对象// "" helloKafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);// 2 发送数据for (int i = 0; i < 500; i++) {kafkaProducer.send(new ProducerRecord<>("first", "atguigu" + i), new Callback() {@Overridepublic void onCompletion(RecordMetadata metadata, Exception exception) {if (exception == null){System.out.println("主题： "+metadata.topic() + " 分区： "+ metadata.partition());}}});Thread.sleep(2);}// 3 关闭资源kafkaProducer.close();
}

同步发送 API

// 未处理前会阻塞
kafkaProducer.send(new ProducerRecord<>("first","atguigu"+i)).get();

1.3 生产者分区

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

自定义分区器

public class MyPartitioner implements Partitioner {@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {// 获取数据 atguigu  helloString msgValues = value.toString();int partition;if (msgValues.contains("atguigu")){partition = 0;}else {partition = 1;}return partition;}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> configs) {}
}// 关联自定义分区器,使用全类名
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.atguigu.kafka.producer.MyPartitioner");
kafkaProducer.send(new ProducerRecord<>("first","hello" + i), new Callback() {});

1.4 提高吞吐量

//批次大小和等待时间二者只要有一个满足就会发送
public static void main(String[] args) {// 0 配置, 创建 kafka 生产者的配置对象Properties properties = new Properties();// 连接kafka集群,给 kafka 配置对象添加配置信息：bootstrap.serversproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 序列化properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());// 缓冲区大小，默认 32M：buffer.memoryproperties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);// 批次大小,，默认 16Kproperties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);// linger.ms,等待时间默认 0properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);// 压缩，默认 none，可配置值 gzip、snappy、lz4 和 zstdproperties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");// 1 创建生产者KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);// 2 发送数据for (int i = 0; i < 5; i++) {kafkaProducer.send(new ProducerRecord<>("first","atguigu"+i));}// 3 关闭资源kafkaProducer.close();
}

1.5 数据可靠性

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

数据完全可靠条件 = ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

可靠性总结：

• acks=0，生产者发送过来数据就不管了，可靠性差，效率高；
• acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等；
• acks=-1，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低；
• 在生产环境中，acks=0很少使用；acks=1，一般用于传输普通日志，允许丢个别数据；acks=-1，一般用于传输和钱相关的数据，对可靠性要求比较高的场景。但是acks=-1时，有重复消费的可能

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

1.6 数据去重

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

幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条(必须按照顺序进行落盘)，保证了不重复。重复数据的判断标准：具有**<PID, Partition, SeqNumber>相同主键的消息提交时，Broker只会持久化一条。其中PID是Kafka每次重启都会分配一个新的；Partition 表示分区号；Sequence Number是单调自增的**。所以幂等性只能保证的是在单分区单会话内不重复。开启幂等性：开启参数 enable.idempotence 默认为 true，false 关闭

事务原理

// 共5个API
// 1 初始化事务
void initTransactions();
// 2 开启事务
void beginTransaction() throws ProducerFencedException;
// 3 在事务内提交已经消费的偏移量（主要用于消费者）
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,String consumerGroupId) throws 
ProducerFencedException;
// 4 提交事务
void commitTransaction() throws ProducerFencedException;
// 5 放弃事务（类似于回滚事务的操作）
void abortTransaction() throws ProducerFencedException;// 举例
public static void main(String[] args) {// 0 配置Properties properties = new Properties();// 连接集群 bootstrap.serversproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");// 指定对应的key和value的序列化类型 key.serializer
//        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 指定事务idproperties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tranactional_id_01");// 1 创建kafka生产者对象// "" helloKafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);kafkaProducer.initTransactions();kafkaProducer.beginTransaction();try {// 2 发送数据for (int i = 0; i < 5; i++) {kafkaProducer.send(new ProducerRecord<>("first", "atguigu" + i));}int i = 1 / 0;kafkaProducer.commitTransaction();} catch (Exception e) {kafkaProducer.abortTransaction();} finally {// 3 关闭资源kafkaProducer.close();}
}

1.7 数据顺序

对于单分区来说，数据是必定有序的；多分区，分区与分区间无序

2、Kafka Broker

2.1 Kafka Broker 工作流程

bin/zkCli.sh查看zookeeper里面的文件，也可以下载PrettyZoo(推荐)可视化软件查看Zookeeper 存储的 Kafka 信息

Kafka Broker 总体工作流程

# 查看/kafka/brokers/ids 路径上的节点,首先进入了zk客户端
ls /kafka/brokers/ids
# 查看/kafka/controller 路径上的数据
get /kafka/controller
# 查看/kafka/brokers/topics/first/partitions/0/state 路径上的数据
get /kafka/brokers/topics/first/partitions/0/state
# 停止 hadoop104 上的 kafka
bin/kafka-server-stop.sh
# 然后去查看有没有变化
bin/kafka-server-start.sh -daemon ./config/server.properties

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。一般不建议修改，交给系统自己管理

2.2 节点服役和退役

服役新节点

# 首先配置好一台新的机器，例如hadoop105
# 单独启动 hadoop105 中的 kafka
bin/kafka-server-start.sh -daemon ./config/server.properties
# 执行负载均衡操作
# 创建一个要均衡的topic,在hadoop102
vim topics-to-move.json
{
"topics": [{"topic": "first"}
],"version": 1
}# 生成一个负载均衡的计划,会成列出几个计划，自己选一个计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate
# 创建副本存储计划（所有副本存储在 broker0、broker1、broker2、broker3 中），从上面的计划中选取
vim increase-replication-factor.json
# 执行副本存储计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute
# 验证副本存储计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify

退役旧节点

# 先按照退役一台节点，生成执行计划，然后按照服役时操作流程执行负载均衡
vim topics-to-move.json
{
"topics": [{"topic": "first"}
],"version": 1
}# 创建执行计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2" --generate
# 创建副本存储计划（所有副本存储在 broker0、broker1、broker2 中）,从上面选
vim increase-replication-factor.json
# 执行副本存储计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute
# 验证副本存储计划
# 查看一下信息
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first
# 执行停止命令
bin/kafka-server-stop.sh

2.3 Kafka 副本基本信息

• 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 副本同步时，延迟过多的副本。

Leader 选举流程

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

# 创建一个新的 topic，4 个分区，4 个副本
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic atguigu1 --partitions 4 --replication-factor 4
# 查看 Leader 分布情况
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu1
# 停止掉 hadoop105 的 kafka 进程，并查看 Leader 分区情况
bin/kafka-server-stop.sh
# 停止掉 hadoop104 的 kafka 进程，并查看 Leader 分区情况
# 启动 hadoop105 的 kafka 进程，并查看 Leader 分区情况
bin/kafka-server-start.sh -daemon config/server.properties
# 其他也可以自己实验

2.4 Leader 和 Follower 故障处理细节

2.5 分区副本分配与负载平衡

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

# 创建一个新的topic，4个分区，两个副本，名称为three。将 该topic的所有副本都存储到broker0和broker1两台服务器上
# 创建一个新的 topic，名称为 three
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 4 --replication-factor 2 --topic three
# 查看分区副本存储情况
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic three
# 创建副本存储计划（所有副本都指定存储在 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]}]
}# 执行副本存储计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute
# 验证副本存储计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify
# 查看分区副本存储情况
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic three

参数名称	描述
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 负载是否平衡的间隔时间

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

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 3 --replication-factor 1 --topic four
# 手动增加副本存储
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]}]
}# 执行副本存储计划
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute

2.6 文件存储

# 思考：Topic 数据到底存储在什么位置
# 查看 hadoop102（或者 hadoop103、hadoop104）的/opt/module/kafka/datas/first-1（first-0、first-2）路径上的文件
# 直接查看 log 日志，发现是乱码
# 通过工具查看 index 和 log 信息
kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.index
kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.log# 说明：日志存储参数配置
# log.segment.bytesKafka 中 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，表示无穷大
• compact 日志压缩：**对于相同key的不同value值，只保留最后一个版本，**log.cleanup.policy = compact 所有数据启用压缩策略
  • 压缩后的offset可能是不连续的，当从这些offset消费消息时，将会拿到比这个offset大的offset对应的消息，实际上会拿到offset为7的消息，并从这个位置开始消费
  • 这种策略只适合特殊场景，比如消息的key是用户ID，value是用户的资料，通过这种压缩策略，整个消息集里就保存了所有用户最新的资料

2.7 高效读写数据(💥)

• 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。一般不建议修改，交给系统自己管理。

2.8 自动创建主题

如果 broker 端配置参数 auto.create.topics.enable 设置为 true（默认值是 true），那么当生产者向一个未创建的主题发送消息时，会自动创建一个分区数为 num.partitions（默认值为1）、副本因子为 default.replication.factor（默认值为 1）的主题。除此之外，当一个消费者开始从未知主题中读取消息时，或者当任意一个客户端向未知主题发送元数据请求时，都会自动创建一个相应主题。这种创建主题的方式是非预期的，增加了主题管理和维护的难度。

生产环境建议将该参数设置为 false。

3、Kafka 消费者(重点)

3.1 Kafka 消费方式

• pull（拉）模 式

  consumer采用从broker中主动拉取数据。Kafka采用这种方式

• push（推）模式

  Kafka没有采用这种方式，因为由broker 决定消息发送速率，很难适应所有消费者的消费速率。例如推送的速度是50m/s， Consumer1、Consumer2就来不及处理消息。pull模式不足之处是，如果Kafka没有数据，消费者可能会陷入循环中，一直返回空数据

3.2 Kafka 消费者工作流程

Consumer Group（CG）：消费者组，由多个consumer组成。形成一个消费者组的条件，是所有消费者的groupid相同

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

参数名称	描述
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 条。

3.3 消费者API

public static void main(String[] args) {// 0 配置Properties properties = new Properties();// 连接 bootstrap.serversproperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 配置消费者组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,"test5");// 设置分区分配策略properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.StickyAssignor");// 1 创建一个消费者  "", "hello"KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);// 2 订阅主题 firstArrayList<String> topics = new ArrayList<>();topics.add("first");kafkaConsumer.subscribe(topics);// 2 订阅主题对应的分区//ArrayList<TopicPartition> topicPartitions = new ArrayList<>();//topicPartitions.add(new TopicPartition("first",0));//kafkaConsumer.assign(topicPartitions);// 消费者组把组id配置成同一个即可，这样就会处理自己监听的分区了// 3 消费数据while (true){ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord);}kafkaConsumer.commitAsync();}
}// bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first

3.4 分区的分配以及再平衡

具体详见官网:https://kafka.apache.org/documentation/#consumerapi

参数名称	描述
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 以及再平衡**

RoundRobin 以及再平衡

Sticky 以及再平衡

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

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

3.5 offset 位移

从0.9版本开始，consumer默认将offset保存在Kafka一个内置的topic中，该topic为__consumer_offsets。__consumer_offsets 主题里面采用 key 和 value 的方式存储数据。key 是 group.id+topic+分区号，value 就是当前 offset 的值。每隔一段时间，kafka 内部会对这个 topic 进行compact，也就是每个 group.id+topic+分区号就保留最新数据

# 在配置文件 config/consumer.properties 中添加配置 exclude.internal.topics=false，默认是 true，表示不能消费系统主题。为了查看该系统主题数据，所以该参数修改为 false
# 改完后进行分发
# 采用命令行方式，创建一个新的 topic
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic atguigu --partitions 2 --replication-factor 2
# 启动生产者往 atguigu 生产数据
bin/kafka-console-producer.sh --topic atguigu --bootstrap-server hadoop102:9092
# 启动消费者消费 atguigu 数据
# 注意：指定消费者组名称，更好观察数据存储位置（key 是 group.id+topic+分区号）
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic atguigu --group test
# 查看消费者消费主题__consumer_offsets
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

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

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

// 自动提交
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);// 提交时间间隔
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000);

手动提交 offset

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

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

public static void main(String[] args) {// 0 配置Properties properties = new Properties();// 连接 bootstrap.serversproperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 配置消费者组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");// 手动提交properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);// 1 创建一个消费者  "", "hello"KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);// 2 订阅主题 firstArrayList<String> topics = new ArrayList<>();topics.add("first");kafkaConsumer.subscribe(topics);// 3 消费数据while (true){ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord);}// 手动提交offsetkafkaConsumer.commitSync();kafkaConsumer.commitAsync();}
}

3.6 offset其他情况

auto.offset.reset = earliest | latest | none 默认是 latest。当 Kafka 中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量时（例如该数据已被删除），该怎么办？

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

// 指定位置进行消费
Set<TopicPartition> assignment = kafkaConsumer.assignment();//  保证分区分配方案已经制定完毕
while (assignment.size() == 0){kafkaConsumer.poll(Duration.ofSeconds(1));assignment = kafkaConsumer.assignment();
}// 指定消费的offset
for (TopicPartition topicPartition : assignment) {kafkaConsumer.seek(topicPartition,600);
}

指定时间消费

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

public static void main(String[] args) {// 0 配置信息Properties properties = new Properties();// 连接properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,"test3");// 1 创建消费者KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);// 2 订阅主题ArrayList<String> topics = new ArrayList<>();topics.add("first");kafkaConsumer.subscribe(topics);// 指定位置进行消费Set<TopicPartition> assignment = kafkaConsumer.assignment();//  保证分区分配方案已经制定完毕while (assignment.size() == 0){kafkaConsumer.poll(Duration.ofSeconds(1));assignment = kafkaConsumer.assignment();}// 希望把时间转换为对应的offsetHashMap<TopicPartition, Long> topicPartitionLongHashMap = new HashMap<>();// 封装对应集合for (TopicPartition topicPartition : assignment) {topicPartitionLongHashMap.put(topicPartition,System.currentTimeMillis() - 1 * 24 * 3600 * 1000);}Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap = kafkaConsumer.offsetsForTimes(topicPartitionLongHashMap);// 指定消费的offsetfor (TopicPartition topicPartition : assignment) {OffsetAndTimestamp offsetAndTimestamp = topicPartitionOffsetAndTimestampMap.get(topicPartition);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);}}
}

漏消费和重复消费

3.7 消费者事务

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

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

• 如果是Kafka消费能力不足，则可以考虑增加Topic的分区数，并且同时提升消费组的消费者数量，消费者数 = 分区数。（两者缺一不可）
• 如果是下游的数据处理不及时：提高每批次拉取的数量。批次拉取数据过少（拉取数据/处理时间 < 生产速度），使处理的数据小于生产的数据，也会造成数据积压

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

4、Kafka-Eagle 监控

4.1 概述与环境准备

Kafka-Eagle 框架可以监控 Kafka 集群的整体运行情况，在生产环境中经常使用。Kafka-Eagle 的安装依赖于 MySQL，MySQL 主要用来存储可视化展示的数据。Mysql安装可以参考之前hive的学习笔记

4.2 Kafka 环境准备

# 关闭 Kafka 集群
kf.sh stop# 修改/opt/module/kafka/bin/kafka-server-start.sh 命令中
vim bin/kafka-server-start.sh
# 修改如下参数值
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; thenexport KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
fi
# 为
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; thenexport 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 之前要分发之其他节点
xsync kafka-server-start.sh
kf.sh start

4.3 Kafka-Eagle 安装

官网：https://www.kafka-eagle.org/

wget https://github.com/smartloli/kafka-eagle-bin/archive/v3.0.1.tar.gz
tar -zxvf v3.0.1.tar.gz
cd kafka-eagle-bin-3.0.1/
tar -zxvf efak-web-3.0.1-bin.tar.gz -C /opt/module/
mv efak-web-3.0.1/ efak
# 进入后修改配置
vim system-config.properties
# 自己按需修改，主要修改三个地方，这是单机单kafka集群
######################################
efak.zk.cluster.alias=cluster1
cluster1.zk.list=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka
#cluster2.zk.list=xdn10:2181,xdn11:2181,xdn12:2181
# kafka offset storage
######################################
cluster1.efak.offset.storage=kafka
#cluster2.efak.offset.storage=zk
# 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# 添加环境变量
sudo vim /etc/profile.d/my_env.sh
# kafkaEFAK
export KE_HOME=/opt/module/efak
export PATH=$PATH:$KE_HOME/bin
# 注意：刷新环境变量
source /etc/profile# 启动，启动之前需要先启动 ZK 以及 KAFKA
kf.sh start
bin/ke.sh start
# 说明：如果停止 efak，执行命令
bin/ke.sh stop# http://hadoop102:8048/ 访问
# 用户名密码admin/123456

5、Kafka-Kraft 模式

5.1 Kafka-Kraft 架构

左图为 Kafka 现有架构，元数据在 zookeeper 中，运行时动态选举 controller，由controller 进行 Kafka 集群管理。右图为 kraft 模式架构（实验性），不再依赖zookeeper 集群，而是用三台 controller 节点代替 zookeeper，元数据保存在 controller 中，由 controller 直接进行 Kafka 集群管理。这样做的好处有以下几个：

• Kafka 不再依赖外部框架，而是能够独立运行；
• controller 管理集群时，不再需要从 zookeeper 中先读取数据，集群性能上升；
• 由于不依赖 zookeeper，集群扩展时不再受到 zookeeper 读写能力限制；
• controller 不再动态选举，而是由配置文件规定。这样我们可以有针对性的加强controller 节点的配置，而不是像以前一样对随机 controller 节点的高负载束手无策

5.2 Kafka-Kraft 集群部署

# 再次解压一份 kafka 安装包
tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/module/
mv kafka_2.12-3.0.0/ kafka2# 在 hadoop102 上修改/opt/module/kafka2/config/kraft/server.properties 配置文件
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/kafka2/data# 分发 kafka2
xsync kafka2/
# 在 hadoop103 和 hadoop104 上 需 要 对 node.id 相应改变 ， 值 需 要 和controller.quorum.voters 对应
# 在 hadoop103 和 hadoop104 上需要 根据各自的主机名称，修改相应的advertised.Listeners 地址# 初始化集群数据目录
# 首先生成存储目录唯一 ID
bin/kafka-storage.sh random-uuid
# 用该 ID 格式化 kafka 存储目录（三台节点），随机数是上面生产的
bin/kafka-storage.sh format -t J7s9e8PPTKOO47PxzI39VA -c /opt/module/kafka2/config/kraft/server.properties
# 启动 kafka 集群，三台都要启动
bin/kafka-server-start.sh -daemon config/kraft/server.properties# 停止 kafka 集群
bin/kafka-server-stop.sh

5.3 Kafka-Kraft 集群启动停止脚本

在/home/atguigu/bin 目录下创建文件 kf2.sh 脚本文件

#! /bin/bash
case $1 in
"start"){for i in hadoop102 hadoop103 hadoop104doecho " --------启动 $i Kafka2-------"ssh $i "/opt/module/kafka2/bin/kafka-server-start.sh -daemon /opt/module/kafka2/config/kraft/server.properties"done
};;
"stop"){
for i in hadoop102 hadoop103 hadoop104doecho " --------停止 $i Kafka2-------"ssh $i "/opt/module/kafka2/bin/kafka-server-stop.sh "done
};;
esac

# 添加执行权限
chmod +x kf2.sh
kf2.sh start
kf2.sh stop

6、Kafka配置文件说明

6.1 Server.properties配置文件说明

#broker的全局唯一编号，不能重复
broker.id=0#用来监听链接的端口，producer或consumer将在此端口建立连接
port=9092#处理网络请求的线程数量
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运行日志存放的路径
log.dirs=/export/data/kafka/#topic在当前broker上的分片个数
num.partitions=2#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1#segment文件保留的最长时间，超时将被删除
log.retention.hours=168#滚动生成新的segment文件的最大时间
log.roll.hours=1#日志文件中每个segment的大小，默认为1G
log.segment.bytes=1073741824#周期性检查文件大小的时间
log.retention.check.interval.ms=300000#日志清理是否打开
log.cleaner.enable=true#broker需要使用zookeeper保存meta数据
zookeeper.connect=zk01:2181,zk02:2181,zk03:2181#zookeeper链接超时时间
zookeeper.connection.timeout.ms=6000#partion buffer中，消息的条数达到阈值，将触发flush到磁盘
log.flush.interval.messages=10000#消息buffer的时间，达到阈值，将触发flush到磁盘
log.flush.interval.ms=3000#删除topic需要server.properties中设置delete.topic.enable=true否则只是标记删除
delete.topic.enable=true#此处的host.name为本机IP(重要),如果不改,则客户端会抛出:Producer connection to localhost:9092 unsuccessful 错误!
host.name=kafka01advertised.host.name=192.168.140.128# producer生产者配置文件说明
#指定kafka节点列表，用于获取metadata，不必全部指定
metadata.broker.list=node01:9092,node02:9092,node03:9092
# 指定分区处理类。默认kafka.producer.DefaultPartitioner，表通过key哈希到对应分区
#partitioner.class=kafka.producer.DefaultPartitioner
# 是否压缩，默认0表示不压缩，1表示用gzip压缩，2表示用snappy压缩。压缩后消息中会有头来指明消息压缩类型，故在消费者端消息解压是透明的无需指定。
compression.codec=none
# 指定序列化处理类
serializer.class=kafka.serializer.DefaultEncoder
# 如果要压缩消息，这里指定哪些topic要压缩消息，默认empty，表示不压缩。
#compressed.topics=# 设置发送数据是否需要服务端的反馈,有三个值0,1,-1
# 0: producer不会等待broker发送ack 
# 1: 当leader接收到消息之后发送ack 
# -1: 当所有的follower都同步消息成功后发送ack. 
request.required.acks=0 # 在向producer发送ack之前,broker允许等待的最大时间 ，如果超时,broker将会向producer发送一个error ACK.意味着上一次消息因为某种原因未能成功(比如follower未能同步成功) 
request.timeout.ms=10000# 同步还是异步发送消息，默认“sync”表同步，"async"表异步。异步可以提高发送吞吐量,
# 也意味着消息将会在本地buffer中,并适时批量发送，但是也可能导致丢失未发送过去的消息
producer.type=sync# 在async模式下,当message被缓存的时间超过此值后,将会批量发送给broker,默认为5000ms
# 此值和batch.num.messages协同工作.
queue.buffering.max.ms = 5000# 在async模式下,producer端允许buffer的最大消息量
# 无论如何,producer都无法尽快的将消息发送给broker,从而导致消息在producer端大量沉积
# 此时,如果消息的条数达到阀值,将会导致producer端阻塞或者消息被抛弃，默认为10000
queue.buffering.max.messages=20000# 如果是异步，指定每次批量发送数据量，默认为200
batch.num.messages=500# 当消息在producer端沉积的条数达到"queue.buffering.max.meesages"后 
# 阻塞一定时间后,队列仍然没有enqueue(producer仍然没有发送出任何消息) 
# 此时producer可以继续阻塞或者将消息抛弃,此timeout值用于控制"阻塞"的时间 
# -1: 无阻塞超时限制,消息不会被抛弃 
# 0:立即清空队列,消息被抛弃 
queue.enqueue.timeout.ms=-1# 当producer接收到error ACK,或者没有接收到ACK时,允许消息重发的次数 
# 因为broker并没有完整的机制来避免消息重复,所以当网络异常时(比如ACK丢失) 
# 有可能导致broker接收到重复的消息,默认值为3.
message.send.max.retries=3# producer刷新topic metada的时间间隔,producer需要知道partition leader的位置,以及当前topic的情况 
# 因此producer需要一个机制来获取最新的metadata,当producer遇到特定错误时,将会立即刷新 
# (比如topic失效,partition丢失,leader失效等),此外也可以通过此参数来配置额外的刷新机制，默认值600000 
topic.metadata.refresh.interval.ms=60000

6.2 consumer消费者配置详细说明

# zookeeper连接服务器地址
zookeeper.connect=zk01:2181,zk02:2181,zk03:2181
# zookeeper的session过期时间，默认5000ms，用于检测消费者是否挂掉
zookeeper.session.timeout.ms=5000
#当消费者挂掉，其他消费者要等该指定时间才能检查到并且触发重新负载均衡
zookeeper.connection.timeout.ms=10000
# 指定多久消费者更新offset到zookeeper中。注意offset更新时基于time而不是每次获得的消息。一旦在更新zookeeper发生异常并重启，将可能拿到已拿到过的消息
zookeeper.sync.time.ms=2000
#指定消费 
group.id=itcast
# 当consumer消费一定量的消息之后,将会自动向zookeeper提交offset信息 
# 注意offset信息并不是每消费一次消息就向zk提交一次,而是现在本地保存(内存),并定期提交,默认为true
auto.commit.enable=true
# 自动更新时间。默认60 * 1000
auto.commit.interval.ms=1000
# 当前consumer的标识,可以设定,也可以有系统生成,主要用来跟踪消息消费情况,便于观察
conusmer.id=xxx 
# 消费者客户端编号，用于区分不同客户端，默认客户端程序自动产生
client.id=xxxx
# 最大取多少块缓存到消费者(默认10)
queued.max.message.chunks=50
# 当有新的consumer加入到group时,将会reblance,此后将会有partitions的消费端迁移到新  的consumer上,如果一个consumer获得了某个partition的消费权限,那么它将会向zk注册 "Partition Owner registry"节点信息,但是有可能此时旧的consumer尚没有释放此节点, 此值用于控制,注册节点的重试次数. 
rebalance.max.retries=5# 获取消息的最大尺寸,broker不会像consumer输出大于此值的消息chunk 每次feth将得到多条消息,此值为总大小,提升此值,将会消耗更多的consumer端内存
fetch.min.bytes=6553600# 当消息的尺寸不足时,server阻塞的时间,如果超时,消息将立即发送给consumer
fetch.wait.max.ms=5000
socket.receive.buffer.bytes=655360
# 如果zookeeper没有offset值或offset值超出范围。那么就给个初始的offset。有smallest、largest、anything可选，分别表示给当前最小的offset、当前最大的offset、抛异常。默认largest
auto.offset.reset=smallest
# 指定序列化处理类
derializer.class=kafka.serializer.DefaultDecoder

三、Kafka外部系统集成

1、集成 Flume

1.1 Flume生产者

# 启动 kafka 集群
zk.sh start
kf.sh start
# 启动 kafka 消费者,这里我使用first的topic
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
# 安装好flume，具体详见之前的笔记Flume1.9基础学习# 开始配置flume生产者
# 在 hadoop102 节点的 Flume 的 job 目录下创建 file_to_kafka.conf
mkdir jobs
vim jobs/file_to_kafka.conf# 1 组件定义
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 2 配置 source
a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /opt/module/applog/app.*
a1.sources.r1.positionFile = /opt/module/flume/taildir_position.json
# 3 配置 channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 4 配置 sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092
a1.sinks.k1.kafka.topic = first
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
# 5 拼接组件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1# 启动 Flume
bin/flume-ng agent -c conf/ -n a1 -f jobs/file_to_kafka.conf &
# 向/opt/module/applog/app.log 里追加数据，查看 kafka 消费者消费情况
mkdir applog
echo hello >> /opt/module/applog/app.log
# 观察 kafka 消费者，能够看到消费的 hello 数据

1.2 Flume 消费者

# 拉取kafka消息到控制台
# 在 hadoop102 节点的 Flume 的/opt/module/flume/jobs 目录下创建 kafka_to_file.conf
vim kafka_to_file.conf# 1 组件定义
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 2 配置 source
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.batchSize = 50
a1.sources.r1.batchDurationMillis = 200
a1.sources.r1.kafka.bootstrap.servers = hadoop102:9092
a1.sources.r1.kafka.topics = first
a1.sources.r1.kafka.consumer.group.id = custom.g.id
# 3 配置 channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 4 配置 sink
a1.sinks.k1.type = logger
# 5 拼接组件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1# 启动 Flume
bin/flume-ng agent -c conf/ -n a1 -f jobs/kafka_to_file.conf -Dflume.root.logger=INFO,console
# 启动 kafka 生产者
bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first

2、集成 Flink

Flink 是一个在大数据开发中非常常用的组件。可以用于 Kafka 的生产者，也可以用于Flink 的消费者。首先创建maven项目

<dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.13.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.12</artifactId><version>1.13.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_2.12</artifactId><version>1.13.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka_2.12</artifactId><version>1.13.0</version></dependency>
</dependencies>

将 log4j.properties 文件添加到 resources 里面，就能更改打印日志的级别为 error

log4j.rootLogger=error, stdout,R
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %5p --- [%50t] %-80c(line:%5L) : %m%nlog4j.appender.R=org.apache.log4j.RollingFileAppender
log4j.appender.R.File=../log/agent.log
log4j.appender.R.MaxFileSize=1024KB
log4j.appender.R.MaxBackupIndex=1log4j.appender.R.layout=org.apache.log4j.PatternLayout
log4j.appender.R.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %5p --- [%50t] %-80c(line:%6L) : %m%n

在 java 文件夹下创建包名为 com.atguigu.flink，首先是测试flink生产者代码

public static void main(String[] args) throws Exception {// 0 初始化 flink 环境StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(3);// 1 读取集合中数据ArrayList<String> wordsList = new ArrayList<>();wordsList.add("hello");wordsList.add("world");DataStream<String> stream = env.fromCollection(wordsList);// 2 kafka 生产者配置信息Properties properties = new Properties();properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");// 3 创建 kafka 生产者FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>("first",new SimpleStringSchema(),properties);// 4 生产者和 flink 流关联stream.addSink(kafkaProducer);// 5 执行env.execute();
}// 启动 Kafka 消费者
// bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
// 执行 FlinkKafkaProducer1 程序，观察 kafka 消费者控制台情况

Flink 消费者

public static void main(String[] args) throws Exception {// 0 初始化 flink 环境StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(3);// 1 kafka 消费者配置信息Properties properties = new Properties();properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");// 2 创建 kafka 消费者FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("first",new SimpleStringSchema(),properties);// 3 消费者和 flink 流关联env.addSource(kafkaConsumer).print();// 4 执行env.execute();
}// 启动 FlinkKafkaConsumer1 消费者
// 启动 kafka 生产者
// bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first

3、集成 SpringBoot

首先是导入依赖

<!-- spring-kafka -->
<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId>
</dependency>

3.1 简单Demo

添加lombok插件，spring-boot-starter-web依赖以及spring-kafka依赖。首先修改 SpringBoot 核心配置文件 application.propeties, 添加生产者相关信息

# 应用名称
spring.application.name=atguigu_springboot_kafka
# 指定 kafka 的地址
spring.kafka.bootstrap-servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
#指定 key 和 value 的序列化器
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer# =========消费者配置开始=========
# 指定 kafka 的地址
# spring.kafka.bootstrap-servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
# 指定 key 和 value 的反序列化器
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#指定消费者组的 group_id
spring.kafka.consumer.group-id=atguigu
# =========消费者配置结束=========

创建 controller 从浏览器接收数据, 并写入指定的 topic

@RestController
public class ProducerController {// Kafka 模板用来向 kafka 发送数据@AutowiredKafkaTemplate<String, String> kafka;@RequestMapping("/atguigu")public String data(String msg) {kafka.send("first", msg);return "ok";}
}// 在浏览器中给/atguigu 接口发送数据

SpringBoot 消费者

@Configuration
public class KafkaConsumer {// 指定要监听的 topic@KafkaListener(topics = "first")public void consumeTopic(String msg) { // 参数: 收到的 valueSystem.out.println("收到的信息: " + msg);}
}// 向 first 主题发送数据
// bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first

3.2 完整Demo

这次是yml配置文件

# kafka 配置
spring:kafka:bootstrap-servers: localhost:9092producer:# 发生错误后，消息重发的次数。retries: 1#当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。batch-size: 16384# 设置生产者内存缓冲区的大小。buffer-memory: 33554432# 键的序列化方式key-serializer: org.apache.kafka.common.serialization.StringSerializer# 值的序列化方式value-serializer: org.apache.kafka.common.serialization.StringSerializer# acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。# acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。# acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。acks: 1consumer:# 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5Dauto-commit-interval: 1S# 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：# latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）# earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录auto-offset-reset: earliest# 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量enable-auto-commit: false# 键的反序列化方式key-deserializer: org.apache.kafka.common.serialization.StringDeserializer# 值的反序列化方式value-deserializer: org.apache.kafka.common.serialization.StringDeserializerlistener:# 在侦听器容器中运行的线程数。concurrency: 5# listner负责ack，每调用一次，就立即commitack-mode: manual_immediatemissing-topics-fatal: false

创建消息生产者

@Component
public class KafkaProducer {private Logger logger = LoggerFactory.getLogger(KafkaProducer.class);@Resourceprivate KafkaTemplate<String, Object> kafkaTemplate;public static final String TOPIC_TEST = "Hello-Kafka";public static final String TOPIC_GROUP = "test-consumer-group";public void send(Object obj) {String obj2String = JSON.toJSONString(obj);logger.info("准备发送消息为：{}", obj2String);// 发送消息ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_TEST, obj);future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {@Overridepublic void onFailure(Throwable throwable) {//发送失败的处理logger.info(TOPIC_TEST + " - 生产者 发送消息失败：" + throwable.getMessage());}@Overridepublic void onSuccess(SendResult<String, Object> stringObjectSendResult) {//成功的处理logger.info(TOPIC_TEST + " - 生产者 发送消息成功：" + stringObjectSendResult.toString());}});}}

创建消息消费者

@Component
public class KafkaConsumer {private Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);@KafkaListener(topics = KafkaProducer.TOPIC_TEST, groupId = KafkaProducer.TOPIC_GROUP)public void topicTest(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {Optional<?> message = Optional.ofNullable(record.value());if (message.isPresent()) { // 包含非空值，则执行Object msg = message.get();logger.info("topic_test 消费了： Topic:" + topic + ",Message:" + msg);ack.acknowledge(); // 确认成功消费一个消息}}}

发送测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class KafkaProducerTest {private Logger logger = LoggerFactory.getLogger(KafkaProducerTest.class);@Resourceprivate KafkaProducer kafkaProducer; // 注意使用自己创建的，看清楚！/*测试之前需要开启 Kafka 服务启动 Zookeeper：bin/zookeeper-server-start.sh -daemon config/zookeeper.properties启动 Kafka：bin/kafka-server-start.sh -daemon config/server.properties测试结果数据：准备发送消息为："你好，我是Lottery 001"Hello-Kafka - 生产者 发送消息成功：SendResult [producerRecord=ProducerRecord(topic=Hello-Kafka, partition=null,headers=RecordHeaders(headers = [], isReadOnly = true), key=null, value=你好，我是Lottery 001, timestamp=null),recordMetadata=Hello-Kafka-0@47]topic_test 消费了： Topic:Hello-Kafka,Message:你好，我是Lottery 001*/@Testpublic void test_send() throws InterruptedException {// 循环发送消息while (true) {kafkaProducer.send("你好，我是Lottery 001");Thread.sleep(3500);}}}

4、集成 Spark

创建一个 maven 项目 spark-kafka，在项目 spark-kafka 上点击右键，Add Framework Support=》勾选 scala，在 main 下创建 scala 文件夹，并右键 Mark Directory as Sources Root=>在 scala 下创建包名为 com.atguigu.spark，添加配置文件

<dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-10_2.12</artifactId><version>3.0.0</version></dependency>
</dependencies>

将 log4j.properties 文件添加到 resources 里面，就能更改打印日志的级别为 error，见上面

Spark 生产者

object SparkKafkaProducer {def main(args: Array[String]): Unit = {// 0 kafka 配置信息val properties = new Properties()properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092,hadoop104:9092")properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])// 1 创建 kafka 生产者var producer = new KafkaProducer[String, String](properties)// 2 发送数据for (i <- 1 to 5){producer.send(newProducerRecord[String,String]("first","atguigu" + i))}// 3 关闭资源producer.close()}
}// 启动 Kafka 消费者

Spark 消费者，添加配置文件

<dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-10_2.12</artifactId><version>3.0.0</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.12</artifactId><version>3.0.0</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_2.12</artifactId><version>3.0.0</version></dependency>
</dependencies>

object SparkKafkaConsumer {def main(args: Array[String]): Unit = {//1.创建 SparkConfval sparkConf: SparkConf = newSparkConf().setAppName("sparkstreaming").setMaster("local[*]")//2.创建 StreamingContextval ssc = new StreamingContext(sparkConf, Seconds(3))//3.定义 Kafka 参数：kafka 集群地址、消费者组名称、key 序列化、value 序列化val kafkaPara: Map[String, Object] = Map[String, Object](ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG ->"hadoop102:9092,hadoop103:9092,hadoop104:9092",ConsumerConfig.GROUP_ID_CONFIG -> "atguiguGroup",ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG ->classOf[StringDeserializer],ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG ->classOf[StringDeserializer])//4.读取 Kafka 数据创建 DStreamval kafkaDStream: InputDStream[ConsumerRecord[String, String]] =KafkaUtils.createDirectStream[String, String](ssc,LocationStrategies.PreferConsistent, //优先位置ConsumerStrategies.Subscribe[String, String](Set("first"), kafkaPara)// 消费策略：（订阅多个主题，配置参数）)//5.将每条消息的 KV 取出val valueDStream: DStream[String] = kafkaDStream.map(record => record.value())//6.计算 WordCountvalueDStream.print()//7.开启任务ssc.start()ssc.awaitTermination()}
}

四、Kafka生产调优

1、Kafka 硬件配置选择

1.1 场景说明

• 100 万日活，每人每天 100 条日志，每天总共的日志条数是 100 万 * 100 条 = 1 亿条
• 1 亿/24 小时/60 分/60 秒 = 1150 条/每秒钟
• 每条日志大小：0.5k - 2k（取 1k）
• 1150 条/每秒钟 * 1k ≈ 1m/s
• 高峰期每秒钟：1150 条 * 20 倍 = 23000 条
• 每秒多少数据量：20MB/s

1.2 服务器台数选择

服务器台数= 2 * （生产者峰值生产速率 * 副本 / 100） + 1= 2 * （20m/s * 2 / 100） + 1= 3 台

建议 3 台服务器

1.3 内存选择

Kafka 内存组成：堆内存 + 页缓存，Kafka 堆内存建议每个节点：10g ~ 15g，在 kafka-server-start.sh 中修改

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; thenexport KAFKA_HEAP_OPTS="-Xmx10G -Xms10G"
fi

查看kafka内存使用情况

jps
# 根据 Kafka 进程号，查看 Kafka 的 GC 情况
jstat -gc 2321 1s 10# S0C：第一个幸存区的大小； S1C：第二个幸存区的大小
# S0U：第一个幸存区的使用大小； S1U：第二个幸存区的使用大小
# EC：伊甸园区的大小； EU：伊甸园区的使用大小
# OC：老年代大小； OU：老年代使用大小
# MC：方法区大小； MU：方法区使用大小
# CCSC:压缩类空间大小； CCSU:压缩类空间使用大小
# YGC：年轻代垃圾回收次数； YGCT：年轻代垃圾回收消耗时间
# FGC：老年代垃圾回收次数； FGCT：老年代垃圾回收消耗时间
# GCT：垃圾回收消耗总时间；
# 根据 Kafka 进程号，查看 Kafka 的堆内存
jmap -heap 2321

页缓存：页缓存是 Linux 系统服务器的内存。我们只需要保证 1 个 segment（1g）中25%的数据在内存中就好。每个节点页缓存大小 =（分区数 * 1g * 25%）/ 节点数。例如 10 个分区，页缓存大小=（10 * 1g * 25%）/ 3 ≈ 1g，建议服务器内存大于等于 11G

1.4 其他硬件配置

磁盘选择

kafka 底层主要是顺序写，固态硬盘和机械硬盘的顺序写速度差不多。建议选择普通的机械硬盘。每天总数据量：1 亿条 * 1k ≈ 100g，100g * 副本 2 * 保存时间 3 天 / 0.7 ≈ 1T，建议三台服务器硬盘总大小，大于等于 1T

CPU 选择

• num.io.threads = 8 负责写磁盘的线程数，整个参数值要占总核数的 50%
• num.replica.fetchers = 1 副本拉取线程数，这个参数占总核数的 50%的 1/3
• num.network.threads = 3 数据传输线程数，这个参数占总核数的 50%的 2/3。建议 32 个 cpu core

网络选择

网络带宽 = 峰值吞吐量 ≈ 20MB/s 选择千兆网卡即可。100Mbps 单位是 bit；10M/s 单位是 byte ; 1byte = 8bit，100Mbps/8 = 12.5M/s。一般百兆的网卡（100Mbps ）、千兆的网卡（1000Mbps）、万兆的网卡（10000Mbps）。

2、生产者、Topic、消费者

具体的参数等详见第二章

3、Kafka 总体调优

3.1 提升吞吐量

• 生产者buffer.memory：发送消息的缓冲区大小，默认值是 32m，可以增加到 64m
• 生产batch.size：默认是 16k。如果 batch 设置太小，会导致频繁网络请求，吞吐量下降；如果 batch 太大，会导致一条消息需要等待很久才能被发送出去，增加网络延时
• 生产linger.ms，这个值默认是 0，意思就是消息必须立即被发送。一般设置一个 5-100毫秒。如果 linger.ms 设置的太小，会导致频繁网络请求，吞吐量下降；如果 linger.ms 太长，会导致一条消息需要等待很久才能被发送出去，增加网络延时
• 生产compression.type：默认是 none，不压缩，但是也可以使用 lz4 压缩，效率还是不错的，压缩之后可以减小数据量，提升吞吐量，但是会加大 producer 端的 CPU 开销
• 增加topic分区
• 消费者调整 fetch.max.bytes 大小，默认是 50m；调整 max.poll.records 大小，默认是 500 条
• 增加下游消费者处理能力

3.2 数据精准一次

• 生产者角度acks 设置为-1 （acks=-1）；幂等性（enable.idempotence = true） + 事务
• broker 服务端角度，分区副本大于等于 2 （–replication-factor 2），ISR 里应答的最小副本数量大于等于 2 （min.insync.replicas = 2）
• 消费者事务 + 手动提交 offset （enable.auto.commit = false）；消费者输出的目的地必须支持事务（MySQL、Kafka）

3.3 合理设置分区数

创建一个只有 1 个分区的 topic。测试这个 topic 的 producer 吞吐量和 consumer 吞吐量。假设他们的值分别是 Tp 和 Tc，单位可以是 MB/s。然后假设总的目标吞吐量是 Tt，那么分区数 = Tt / min（Tp，Tc）

例如：producer 吞吐量 = 20m/s；consumer 吞吐量 = 50m/s，期望吞吐量 100m/s；分区数 = 100 / 20 = 5 分区，分区数一般设置为：3-10 个，分区数不是越多越好，也不是越少越好，需要搭建完集群，进行压测，再灵活调整分区个数。

3.4 单条日志大于 1m

参数名称	描述
message.max.bytes	默认 1m，broker 端接收每个批次消息最大值
max.request.size	默认 1m，生产者发往 broker 每个请求消息最大值，针对 topic级别设置消息体的大小
replica.fetch.max.bytes	默认 1m，副本同步数据，每个批次消息最大值
fetch.max.bytes	默认Default: 52428800（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此这不是一个绝对最大值。一批次的大小受message.max.bytes （broker config）or max.message.bytes （topic config）影响。

3.5 服务器宕机

• 先尝试重新启动一下，如果能启动正常，那直接解决
• 如果重启不行，考虑增加内存、增加 CPU、网络带宽
• 如果将 kafka 整个节点误删除，如果副本数大于等于 2，可以按照服役新节点的方式重新服役一个新节点，并执行负载均衡

4、集群压力测试

• 生产者压测：kafka-producer-perf-test.sh
• 消费者压测：kafka-consumer-perf-test.sh

# =============生产者压测=============
# 创建一个 test topic，设置为 3 个分区 3 个副本
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --replication-factor 3 --partitions 3 --topic test
# 在/opt/module/kafka/bin 目录下面有这两个文件
bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=16384 linger.ms=0
# record-size 是一条信息有多大，单位是字节，本次测试设置为 1k
# num-records 是总共发送多少条信息，本次测试设置为 100 万条。
# throughput 是每秒多少条信息，设成-1，表示不限流，尽可能快的生产数据，可测出生产者最大吞吐量。本次实验设置为每秒钟 1 万条。
# producer-props 后面可以配置生产者相关参数，batch.size 配置为 16k# 结果发现是9MB/s左右# 调整 batch.size 大小,batch.size 默认值是 16k。本次实验 batch.size 设置为 32k
# batch.size 默认值是 16k。本次实验 batch.size 设置为 4k
bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=4096 linger.ms=0
# 调整 linger.ms 时间
# linger.ms 默认是 0ms。本次实验 linger.ms 设置为 50ms
bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=4096 linger.ms=50# 调整压缩方式
# 默认的压缩方式是 none。本次实验 compression.type 设置为 snappy
bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=4096 linger.ms=50 compression.type=snappy
# 默认的压缩方式是 none。本次实验 compression.type 设置为 zstd和gzip和lz4# 调整缓存大小
# 默认生产者端缓存大小 32m。本次实验 buffer.memory 设置为 64m
bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092 batch.size=4096 linger.ms=50 buffer.memory=67108864# =======================Consumer 压力测试==========
# 修改/opt/module/kafka/config/consumer.properties 文件中的一次拉取条数为 500
max.poll.records=500
# 消费 100 万条日志进行压测
bin/kafka-consumer-perf-test.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --topic test --messages 1000000 --consumer.config config/consumer.properties
# --bootstrap-server 指定 Kafka 集群地址
# --topic 指定 topic 的名称
# --messages 总共要消费的消息个数。本次实验 100 万条
# 修改/opt/module/kafka/config/consumer.properties 文件中的一次拉取条数为 2000# 调整 fetch.max.bytes 大小为 100m
# 修改/opt/module/kafka/config/consumer.properties 文件中的拉取一批数据大小 100m
fetch.max.bytes=104857600
bin/kafka-consumer-perf-test.sh --broker-list hadoop102:9092,hadoop103:9092,hadoop104:9092 --topic test --messages 1000000 --consumer.config config/consumer.properties