Kafka知识概况

Posted 2023-02-15 GeorgeLin98

Kafka知识概况

• Kafka简介
• Kafka 生产者
• Kafka Broker
• Kafka 消费者
• Kafka-Eagle 监控
• Kafka-Kraft 模式
• 集成 SpringBoot

Kafka简介

消息队列简介：

• 目 前企业中比较常见的消息队列产 品主 要有 Kafka、ActiveMQ 、RabbitMQ 、RocketMQ 等。
• 在大数据场景主要采用 Kafka 作为消息队列。在 JavaEE 开发中主要采用 ActiveMQ、 RabbitMQ、RocketMQ。
• 传统的消息队列的主要应用场景包括：缓存/消峰、解耦和异步通信。
  ①缓冲/消峰：有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。
  
  ②解耦：允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。
  
  ③异步通信：允许用户把一个消息放入队列，但并不立即处理它，然后在需要的时候再去处理它们。
  
• 消息队列的两种模式：
  ①点对点模式：消费者主动拉取数据，消息收到后清除消息。
  ②发布/订阅模式：
  <1>可以有多个topic主题（浏览、点赞、收藏、评论等）。
  <2>消费者消费数据之后，不删除数据。
  <3>每个消费者相互独立，都可以消费到数据。

Kafka 概述：

• Kafka传统定义：Kafka是一个分布式的基于发布/订阅模式的消息队列（Message Queue），主要应用于大数据实时处理领域。
• Kafka最新定义 ： Kafka是 一个开源的分布式事件流平台（Event Streaming Platform），被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用。
• Kafka 基础架构：
  ①为方便扩展，并提高吞吐量，一个topic分为多个partition
  ②配合分区的设计，提出消费者组的概念，组内每个消费者并行消费
  ③为提高可用性，为每个partition增加若干副本，类似NameNode HA
  ④ZK中记录谁是leader，Kafka2.8.0以后也可以配置不采用ZK
• 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。

集群部署：

• 解压安装包：tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/module/
• 修改解压后的文件名称：mv kafka_2.12-3.0.0/ kafka
• 进入到/opt/module/kafka 目录，修改配置文件：

cd 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

• 分别在 hadoop103 和 hadoop104上修改配置文件/opt/module/kafka/config/server.properties 中的broker.id=1、broker.id=2
  ①注：broker.id 不得重复，整个集群中唯一。

vim kafka/config/server.properties
修改:
# The id of the broker. This must be set to a unique integer for 
each broker.
broker.id=1

vim kafka/config/server.properties
修改:
# The id of the broker. This must be set to a unique integer for 
each broker.
broker.id=2

• 配置环境变量：

（1）在/etc/profile.d/my_env.sh 文件中增加 kafka 环境变量配置
sudo vim /etc/profile.d/my_env.sh
增加如下内容：
#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin
（2）刷新一下环境变量。
source /etc/profile

• 启动集群：

（1）先启动 Zookeeper 集群，然后启动 Kafka。
zk.sh start
（2）依次在 hadoop102、hadoop103、hadoop104 节点上启动 Kafka。
bin/kafka-server-start.sh -daemon config/server.properties

注意：配置文件的路径要能够到 server.properties。

• 关闭集群：bin/kafka-server-stop.sh

集群启停脚本：

• 在/home/george/bin 目录下创建文件 kf.sh 脚本文件： vim kf.sh

#! /bin/bash
case $1 in
"start")
	for i in hadoop102 hadoop103 hadoop104
 	do
 		echo " --------启动 $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 hadoop104
	do
		echo " --------停止 $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 进程了。

主题命令行操作：

• 查看操作主题命令参数： 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：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
  ①选项说明：
  <1>–topic 定义 topic 名
  <2>–replication-factor 定义副本数
  <3>–partitions 定义分区数
• 查看 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 主题的详情：bin/kafka-topics.sh --bootstrap-server
  hadoop102:9092 --describe --topic first
