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(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
		TopicPartition("first", 0));
		kafkaConsumer.assign(topicPartitions);
		while (true){
			ConsumerRecords<String, String> consumerRecords = 
			kafkaConsumer.poll(Duration.ofSeconds(1));
			for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
				System.out.println(consumerRecord);
			}
		}
	 }
}
测试:
（1）在 IDEA 中执行消费者程序。
（2）在 IDEA 中执行生产者程序在控制台观察生成几个 0 号分区的数据。
（3）在 IDEA 控制台，观察接收到的数据，只能消费到 0 号分区数据表示正确。

• 测试消费者组：

需求：测试同一个主题的分区数据，只能由一个消费者组中的一个消费。
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);
			}
		}
	}
}

测试：
（1）复制一份基础消费者的代码，在 IDEA 中同时启动，即可启动同一个消费者组中的两个消费者。
（2）启动代码中的生产者发送消息，在 IDEA 控制台即可看到两个消费者在消费不同分区的数据（如果只发生到一个分区，可以在发送时增加延迟代码 Thread.sleep(2);
（3）重新发送到一个全新的主题中，由于默认创建的主题分区数为 1，可以看到只能有一个消费者消费到数据。

生产经验——分区的分配以及再平衡：

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

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

• Range 以及再平衡：
  ①Range 分区策略原理：Range 是对每个 topic 而言的。首先对同一个 topic 里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。假如现在有 7 个分区，3 个消费者，排序后的分区将会是0,1,2,3,4,5,6；消费者排序完之后将会是C0,C1,C2。例如，7/3 = 2 余 1 ，除不尽，那么 消费者 C0 便会多消费 1 个分区。 8/3=2余2，除不尽，那么C0和C1分别多消费一个。通过 partitions数/consumer数 来决定每个消费者应该消费几个分区。如果除不尽，那么前面几个消费者将会多消费 1 个分区。分区分配策略之Range注意：如果只是针对 1 个 topic 而言，C0消费者多消费1个分区影响不是很大。但是如果有 N 多个 topic，那么针对每个 topic，消费者 C0都将多消费 1 个分区，topic越多，C0消费的分区会比其他消费者明显多消费 N 个分区。容易产生数据倾斜！
  ②说明：Kafka 默认的分区分配策略就是 Range + CooperativeSticky，所以不需要修改策略。
  

• RoundRobin 以及再平衡：
  ①RoundRobin 针对集群中所有Topic而言。
  ②RoundRobin 轮询分区策略，是把所有的 partition 和所有的consumer 都列出来，然后按照 hashcode 进行排序，最后通过轮询算法来分配 partition 给到各个消费者。
  

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

offset 位移：

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

• 自动提交 offset：为了使我们能够专注于自己的业务逻辑，Kafka提供了自动提交offset的功能。自动提交offset的相关参数：
  ① enable.auto.commit：是否开启自动提交offset功能，默认是true
  ② auto.commit.interval.ms：自动提交offset的时间间隔，默认是5s
  

参数名称	描述
enable.auto.commit	默认值为 true，消费者会自动周期性地向服务器提交偏移量。
auto.commit.interval.ms	如果设置了 enable.auto.commit 的值为 true， 则该值定义了消费者偏移量向 Kafka 提交的频率，默认 5s。

消费者自动提交 offset:

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提交的时机。因此Kafka还提供了手动提交offset的API。
  ②手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。两者的相同点是，都会将本次提交的一批数据最高的偏移量提交；不同点是，同步提交阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；而异步提交则没有失败重试机制，故有可能提交失败。
  <1>commitSync（同步提交）：必须等待offset提交完毕，再去消费下一批数据。
  <2>commitAsync（异步提交） ：发送完提交offset请求后，就开始消费下一批数据了。
  

1）同步提交 offset
由于同步提交 offset 有失败重试机制，故更加可靠，但是由于一直等待提交结果，提交的效率比较低。以下为同步提交 offset 的示例。

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

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

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 位移开始消费。注意：每次执行完，需要修改消费者组名。