• 删除 topic：bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --delete --topic first

生产者命令行操作：

• 查看操作生产者命令参数：bin/kafka-console-producer.sh

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

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

消费者命令行操作：

• 查看操作消费者命令参数：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 主题中的数据：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 生产者

生产者消息发送流程：

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

普通异步发送：

• 导入依赖：

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

• 编写不带回调函数的 API 代码：

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","george" + i));
	 
	 
	 // 5. 关闭资源
	 kafkaProducer.close();
 

• 测试：在服务器上开启消费者：bin/kafka-console-consumer.sh –
  bootstrap-server hadoop102:9092 --topic first
  george 0
  george 1
  george 2
  george 3
  george 4

带回调函数的异步发送：

• 回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是元数据信息（RecordMetadata）和异常信息（Exception），如果 Exception 为 null，说明消息发 送成功，如果Exception 不为null，说明消息发送失败。
• 注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。
• 代码：

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", "atguigu " + 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();

• 测试：
  ①在服务器上开启消费者：bin/kafka-console-consumer.sh –
  bootstrap-server hadoop102:9092 --topic first
  george 0
  george 1
  george 2
  george 3
  george 4
  ②在 IDEA 控制台观察回调信息：
  主题：first->分区：0
  主题：first->分区：0
  主题：first->分区：1
  主题：first->分区：1
  主题：first->分区：1

同步发送 API：

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

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

• 测试：
  ①在服务器上开启 Kafka 消费者：bin/kafka-console-consumer.sh –
  bootstrap-server hadoop102:9092 --topic first
  ②观察控制台中是否接收到消息：bin/kafka-console-consumer.sh –
  bootstrap-server hadoop102:9092 --topic first
  george 0
  george 1
  george 2
  george 3
  george 4

生产者分区：

• Kafka 分区好处：
  ①便于合理使用存储资源，每个Partition在一个Broker上存储，可以把海量的数据按照分区切割成一块一块数据存储在多台Broker上。合理控制分区的任务，可以实现负载均衡的效果。
  ②提高并行度，生产者可以以分区为单位发送数据；消费者可以以分区为单位进行消费数据。
• 生产者发送消息的分区策略：
  ①默认的分区器 DefaultPartitioner：在IDEA中全局查找（ctrl +n）ProducerRecord类，在类中可以看到如下构造方法：

（1）指明partition的情况下，直接将指明的值作为partition值；
例如partition=0，所有数据写入分区0
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers) 
... ...

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

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

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


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


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

• 自定义分区器：
  ①实现步骤

<1>定义类实现 Partitioner 接口。

<2>重写 partition()方法。
/**
 * 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;
		 
		 // 判断消息是否包含 atguigu
 		if (msgValue.contains("atguigu"))
			partition = 0;
 		else 
 			partition = 1;
		
		
 		// 返回分区号
 		return partition;
 	
 	
 	// 关闭资源
	@Override
 	public void close() 
 	
 	
 	// 配置方法
 	@Override
 	public void configure(Map<String, ?> configs) 
 	


<3>使用分区器的方法，在生产者的配置中添加分区器参数。
// 添加自定义分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.george.kafka.producer.MyPartitioner");

生产经验——生产者如何提高吞吐量：

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

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","atguigu " + i));
		
		
			// 5. 关闭资源
		kafkaProducer.close();
 	

生产经验——数据可靠性：

• ack 应答原理：
  
• ACK应答级别：
  
• 数据完全可靠条件 = ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2
• 可靠性总结：
  ①acks=0，生产者发送过来数据就不管了，可靠性差，效率高；
  ②acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等；
  ③acks=-1，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低；
  ④在生产环境中，acks=0很少使用；acks=1，一般用于传输普通日志，允许丢个别数据；acks=-1，一般用于传输和钱相关的数据，对可靠性要求比较高的场景。
• 数据重复分析：
  

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","atguigu " + i));
		
		// 5. 关闭资源
		kafkaProducer.close();
	

生产经验——数据去重：

• 数据传递语义：
  ①至少一次（At Least Once）= ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2。
  ②最多一次（At Most Once）= ACK级别设置为0。
  ③总结：
  <1>At Least Once可以保证数据不丢失，但是不能保证数据不重复；
  <2>At Most Once可以保证数据不重复，但是不能保证数据不丢失。
  ④精确一次（Exactly Once）：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。
  ⑤Kafka 0.11版本以后，引入了一项重大特性：幂等性和事务。
• 幂等性：
  ①幂等性原理：
  <1>幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证了不重复。
  <2>精确一次（Exactly Once） = 幂等性 + 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数量>=2） 。
  <3>重复数据的判断标准：具有<PID, Partition, SeqNumber>相同主键的消息提交时，Broker只会持久化一条。其中PID是Kafka每次重启都会分配一个新的；Partition 表示分区号；Sequence Number是单调自增的。
  <4>所以幂等性只能保证的是在单分区单会话内不重复。
  ②如何使用幂等性：开启参数 enable.idempotence 默认为 true，false 关闭。
• 生产者事务：
  ①Kafka 事务原理：说明：开启事务，必须开启幂等性。
  

<1>Kafka 的事务一共有如下 5 个 API：
// 1 初始化事务
void initTransactions();
// 2 开启事务
void beginTransaction() throws ProducerFencedException;
// 3 在事务内提交已经消费的偏移量（主要用于消费者）
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
 String consumerGroupId) throws 
ProducerFencedException;
// 4 提交事务
void commitTransaction() throws ProducerFencedException;
// 5 放弃事务（类似于回滚事务的操作）
void abortTransaction() throws ProducerFencedException;

<2>单个 Producer，使用事务保证消息的仅一次发送：
public class CustomProducerTransactions 
	public static void main(String[] args) throws  InterruptedException 
		// 1. 创建 kafka 生产者的配置对象
		Properties properties = new Properties();
		
		// 2. 给 kafka 配置对象添加配置信息
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
		
		// key,value 序列化
		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
		// 设置事务 id（必须），事务 id 任意起名
		properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction_id_0");
		
		// 3. 创建 kafka 生产者对象
		KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
		
		// 初始化事务
		kafkaProducer.initTransactions();
		// 开启事务
		kafkaProducer.beginTransaction();
		try 
	 		// 4. 调用 send 方法,发送消息
			for (int i = 0; i < 5; i++) 
				// 发送消息
				kafkaProducer.send(new ProducerRecord<>("first", "atguigu " + i));
		 	
			// int i = 1 / 0;
			// 提交事务
			kafkaProducer.commitTransaction();
		 catch (Exception e) 
			 // 终止事务
			 kafkaProducer.abortTransaction();
	     finally 
			 // 5. 关闭资源
			 kafkaProducer.close();
		
 	

生产经验——数据有序：

生产经验——数据乱序：

• kafka在1.x版本之前保证数据单分区有序，条件如下：max.in.flight.requests.per.connection=1（不需要考虑是否开启幂等性）。
• kafka在1.x及以后版本保证数据单分区有序，条件如下：
  ①未开启幂等性：max.in.flight.requests.per.connection需要设置为1。
  ②开启幂等性：max.in.flight.requests.per.connection需要设置小于等于5。
  <1>原因说明：因为在kafka1.x以后，启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，故无论如何，都可以保证最近5个request的数据都是有序的。
  

Kafka Broker

Kafka Broker 工作流程：

• Zookeeper 存储的 Kafka 信息：
  ①启动 Zookeeper 客户端： bin/zkCli.sh
  ②通过 ls 命令可以查看 kafka 相关信息：ls /kafka
  
• Kafka Broker 总体工作流程：
  
• 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。一般不建议修改，交给系统自己管理。

生产经验——节点服役和退役：

• 服役新节点：
  ①新节点准备
  ②执行负载均衡操作

（1）创建一个要均衡的主题。
 vim topics-to-move.json

	 "topics": [
		 "topic": "first"
	 ],
	 "version": 1


（2）生成一个负载均衡的计划。
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

（3）创建副本存储计划（所有副本存储在 broker0、broker1、broker2、broker3 中）
vim increase-replication-factor.json
输入如下内容：
"version":1,"partitions":["topic":"first","partition":0,"replic
as":[2,3,0],"log_dirs":["any","any","any"],"topic":"first","par
tition":1,"replicas":[3,0,1],"log_dirs":["any","any","any"],"to
pic":"first","partition":2,"replicas":[0,1,2],"log_dirs":["any","
any","any"]]

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

（5）验证副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify

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

（1）创建一个要均衡的主题。
vim topics-to-move.json

 	"topics": [
 		"topic": "first"
 	],
 	"version": 1


（2）创建执行计划。
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2" --generate

（3）创建副本存储计划（所有副本存储在 broker0、broker1、broker2 中）。
vim increase-replication-factor.json
"version":1,"partitions":["topic":"first","partition":0,"replic
as":[2,0,1],"log_dirs":["any","any","any"],"topic":"first","par
tition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"],"to
pic":"first","partition":2,"replicas":[1,2,0],"log_dirs":["any","
any","any"]]

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

（5）验证副本存储计划。
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file  increase-replication-factor.json --verify

Kafka 副本：

• 副本基本信息：
  ①Kafka 副本作用：提高数据可靠性。
  ②Kafka 默认副本 1 个，生产环境一般配置为 2 个，保证数据可靠性；太多副本会增加磁盘存储空间，增加网络上数据传输，降低效率。
  ③Kafka 中副本分为：Leader 和 Follower。Kafka 生产者只会把数据发往 Leader，然后 Follower 找 Leader 进行同步数据。
  ④Kafka 分区中的所有副本统称为 AR（Assigned Repllicas）。
  ⑤AR = ISR + OSR
  <1>ISR，表示和 Leader 保持同步的 Follower 集合。如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值由 replica.lag.time.max.ms参数设定，默认 30s。Leader 发生故障之后，就会从 ISR 中选举新的 Leader。
  <2>OSR，表示 Follower 与 Leader 副本同步时，延迟过多的副本。
• Leader 选举流程：
  ①Kafka 集群中有一个 broker 的 Controller 会被选举为 Controller Leader，负责管理集群broker 的上下线，所有 topic 的分区副本分配和 Leader 选举等工作。
  ②Controller 的信息同步工作是依赖于 Zookeeper 的。
  
• Leader 和 Follower 故障处理细节：
  ①Follower故障处理细节：每个副本的最后一个offset，LEO其实就是最新的offset + 1。
  ②HW（High Watermark）：所有副本中最小的LEO。
  
  
• 分区副本分配：如果 kafka 服务器只有 4 个节点，那么设置 kafka 的分区数大于服务器台数，在 kafka底层如何分配存储副本呢？
  ①创建 16 分区，3 个副本
• 生产经验——手动调整分区副本存储：
  ①在生产环境中，每台服务器的配置和性能不一致，但是Kafka只会根据自己的代码规则创建对应的分区副本，就会导致个别服务器存储压力较大。所有需要手动调整分区副本的存储。
• 生产经验——Leader Partition 负载平衡：
  ①正常情况下，Kafka本身会自动把Leader Partition均匀分散在各个机器上，来保证每台机器的读写吞吐量都是均匀的。但是如果某些broker宕机，会导致Leader Partition过于集中在其他少部分几台broker上，这会导致少数几台broker的读写请求压力过高，其他宕机的broker重启之后都是follower partition，读写请求很低，造成集群负载不均衡。
  

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

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

1）创建 topic
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 3 --replication-factor 1 --topic four

2）手动增加副本存储
（1）创建副本存储计划（所有副本都指定存储在 broker0、broker1、broker2 中）。
 vim increase-replication-factor.json
 输入如下内容：
"version":1,"partitions":["topic":"four","partition":0,"replica
s":[0,1,2],"topic":"four","partition":1,"replicas":[0,1,2],"t
opic":"four","partition":2,"replicas":[0,1,2]]

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

文件存储：

• 文件存储机制：
  ①Topic 数据的存储机制：Topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是Producer生产的数据。Producer生产的数据会被不断追加到该log文件末端，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制，将每个partition分为多个segment。每个segment包括：“.index”文件、“.log”文件和.timeindex等文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号，例如：first-0。
  
  ②思考：Topic 数据到底存储在什么位置----> /opt/module/kafka/datas/first-1（first-0、first-2）路径上的文件。
  ③index 文件和 log 文件详解：
  
  <1>说明：日志存储参数配置

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

• 文件清理策略：
  ①Kafka 中默认的日志保存时间为 7 天，可以通过调整如下参数修改保存时间。
  <1> log.retention.hours，最低优先级小时，默认 7 天。
  <2> log.retention.minutes，分钟。
  <3> log.retention.ms，最高优先级毫秒。
  <4> log.retention.check.interval.ms，负责设置检查周期，默认 5 分钟。
  ②那么日志一旦超过了设置的时间，怎么处理呢？
  <1>Kafka 中提供的日志清理策略有 delete 和 compact 两种。
  1)delete 日志删除：将过期数据删除：
  （1）log.cleanup.policy = delete 所有数据启用删除策略
  （2）log.retention.bytes，默认等于-1，表示无穷大。
  （3）基于时间：默认打开。以 segment 中所有记录中的最大时间戳作为该文件时间戳。
  （4）基于大小：默认关闭。超过设置的所有日志总大小，删除最早的 segment。
  思考：如果一个 segment 中有一部分数据过期，一部分没有过期，怎么处理？
  2）compact 日志压缩：compact日志压缩：对于相同key的不同value值，只保留最后一个版本。
  （1）log.cleanup.policy = compact 所有数据启用压缩策略
  
  （2）压缩后的offset可能是不连续的，比如上图中没有6，当从这些offset消费消息时，将会拿到比这个offset大的offset对应的消息，实际上会拿到offset为7的消息，并从这个位置开始消费。
  （3）这种策略只适合特殊场景，比如消息的key是用户ID，value是用户的资料，通过这种压缩策略，整个消息集里就保存了所有用户最新的资料。

高效读写数据：

• Kafka 本身是分布式集群，可以采用分区技术，并行度高
• 读数据采用稀疏索引，可以快速定位要消费的数据
• 顺序写磁盘# 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。一般不建议修改，交给系统自己管理。

Kafka 消费者

Kafka 消费方式：

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

Kafka 消费者工作流程：

• Kafka 消费者总体工作流程：
  

• 消费者组原理：
  ①消费者组Consumer Group（CG）：消费者组，由多个consumer组成。形成一个消费者组的条件，是所有消费者的groupid相同。
  <1>消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费。
  <2>消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
  
  
  ②消费者组初始化流程：
  
  ③消费者组详细消费流程：
  

• 消费者重要参数：

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


测试:
（1）在 IDEA 中执行消费者程序
（2）在 Kafka 集群控制台，创建 Kafka 生产者，并输入数据
（3）在 IDEA 控制台观察接收到的数据

• 测试独立消费者（订阅分区）：

需求：创建一个独立消费者，消费 first 主题 0 号分区的数据。
public class CustomConsumerPartition 
	public static void main(String[] args) 
		Properties properties = new Properties();
		properties.put(C
以上是关于Kafka知识概况的主要内容，如果未能解决你的问题，请参考以下文章
 
 1.5万字长文：从 C# 入门 Kafka（Kafka基础知识）
 kafka学习笔记：知识点整理
 kafka学习笔记：知识点整理
 百科知识 固定电话格式是什么
 知识概况
 Extjs 知识点概况