public class CustomConsumerSeek {
	public static void main(String[] args) {
		// 0 配置信息
		Properties properties = new Properties();
		// 连接
		properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
		// key value 反序列化		
		properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());		
		properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
		StringDeserializer.class.getName());
		properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test2");
		
		// 1 创建一个消费者
		KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
		
		// 2 订阅一个主题
		ArrayList<String> topics = new ArrayList<>();
		topics.add("first");
		kafkaConsumer.subscribe(topics);
		Set<TopicPartition> assignment= new HashSet<>();
		while (assignment.size() == 0) {
			kafkaConsumer.poll(Duration.ofSeconds(1));
			// 获取消费者分区分配信息（有了分区分配信息才能开始消费）
			assignment = kafkaConsumer.assignment();
		}
		// 遍历所有分区，并指定 offset 从 1700 的位置开始消费
		for (TopicPartition tp: assignment) {
		kafkaConsumer.seek(tp, 1700);
		}
		// 3 消费该主题数据
		while (true) {
			ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
			for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
				System.out.println(consumerRecord);
			}
		}
	}
}

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

public class CustomConsumerForTime {
	public static void main(String[] args) {
		// 0 配置信息
		Properties properties = new Properties();
		// 连接
		properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
		// key value 反序列化
		properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
		properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
		properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test2");
		// 1 创建一个消费者
		KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
		
		// 2 订阅一个主题
		ArrayList<String> topics = new ArrayList<>();
		topics.add("first");
		kafkaConsumer.subscribe(topics);
		Set<TopicPartition> assignment = new HashSet<>();
		while (assignment.size() == 0) {
			kafkaConsumer.poll(Duration.ofSeconds(1));
			// 获取消费者分区分配信息（有了分区分配信息才能开始消费）
			assignment = kafkaConsumer.assignment();
		}
		HashMap<TopicPartition, Long> timestampToSearch = new HashMap<>();
		// 封装集合存储，每个分区对应一天前的数据
		for (TopicPartition topicPartition : assignment) {
			timestampToSearch.put(topicPartition, 
			System.currentTimeMillis() - 1 * 24 * 3600 * 1000);
		}
		// 获取从 1 天前开始消费的每个分区的 offset
		Map<TopicPartition, OffsetAndTimestamp> offsets = 
		kafkaConsumer.offsetsForTimes(timestampToSearch);
		// 遍历每个分区，对每个分区设置消费时间。
		for (TopicPartition topicPartition : assignment) {
			OffsetAndTimestamp offsetAndTimestamp = 
			offsets.get(topicPartition);
			// 根据时间指定开始消费的位置
			if (offsetAndTimestamp != null){
				kafkaConsumer.seek(topicPartition, 
				offsetAndTimestamp.offset());
			}
 		}
 		
		// 3 消费该主题数据
		while (true) {
			ConsumerRecords<String, String> consumerRecords = 
			kafkaConsumer.poll(Duration.ofSeconds(1));
			for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
				System.out.println(consumerRecord);
			}
		}
	}
}

• 漏消费和重复消费：
  ①重复消费：已经消费了数据，但是 offset 没提交。
  ②漏消费：先提交 offset 后消费，有可能会造成数据的漏消费。
  ③重复消费与漏消费：
  <1>场景1：重复消费。自动提交offset引起。
  
  <2>场景1：漏消费。设置offset为手动提交，当offset被提交时，数据还在内存中未落盘，此时刚好消费者线程被kill掉，那么offset已经提交，但是数据未处理，导致这部分内存中的数据丢失。
  

生产经验——消费者事务：

• 如果想完成Consumer端的精准一次性消费，那么需要Kafka消费端将消费过程和提交offset过程做原子绑定。此时我们需要将Kafka的offset保存到支持事务的自定义介质（比 如 MySQL）。这部分知识会在后续项目部分涉及。
  

生产经验——数据积压（消费者如何提高吞吐量）：

• 如果是Kafka消费能力不足，则可以考虑增 加Topic的分区数，并且同时提升消费组的消费者 数量，消费者数 = 分区数。（两者缺一不可）
  

• 如果是下游的数据处理不及时：提高每批次拉取的数量。批次拉取数据过少（拉取数据/处理时间 < 生产速度），使处理的数据小于生产的数据，也会造成数据积压。
  

参数名称	描述
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-Eagle 框架可以监控 Kafka 集群的整体运行情况，在生产环境中经常使用。
• Kafka-Eagle 的安装依赖于 MySQL，MySQL 主要用来存储可视化展示的数据。

Kafka 环境准备：

• 关闭Kafka集群
• 修改/opt/module/kafka/bin/kafka-server-start.sh命令中

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-Eagle 安装：

• 上传压缩包 kafka-eagle-bin-2.0.8.tar.gz 到集群/opt/software 目录
• 解压到本地：

tar -zxvf kafka-eagle-bin-2.0.8.tar.gz

• 进入刚才解压的目录

ll

• 将 efak-web-2.0.8-bin.tar.gz 解压至/opt/module

tar -zxvf efak-web-2.0.8-bin.tar.gz -C /opt/module/

• 修改名称：

mv efak-web-2.0.8/ efak

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

[atguigu@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.truststor
e
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.ScramL
oginModule 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.PlainL
oginModule 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=UT
F-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=U
TF-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
source /etc/profile

• 启动：

（1）注意：启动之前需要先启动 ZK 以及 KAFKA。
 kf.sh start
（2）启动 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，执行命令。
bin/ke.sh stop

Kafka-Eagle 页面操作：

• 登录页面查看监控数据：http://192.168.10.102:8048/
  
  

Kafka-Kraft 模式

Kafka-Kraft 架构：

• 左图为 Kafka 现有架构，元数据在 zookeeper 中，运行时动态选举 controller，由 controller 进行Kafka 集群管理。右图为 kraft 模式架构（实验性），不再依赖 zookeeper 集群， 而是用三台 controller节点代替 zookeeper，元数据保存在 controller中，由controller 直接进行 Kafka集群管理。
  
• 这样做的好处有以下几个：
  ①Kafka 不再依赖外部框架，而是能够独立运行。
  ②controller 管理集群时，不再需要从 zookeeper 中先读取数据，集群性能上升。
  ③由于不依赖 zookeeper，集群扩展时不再受到 zookeeper 读写能力限制；
  ④controller 不再动态选举，而是由配置文件规定。这样我们可以有针对性的加强。
  ⑤controller 节点的配置，而不是像以前一样对随机 controller 节点的高负载束手无策。

Kafka-Kraft 集群部署：

1）再次解压一份 kafka 安装包
tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/module/
2）重命名为 kafka2
mv kafka_2.12-3.0.0/ kafka2
3）在 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@hado
op104: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:PLA
INTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
#kafka 数据存储目录
log.dirs=/opt/module/kafka2/data

5）初始化集群数据目录
（1）首先生成存储目录唯一 ID。
bin/kafka-storage.sh random-uuid J7s9e8PPTKOO47PxzI39VA
（2）用该 ID 格式化 kafka 存储目录（三台节点）。
bin/kafka-storage.sh format -t  J7s9e8PPTKOO47PxzI39VA -c 
/opt/module/kafka2/config/kraft/server.properties

6）启动 kafka 集群
bin/kafka-server-start.sh -daemon config/kraft/server.properties
7）停止 kafka 集群
bin/kafka-server-stop.sh

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

启动：
/opt/module/kafka2/bin/kafka-server-start.sh -daemon /opt/module/kafka2/config/kraft/server.properties

停止：
/opt/module/kafka2/bin/kafka-server-stop.sh

集成 SpringBoot

环境准备：

• 在New Project步骤处选择Messaging然后再选择Spring for Apache Kafka。
• 生成项目后会发现pom.xml中

<dependency>
	 <groupId>org.springframework.boot</groupId>
	 <artifactId>spring-boot-starter-web</artifactId>
</dependency>

SpringBoot 生产者：

• 修改 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.keyserializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.valueserializer=org.apache.kafka.common.serialization.StringSerializer

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

@RestController
public class ProducerController {
 
	// Kafka 模板用来向 kafka 发送数据
	@Autowired
	KafkaTemplate<String, String> kafka;
	
	@RequestMapping("/atguigu")
	public String data(String msg) {
		kafka.send("first", msg);
		return "ok";
	}
}

SpringBoot 消费者：

• 修改 SpringBoot 核心配置文件 application.propeties：

# =========消费者配置开始=========
# 指定 kafka 的地址
spring.kafka.bootstrapservers=hadoop102:9092,hadoop103:9092,hadoop104:9092

# 指定 key 和 value 的反序列化器
spring.kafka.consumer.keydeserializer=org.apache.kafka.common.serialization.StringDeserial
izer
spring.kafka.consumer.valuedeserializer=org.apache.kafka.common.serialization.StringDeserial
izer

#指定消费者组的 group_id
spring.kafka.consumer.group-id=atguigu
# =========消费者配置结束=========

• 创建类消费 Kafka 中指定 topic 的数据：

@Configuration
public class KafkaConsumer {
 
	// 指定要监听的 topic
	@KafkaListener(topics = "first")
	public void consumeTopic(String msg) { // 参数: 收到的 value
		System.out.println("收到的信息: " + msg);
	}
}