Kafka |
您所在的位置：网站首页 › kafka消费者订阅多个topic › Kafka |
Kafka
|
资料来源视频：尚硅谷-Kafka3.x教程 一、Kafka概述 1.1 定义Kafka传统定义:Kafka是一个分布式的基于发布/订阅模式的消息队列(MessageQueue)，主要应用于大数据实时处理领域。 发布订阅:消息的发布者不会将消息直接发送给特定的订阅者，而是将发布的消息分为不同的类别，订阅者只接收感兴趣的消息 Kafka最新定义:Kafka是一个开源的分布式事件流平台(Event StreamingPlatform)，被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用
目前企业中比较常见的消息队列产品主要有Kafka、ActiveMQ 、RabbitMQ 、RocketMQ等。 在大数据场景主要采用 Kafka 作为消息队列。在JavaEE开发中主要采用 ActiveMQ、RabbitMQ、RocketMQ。 1.2.1 传统消息队列的应用场景传统的消息队列的主要应用场景包括：缓存/削峰、解耦和异步通信、 消息队列的应用场景——缓冲/消峰 缓冲/消峰：有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。
消息队列的应用场景——解耦 解耦：允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。
消息队列的应用场景——异步通信 异步通信：允许用户把一个消息放入队列，但并不立即处理它，然后在需要的时候再去处理它们。
1）点对点模式 消费者主动拉取数据，消息收到后清除消息
2）发布/订阅模式 可以有多个topic主题（浏览、点赞、收藏、评论等）消费者消费数据之后，不删除数据每个消费者相互独立，都可以消费到数据
0）官方下载地址：Apache Kafka 1）解压安装包 [atguigu@hadoop102 software]$ tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/module/2）修改解压后的文件名称 [atguigu@hadoop102 module]$ mv kafka_2.12-3.0.0/ kafka3）进入到/opt/module/kafka 目录，修改配置文件 [atguigu@hadoop102 kafka]$ cd config/ [atguigu@hadoop102 config]$ vim server.properties输入以下内容： #broker的全局唯一编号，不能重复，只能是数字。 broker.id=0 #处理网络请求的线程数量 num.network.threads=3 #用来处理磁盘IO的线程数量 num.io.threads=8 #发送套接字的缓冲区大小 socket.send.buffer.bytes=102400 #接收套接字的缓冲区大小 socket.receive.buffer.bytes=102400 #请求套接字的缓冲区大小 socket.request.max.bytes=104857600 #kafka运行日志(数据)存放的路径，路径不需要提前创建，kafka 自动帮你创建，可以配置多个磁盘路径，路径与路径之间可以用"，"分隔 log.dirs=/opt/module/kafka/datas #topic 在当前 broker 上的分区个数 num.partitions=1 #用来恢复和清理 data 下数据的线程数量 num.recovery.threads.per.data.dir=1 # 每个topic创建时的副本数，默认时 1 个副本 offsets.topic.replication.factor=1 #segment文件保留的最长时间，超时将被删除 log.retention.hours=168 #每个segment文件的大小，默认最大 1G log.segment.bytes=1073741824 # 查过期数据的时间，默认5分钟检查一次是否数据过期 log.retention.check.interval.ms=300000 #配置连接Zookeeper集群地址（在 zk 根目录下创建/kafka，方便管理） zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka4）分发安装包 [atguigu@hadoop102 module]$ xsync kafka/5）分别在 hadoop103 和 hadoop104 上修改配置文件/opt/module/kafka/config/server.properties中的 broker.id=1、broker.id=2 注：broker.id 不得重复，整个集群中唯一。 [atguigu@hadoop103 module]$ vim kafka/config/server.properties # The id of the broker. This must be set to a unique integer for each broker. broker.id=1 [atguigu@hadoop104 module]$ vim kafka/config/server.properties # The id of the broker. This must be set to a unique integer for each broker. broker.id=26）配置环境变量 （1）在/etc/profile.d/my_env.sh 文件中增加 kafka 环境变量配置 [atguigu@hadoop102 module]$ sudo vim /etc/profile.d/my_env.sh #KAFKA_HOME export KAFKA_HOME=/opt/module/kafka export PATH=$PATH:$KAFKA_HOME/bin（2）刷新一下环境变量。 [atguigu@hadoop102 module]$ source /etc/profile（3）分发环境变量文件到其他节点，并 source。 [atguigu@hadoop102 module]$ sudo /home/atguigu/bin/xsync /etc/profile.d/my_env.sh [atguigu@hadoop103 module]$ source /etc/profile [atguigu@hadoop104 module]$ source /etc/profile7）启动集群 （1）先启动 Zookeeper 集群，然后启动 Kafka。 [atguigu@hadoop102 kafka]$ zk.sh start（2）依次在 hadoop102、hadoop103、hadoop104 节点上启动 Kafka。 [atguigu@hadoop102 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties [atguigu@hadoop103 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties [atguigu@hadoop104 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties8）关闭集群 [atguigu@hadoop102 kafka]$ bin/kafka-server-stop.sh [atguigu@hadoop103 kafka]$ bin/kafka-server-stop.sh [atguigu@hadoop104 kafka]$ bin/kafka-server-stop.sh 2.1.3 集群启停脚本1）在/home/atguigu/bin 目录下创建文件 kf.sh 脚本文件 [atguigu@hadoop102 bin]$ vim kf.sh #! /bin/bash case $1 in "start"){ for i in hadoop102 hadoop103 hadoop104 do echo " --------启动 $i Kafka-------" ssh $i "/opt/module/kafka/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 };; esac2）添加执行权限 [atguigu@hadoop102 bin]$ chmod +x kf.sh3）启动集群命令 [atguigu@hadoop102 ~]$ kf.sh start4）停止集群命令 [atguigu@hadoop102 ~]$ kf.sh stop注意：停止 Kafka 集群时，一定要等 Kafka 所有节点进程全部停止后再停止 Zookeeper集群。 1因为 Zookeeper 集群当中记录着 Kafka 集群相关信息，Zookeeper 集群一旦先停止，Kafka 集群就没有办法再获取停止进程的信息，只能手动杀死 Kafka 进程了。 2.1.4 MacOs安装使用可以通过homebrew安装 brew install kafka安装路径 # 安装路径 /opt/homebrew/Cellar/kafka/3.4.0/bin # 配置项路径 /opt/homebrew/etc/kafka # 文件路径 /opt/homebrew/var/lib/kafka-logs启动命令 cd /opt/homebrew/Cellar/kafka/3.4.0/bin # 启动zookeeper zookeeper-server-start -daemon /opt/homebrew/etc/kafka/zookeeper.properties # 启动kafka kafka-server-start -daemon /opt/homebrew/etc/kafka/server.properties停止命令 cd /opt/homebrew/Cellar/kafka/3.4.0/bin # 停止kafka kafka-server-stop # 停止kafka kafka-server-stop 2.2 命令行操作
1）查看操作主题命令参数 [root@localhost kafka]# bin/kafka-topics.sh
2）查看当前服务器中的所有 topic [root@localhost kafka]# bin/kafka-topics.sh --bootstrap-server localhost:9092 --list3）创建 first topic [root@localhost kafka]# bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --partitions 1 --replication-factor 1 --topic first Created topic first.选项说明： --topic 定义 topic 名 --replication-factor 定义副本数 --partitions 定义分区数 4）查看 first 主题的详情 [root@localhost kafka]# bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic first Topic: first TopicId: TCr1n2_vQTu7WXQVFHD-PA PartitionCount: 1 ReplicationFactor: 1 Configs: Topic: first Partition: 0 Leader: 0 Replicas: 0 Isr: 05）修改分区数（注意：分区数只能增加，不能减少） [root@localhost kafka]$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic first --partitions 36）再次查看 first 主题的详情 [root@localhost kafka]# bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic first7）删除 topic [root@localhost kafka]# bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic first 2.2.2 生产者命令行操作1）查看操作生产者命令参数 [root@localhost kafka]# bin/kafka-console-producer.sh
2）发送消息 [root@localhost kafka]# bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic first >hello 2.2.3 消费者命令行操作1）查看操作消费者命令参数 [root@localhost kafka]# bin/kafka-console-consumer.sh
2）消费消息 （1）消费 first 主题中的数据 [root@localhost kafka]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first（2）把主题中所有的数据都读取出来（包括历史数据）。 [root@localhost kafka]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first --from-beginning查看topic ###ip位本机地址 bin/kafka-topics.sh --bootstrap-server localhost:9092 --list [root@localhost kafka]# bin/kafka-topics.sh --bootstrap-server localhost:9092 --list __consumer_offsets bc_data_return equipment_bc_data_return identifier_data_expert_topic identifier_data_task_status_report identifier_data_task_status_topic nginx-access-log order-info prnm-monitor-node-info-topic prnm-passive-rec-node-info-topic创建topic ###ip位本机地址###ip位本机地址 bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic bc_data_return --create --replication-factor 1 --partitions 1 bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic equipment_bc_data_return --create --replication-factor 1 --partitions 1查看topic详细信息 bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic bc_data_return --describe 三、Kafka生产者 3.1 生产者消息发送流程 3.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。 3.2 异步发送API 3.2.1 普通异步发送1）需求：创建Kafka生产者，采用异步的方式发送到Kafka Broker
2）代码编写 （1）创建工程 kafka （2）导入依赖 org.apache.kafka kafka-clients 3.0.0（3）创建包名：com.atguigu.kafka.producer （4）编写不带回调函数的 API 代码 public class CustomProducer { private final static String SERVER_CONFIG = "localhost:9092"; public static void main(String[] args) { // 1. 创建 kafka 生产者的配置对象 Properties properties = new Properties(); // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 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 kafkaProducer = new KafkaProducer(properties); // 4. 调用 send 方法,发送消息 for (int i = 0; i < 5; i++) { kafkaProducer.send(new ProducerRecord("first", "xiang " + i)); } // 5. 关闭资源 kafkaProducer.close(); } }3）测试 开启 Kafka 消费者: [root@localhost kafka]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first xiang 0 xiang 1 xiang 2 xiang 3 xiang 4 3.2.2 带回调函数的异步发送回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是元数据信息（RecordMetadata）和异常信息（Exception），如果 Exception 为 null，说明消息发送成功，如果 Exception 不为 null，说明消息发送失败。 注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。 public static void main(String[] args) { // 1. 创建 kafka 生产者的配置对象 Properties properties = new Properties(); // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 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 kafkaProducer = new KafkaProducer(properties); // 4. 调用 send 方法,发送消息 for (int i = 0; i < 5; i++) { kafkaProducer.send(new ProducerRecord("first", "xiang " + i), new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception == null) { // 没有异常,输出信息到控制台 System.out.println(" 主题： " + metadata.topic() + "->" + "分区：" + metadata.partition()); } else { // 出现异常打印 exception.printStackTrace(); } } }); } // 5. 关闭资源 kafkaProducer.close(); }测试:
开启 Kafka 消费者: [root@localhost kafka]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first xiang 0 xiang 1 xiang 2 xiang 3 xiang 4 3.3 同步发送API只需在异步发送的基础上，再调用一下 get()方法即可。 public static void main(String[] args) throws ExecutionException, InterruptedException { // 1. 创建 kafka 生产者的配置对象 Properties properties = new Properties(); // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 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 kafkaProducer = new KafkaProducer(properties); // 4. 调用 send 方法,发送消息 for (int i = 0; i < 5; i++) { RecordMetadata metadata = kafkaProducer.send(new ProducerRecord("first", "xiang " + i)).get(); System.out.println(" 主题： " + metadata.topic() + "->" + "分区：" + metadata.partition()); } // 5. 关闭资源 kafkaProducer.close(); } 3.4 生产者分区 3.4.1 分区好处（1）便于合理使用存储资源，每个Partition在一个Broker上存储，可以把海量的数据按照分区切割成一块一块数据存储在多台Broker上。合理控制分区的任务，可以实现负载均衡的效果。 （2）提高并行度，生产者可以以分区为单位发送数据；消费者可以以分区为单位进行消费数据。
1）默认的分区器 DefaultPartitioner
2）案例一：将数据发往指定 partition 例如，将所有数据发往分区 0 中 public static void main(String[] args) { // 1. 创建 kafka 生产者的配置对象 Properties properties = new Properties(); // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 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 kafkaProducer = new KafkaProducer(properties); // 4. 调用 send 方法,发送消息 for (int i = 0; i < 5; i++) { kafkaProducer.send(new ProducerRecord("first", 0, "", "xiang " + i), new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception == null) { // 没有异常,输出信息到控制台 System.out.println(" 主题： " + metadata.topic() + "->" + "分区：" + metadata.partition()); } else { // 出现异常打印 exception.printStackTrace(); } } }); } // 5. 关闭资源 kafkaProducer.close(); }
3）案例二：没有指明 partition 值但有 key 的情况下 没有指明 partition 值但有 key 的情况下，将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值。 public static void main(String[] args) { // 1. 创建 kafka 生产者的配置对象 Properties properties = new Properties(); // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 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 kafkaProducer = new KafkaProducer(properties); // 4. 调用 send 方法,发送消息 for (int i = 0; i < 5; i++) { kafkaProducer.send(new ProducerRecord("first", i + "", "xiang " + i), new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception == null) { // 没有异常,输出信息到控制台 System.out.println(" 主题： " + metadata.topic() + "->" + "分区：" + metadata.partition()); } else { // 出现异常打印 exception.printStackTrace(); } } }); } // 5. 关闭资源 kafkaProducer.close(); }
如果研发人员可以根据企业需求，自己重新实现分区器。 1）需求 例如我们实现一个分区器实现，发送过来的数据中如果包含 xiang，就发往 0 号分区，不包含 xiang，就发往 1 号分区。 2）实现步骤 定义类实现 Partitioner 接口。 public class MyPartitioner implements Partitioner { @Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { // 获取消息 String msgValue = value.toString(); // 创建 partition int partition; // 判断消息是否包含 xiang if (msgValue.contains("xiang")) { partition = 0; } else { partition = 1; } // 返回分区号 return partition; } @Override public void close() { } @Override public void configure(Map configs) { } }（3）使用分区器的方法，在生产者的配置中添加分区器参数。 public static void main(String[] args) { // 1. 创建 kafka 生产者的配置对象 Properties properties = new Properties(); // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 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()); //自定义partition properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class.getName()); // 3. 创建 kafka 生产者对象 KafkaProducer kafkaProducer = new KafkaProducer(properties); // 4. 调用 send 方法,发送消息 for (int i = 0; i < 5; i++) { kafkaProducer.send(new ProducerRecord("first", "xiang " + i), new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception == null) { // 没有异常,输出信息到控制台 System.out.println(" 主题： " + metadata.topic() + "->" + "分区：" + metadata.partition()); } else { // 出现异常打印 exception.printStackTrace(); } } }); } // 5. 关闭资源 kafkaProducer.close(); } 3.5 生产经验-提高吞吐量
设置批次大小、等待时间等： public static void main(String[] args) { // 1. 创建 kafka 生产者的配置对象 Properties properties = new Properties(); // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 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()); // 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 kafkaProducer = new KafkaProducer(properties); // 4. 调用 send 方法,发送消息 for (int i = 0; i < 5; i++) { kafkaProducer.send(new ProducerRecord("first", "xiang " + i)); } // 5. 关闭资源 kafkaProducer.close(); } 3.6 生产经验-数据可靠性0）回顾发送流程
1）ack 应答原理
数据完全可靠条件 = ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2 可靠性总结： acks=0，生产者发送过来数据就不管了，可靠性差，效率高；acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等；acks=-1，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低；在生产环境中，acks=0很少使用；acks=1，一般用于传输普通日志，允许丢个别数据；acks=-1，一般用于传输和钱相关的数据，对可靠性要求比较高的场景。 数据重复分析： acks: -1(all) 生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答
2）代码配置 //ack方式 properties.put(ProducerConfig.ACKS_CONFIG, "all"); //重试次数 properties.put(ProducerConfig.RETRIES_CONFIG, 3); //... //发送 kafkaProducer.send(newProducerRecord("first","atguigu " + i)); 3.7 生产经验-数据去重 3.7.1 数据传递语义至少一次（At Least Once） ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2 最多一次（At Most Once） ACK级别设置为0 总结： At Least Once可以保证数据不丢失，但是不能保证数据不重复；At Most Once可以保证数据不重复，但是不能保证数据不丢失。精确一次（Exactly Once） 对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。Kafka 0.11版本以后，引入了一项重大特性：幂等性和事务。 3.7.2 幂等性1）幂等性原理 幂等性：就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证了不重复。 精确一次（Exactly Once）： 幂等性 + 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数量>=2） 。 重复数据的判断标准：具有相同主键的消息提交时，Broker只会持久化一条。 PID是Kafka每次重启都会分配一个新的；Partition 表示分区号；Sequence Number是单调自增的。所以幂等性只能保证的是在单分区单会话内不重复。
2）如何使用幂等性 开启参数 enable.idempotence 默认为 true，false 关闭。 3.7.3 生产者事务1）Kafka 事务原理 说明：开启事务，必须开启幂等性。
2）Kafka 的事务一共有如下 5 个 API // 1 初始化事务 void initTransactions(); // 2 开启事务 void beginTransaction() throws ProducerFencedException; // 3 在事务内提交已经消费的偏移量（主要用于消费者） void sendOffsetsToTransaction(Map offsets,String consumerGroupId) throws ProducerFencedException; // 4 提交事务 void commitTransaction() throws ProducerFencedException; // 5 放弃事务（类似于回滚事务的操作） void abortTransaction() throws ProducerFencedException;3）单个 Producer，使用事务保证消息的仅一次发送 public static void main(String[] args) { // 1. 创建 kafka 生产者的配置对象 Properties properties = new Properties(); // 2. 给 kafka 配置对象添加配置信息 properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 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 kafkaProducer = new KafkaProducer(properties); // 初始化事务 kafkaProducer.initTransactions(); // 开启事务 kafkaProducer.beginTransaction(); try { // 4. 调用 send 方法,发送消息 for (int i = 0; i < 5; i++) { // 发送消息 kafkaProducer.send(new ProducerRecord("first", "xiang " + i)); } // int i = 1 / 0; // 提交事务 kafkaProducer.commitTransaction(); } catch (Exception e) { // 终止事务 kafkaProducer.abortTransaction(); } finally { // 5. 关闭资源 kafkaProducer.close(); } } 3.7 生产经验-数据有序
3.8 生产经验-数据乱序
1）kafka在1.x版本之前保证数据单分区有序，条件如下： max.in.flight.requests.per.connection=1（不需要考虑是否开启幂等性）。 2）kafka在1.x及以后版本保证数据单分区有序，条件如下： （1）未开启幂等性 max.in.flight.requests.per.connection需要设置为1。 （2）开启幂等性 max.in.flight.requests.per.connection需要设置小于等于5。 原因说明：因为在kafka1.x以后，启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，故无论如何，都可以保证最近5个request的数据都是有序的。
（1）启动 Zookeeper 客户端。 [atguigu@hadoop102 zookeeper-3.5.7]$ bin/zkCli.sh（2）通过 ls 命令可以查看 kafka 相关信息。 [zk: localhost:2181(CONNECTED) 2] ls /kafka
模拟 Kafka 上下线，Zookeeper 中数据变化 （1）查看/kafka/brokers/ids 路径上的节点。 [zk: localhost:2181(CONNECTED) 2] ls /kafka/brokers/ids [0, 1, 2]（2）查看/kafka/controller 路径上的数据。 [zk: localhost:2181(CONNECTED) 15] get /kafka/controller {"version":1,"brokerid":0,"timestamp":"1637292471777"}（3）查看/kafka/brokers/topics/first/partitions/0/state 路径上的数据。 [zk: localhost:2181(CONNECTED) 16] get /kafka/brokers/topics/first/partitions/0/state {"controller_epoch":24,"leader":0,"version":1,"leader_epoch":18,"isr":[0,1,2]}（4）停止 hadoop104 上的 kafka。 [atguigu@hadoop104 kafka]$ bin/kafka-server-stop.sh（5）再次查看/kafka/brokers/ids 路径上的节点。 [zk: localhost:2181(CONNECTED) 3] ls /kafka/brokers/ids [0, 1]（6）再次查看/kafka/controller 路径上的数据。 [zk: localhost:2181(CONNECTED) 15] get /kafka/controller {"version":1,"brokerid":0,"timestamp":"1637292471777"}（7）再次查看/kafka/brokers/topics/first/partitions/0/state 路径上的数据。 [zk: localhost:2181(CONNECTED) 16] get /kafka/brokers/topics/first/partitions/0/state {"controller_epoch":24,"leader":0,"version":1,"leader_epoch":18,"isr":[0,1]}（8）启动 hadoop104 上的 kafka。 [atguigu@hadoop104 kafka]$ bin/kafka-server-start.sh -daemon ./config/server.properties（9）再次观察（1）、（2）、（3）步骤中的内容。 4.1.3 Broker 重要参数参数名称 描述 replica.lag.time.max.ms ISR 中，如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值，默认 30s。 auto.leader.rebalance.enable 默认是 true。 自动 Leader Partition 平衡。 leader.imbalance.per.broker.percentage 默认是 10%。每个 broker 允许的不平衡的 leader的比率。如果每个 broker 超过了这个值，控制器会触发 leader 的平衡。 leader.imbalance.check.interval.seconds 默认值 300 秒。检查 leader 负载是否平衡的间隔时间。 log.segment.bytes Kafka 中 log 日志是分成一块块存储的，此配置是指 log 日志划分 成块的大小，默认值 1G。 log.index.interval.bytes 默认 4kb，kafka 里面每当写入了 4kb 大小的日志（.log），然后就往 index 文件里面记录一个索引。 log.retention.hours Kafka 中数据保存的时间，默认 7 天。 log.retention.minutes Kafka 中数据保存的时间，分钟级别，默认关闭。 log.retention.ms Kafka 中数据保存的时间，毫秒级别，默认关闭。 log.retention.check.interval.ms 检查数据是否保存超时的间隔，默认是 5 分钟。 log.retention.bytes 默认等于-1，表示无穷大。超过设置的所有日志总大小，删除最早的 segment。 log.cleanup.policy 默认是 delete，表示所有数据启用删除策略；如果设置值为 compact，表示所有数据启用压缩策略。 num.io.threads 默认是 8。负责写磁盘的线程数。整个参数值要占总核数的 50%。 num.replica.fetchers 副本拉取线程数，这个参数占总核数的 50%的 1/3 num.network.threads 默认是 3。数据传输线程数，这个参数占总核数的50%的 2/3 。 log.flush.interval.messages 强制页缓存刷写到磁盘的条数，默认是 long 的最大值，9223372036854775807。一般不建议修改，交给系统自己管理。 log.flush.interval.ms 每隔多久，刷数据到磁盘，默认是 null。一般不建议修改，交给系统自己管理。 4.2 生产经验- 节点服役和退役 4.2.1 服役新节点1）新节点准备 （1）关闭 hadoop104，并右键执行克隆操作。 （2）开启 hadoop105，并修改 IP 地址。 [root@hadoop104 ~]# vim /etc/sysconfig/network-scripts/ifcfgens33（3）在 hadoop105 上，修改主机名称为 hadoop105。 [root@hadoop104 ~]# vim /etc/hostname（4）重新启动 hadoop104、hadoop105。 （5）修改 haodoop105 中 kafka 的 broker.id 为 3。 （6）删除 hadoop105 中 kafka 下的 datas 和 logs。 [atguigu@hadoop105 kafka]$ rm -rf datas/* logs/*（7）启动 hadoop102、hadoop103、hadoop104 上的 kafka 集群。 [atguigu@hadoop102 ~]$ zk.sh start [atguigu@hadoop102 ~]$ kf.sh start（8）单独启动 hadoop105 中的 kafka。 [atguigu@hadoop105 kafka]$ bin/kafka-server-start.sh -daemon ./config/server.properties2）执行负载均衡操作 （1）创建一个要均衡的主题。 [atguigu@hadoop102 kafka]$ vim topics-to-move.json { "topics":[ { "topic":"first" } ], "version":1 }（2）生成一个负载均衡的计划。 [atguigu@hadoop102 kafka]$ 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 Current partition replica assignment {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[0,2,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[2,1,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,0,2],"log_dirs":["any","any","any"]}]} Proposed partition reassignment configuration {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,3,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,1,2],"log_dirs":["any","any","any"]}]}（3）创建副本存储计划（所有副本存储在 broker0、broker1、broker2、broker3 中）。 [atguigu@hadoop102 kafka]$ vim increase-replication-factor.json {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,3,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,1,2],"log_dirs":["any","any","any"]}]}（4）执行副本存储计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute（5）验证副本存储计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify Status of partition reassignment: Reassignment of partition first-0 is complete. Reassignment of partition first-1 is complete. Reassignment of partition first-2 is complete. Clearing broker-level throttles on brokers 0,1,2,3 Clearing topic-level throttles on topic first 4.2.2 退役旧节点1）执行负载均衡操作 先按照退役一台节点，生成执行计划，然后按照服役时操作流程执行负载均衡。 （1）创建一个要均衡的主题。 [atguigu@hadoop102 kafka]$ vim topics-to-move.json { "topics": [ {"topic": "first"} ], "version": 1 }（2）创建执行计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2" --generate Current partition replica assignment {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,2,3],"log_dirs":["any","any","any"]}]} Proposed partition reassignment configuration {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,2,0],"log_dirs":["any","any","any"]}]}（3）创建副本存储计划（所有副本存储在 broker0、broker1、broker2 中）。 [atguigu@hadoop102 kafka]$ vim increase-replication-factor.json {"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[1,2,0],"log_dirs":["any","any","any"]}]}（4）执行副本存储计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute（5）验证副本存储计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify Status of partition reassignment: Reassignment of partition first-0 is complete. Reassignment of partition first-1 is complete. Reassignment of partition first-2 is complete. Clearing broker-level throttles on brokers 0,1,2,3 Clearing topic-level throttles on topic first2）执行停止命令 在 hadoop105 上执行停止命令即可。 [atguigu@hadoop105 kafka]$ bin/kafka-server-stop.sh 4.3 Kafka 副本 4.3.1 副本基本信息（1）Kafka 副本作用：提高数据可靠性。 （2）Kafka 个数：默认副本 1 个，生产环境一般配置为 2 个，保证数据可靠性；太多副本会增加磁盘存储空间，增加网络上数据传输，降低效率。 （3）Kafka 中副本类型：分为Leader 和 Follower。Kafka 生产者只会把数据发往 Leader，然后 Follower 找 Leader 进行同步数据。 （4）Kafka 分区中的所有副本统称为 AR（Assigned Repllicas）。 AR = ISR + OSR ISR，表示和 Leader 保持同步的 Follower 集合。如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值由 replica.lag.time.max.ms参数设定，默认 30s。Leader 发生故障之后，就会从 ISR 中选举新的 Leader。 OSR，表示 Follower 与 Leader 副本同步时，延迟过多的副本。 4.3.2 Leader选举流程Kafka 集群中有一个 broker 的 Controller 会被选举为 Controller Leader，负责管理集群broker 的上下线，所有 topic 的分区副本分配和 Leader 选举等工作。 Controller 的信息同步工作是依赖于 Zookeeper 的。
（1）创建一个新的 topic，4 个分区，4 个副本 [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic atguigu1 --partitions 4 --replication-factor 4 Created topic atguigu1.（2）查看 Leader 分布情况 [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu1 Topic: atguigu1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824 Topic: atguigu1 Partition: 0 Leader: 3 Replicas: 3,0,2,1 Isr: 3,0,2,1 Topic: atguigu1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,2,3,0 Topic: atguigu1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,3,1,2 Topic: atguigu1 Partition: 3 Leader: 2 Replicas: 2,1,0,3 Isr: 2,1,0,3（3）停止掉 hadoop105 的 kafka 进程，并查看 Leader 分区情况 [atguigu@hadoop105 kafka]$ bin/kafka-server-stop.sh [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu1 Topic: atguigu1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824 Topic: atguigu1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,2,1 Topic: atguigu1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,2,0 Topic: atguigu1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,1,2 Topic: atguigu1 Partition: 3 Leader: 2 Replicas: 2,1,0,3 Isr: 2,1,0（4）停止掉 hadoop104 的 kafka 进程，并查看 Leader 分区情况 [atguigu@hadoop104 kafka]$ bin/kafka-server-stop.sh [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu1 Topic: atguigu1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824 Topic: atguigu1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,1 Topic: atguigu1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,0 Topic: atguigu1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,1 Topic: atguigu1 Partition: 3 Leader: 1 Replicas: 2,1,0,3 Isr: 1,0（5）启动 hadoop105 的 kafka 进程，并查看 Leader 分区情况 [atguigu@hadoop105 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu1 Topic: atguigu1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824 Topic: atguigu1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,1,3 Topic: atguigu1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,0,3 Topic: atguigu1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,1,3 Topic: atguigu1 Partition: 3 Leader: 1 Replicas: 2,1,0,3 Isr: 1,0,3（6）启动 hadoop104 的 kafka 进程，并查看 Leader 分区情况 [atguigu@hadoop104 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu1 Topic: atguigu1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824 Topic: atguigu1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,1,3,2 Topic: atguigu1 Partition: 1 Leader: 1 Replicas: 1,2,3,0 Isr: 1,0,3,2 Topic: atguigu1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,1,3,2 Topic: atguigu1 Partition: 3 Leader: 1 Replicas: 2,1,0,3 Isr: 1,0,3,2（7）停止掉 hadoop103 的 kafka 进程，并查看 Leader 分区情况 [atguigu@hadoop103 kafka]$ bin/kafka-server-stop.sh [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic atguigu1 Topic: atguigu1 TopicId: awpgX_7WR-OX3Vl6HE8sVg PartitionCount: 4 ReplicationFactor: 4 Configs: segment.bytes=1073741824 Topic: atguigu1 Partition: 0 Leader: 0 Replicas: 3,0,2,1 Isr: 0,3,2 Topic: atguigu1 Partition: 1 Leader: 2 Replicas: 1,2,3,0 Isr: 0,3,2 Topic: atguigu1 Partition: 2 Leader: 0 Replicas: 0,3,1,2 Isr: 0,3,2 Topic: atguigu1 Partition: 3 Leader: 2 Replicas: 2,1,0,3 Isr: 0,3,2节点下线后再上线，容易造成不平衡，注意及时调整。 4.3.3 Leader 和 Follower故障处理细节概念 LEO（Log End Offset）: 每个副本的最后一个offset，LEO其实就是最新的offset + 1。HW（High Watermark）: 所有副本中最小的LEO 。Follower故障处理细节
Leader故障处理细节
注意：这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。 4.3.4 分区副本分配如果 kafka 服务器只有 4 个节点，那么设置 kafka 的分区数大于服务器台数，在 kafka底层如何分配存储副本呢？ 创建 16 分区，3 个副本 （1）创建一个新的 topic，名称为 second。 [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 16 --replication-factor 3 --topic second（2）查看分区和副本情况。 [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic second Topic: second4 Partition: 0 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2 Topic: second4 Partition: 1 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3 Topic: second4 Partition: 2 Leader: 2 Replicas: 2,3,0 Isr: 2,3,0 Topic: second4 Partition: 3 Leader: 3 Replicas: 3,0,1 Isr: 3,0,1 Topic: second4 Partition: 4 Leader: 0 Replicas: 0,2,3 Isr: 0,2,3 Topic: second4 Partition: 5 Leader: 1 Replicas: 1,3,0 Isr: 1,3,0 Topic: second4 Partition: 6 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1 Topic: second4 Partition: 7 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2 Topic: second4 Partition: 8 Leader: 0 Replicas: 0,3,1 Isr: 0,3,1 Topic: second4 Partition: 9 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2 Topic: second4 Partition: 10 Leader: 2 Replicas: 2,1,3 Isr: 2,1,3 Topic: second4 Partition: 11 Leader: 3 Replicas: 3,2,0 Isr: 3,2,0 Topic: second4 Partition: 12 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2 Topic: second4 Partition: 13 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3 Topic: second4 Partition: 14 Leader: 2 Replicas: 2,3,0 Isr: 2,3,0 Topic: second4 Partition: 15 Leader: 3 Replicas: 3,0,1 Isr: 3,0,1
在生产环境中，每台服务器的配置和性能不一致，但是Kafka只会根据自己的代码规则创建对应的分区副本，就会导致个别服务器存储压力较大。所有需要手动调整分区副本的存储 需求：创建一个新的topic，4个分区，两个副本，名称为three。将 该topic的所有副本都存储到broker0和broker1两台服务器上。
手动调整分区副本存储的步骤如下： （1）创建一个新的 topic，名称为 three。 [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 4 --replication-factor 2 --topic three（2）查看分区副本存储情况。 [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic three（3）创建副本存储计划（所有副本都指定存储在 broker0、broker1 中）。 [atguigu@hadoop102 kafka]$ 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]}] }（4）执行副本存储计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute（5）验证副本存储计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify（6）查看分区副本存储情况。 [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic three 4.3.6 生产经验-Leader Partition 负载平衡正常情况下，Kafka本身会自动把Leader Partition均匀分散在各个机器上，来保证每台机器的读写吞吐量都是均匀的。但是如果某些broker宕机，会导致Leader Partition过于集中在其他少部分几台broker上，这会导致少数几台broker的读写请求压力过高，其他宕机的broker重启之后都是follower partition，读写请求很低，造成集群负载不均衡。
几个概念： auto.leader.rebalance.enable，默认是true。自动Leader Partition 平衡leader.imbalance.per.broker.percentage，默认是10%。每个broker允许的不平衡的leader的比率。如果每个broker超过了这个值，控制器会触发leader的平衡。leader.imbalance.check.interval.seconds，默认值300秒。检查leader负载是否平衡的间隔时间。下面拿一个主题举例说明，假设集群只有一个主题如下图所示：
针对broker0节点，分区2的AR优先副本是0节点，但是0节点却不是Leader节点，所以不平衡数加1，AR副本总数是4 所以broker0节点不平衡率为1/4>10%，需要再平衡。broker2和broker3节点和broker0不平衡率一样，需要再平衡。 Broker1的不平衡数为0，不需要再平衡。 参数名称 描述 auto.leader.rebalance.enable 默认是 true。 自动 Leader Partition 平衡。生产环境中，leader 重选举的代价比较大，可能会带来性能影响，建议设置为 false 关闭。 leader.imbalance.per.broker.percentage 默认是 10%。每个 broker 允许的不平衡的 leader的比率。如果每个 broker 超过了这个值，控制器会触发 leader 的平衡。 leader.imbalance.check.interval.seconds 默认值 300 秒。检查 leader 负载是否平衡的间隔时间。 4.3.7 生产经验-增加副本因子在生产环境当中，由于某个主题的重要等级需要提升，我们考虑增加副本。副本数的增加需要先制定计划，然后根据计划执行。 1）创建 topic [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 3 --replication-factor 1 --topic four2）手动增加副本存储 （1）创建副本存储计划（所有副本都指定存储在 broker0、broker1、broker2 中）。 [atguigu@hadoop102 kafka]$ 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 ] } ] }（2）执行副本存储计划。 [atguigu@hadoop102 kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute 4.4 文件存储 4.4.1 文件存储机制1）Topic 数据的存储机制 Topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是Producer生产的数据。Producer生产的数据会被不断追加到该log文件末端，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制，将每个partition分为多个segment。每个segment包括：“.index”文件、“.log”文件和.timeindex等文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号，例如：first-0。 一个topic分为多个partition一个partition分为多个segment.log 日志文件.index 偏移量索引文件.timeindex 时间戳索引文件其他文件说明：index和log文件以当前segment的第一条消息的offset命名。
2）Topic 数据到底存储在什么位置？ （1）启动生产者，并发送消息。 [atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first >hello world（2）查看 hadoop102（或者 hadoop103、hadoop104的/opt/module/kafka/datas/first-1（first-0、first-2）路径上的文件。 [atguigu@hadoop104 first-1]$ ls 00000000000000000092.index 00000000000000000092.log 00000000000000000092.snapshot 00000000000000000092.timeindex leader-epoch-checkpoint partition.metadata（3）直接查看 log 日志，发现是乱码。 [atguigu@hadoop104 first-1]$ cat 00000000000000000092.log \CYnF|©|©ÿ"hello world（4）通过工具查看 index 和 log 信息。 [atguigu@hadoop104 first-1]$ kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.index Dumping ./00000000000000000000.index offset: 3 position: 152 [atguigu@hadoop104 first-1]$ kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.log Dumping datas/first-0/00000000000000000000.log Starting offset: 0 baseOffset: 0 lastOffset: 1 count: 2 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position:0 CreateTime: 1636338440962 size: 75 magic: 2 compresscodec: none crc: 2745337109 isvalid:true baseOffset: 2 lastOffset: 2 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 75 CreateTime: 1636351749089 size: 77 magic: 2 compresscodec: none crc: 273943004 isvalid: true baseOffset: 3 lastOffset: 3 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 152 CreateTime: 1636351749119 size: 77 magic: 2 compresscodec: none crc: 106207379 isvalid: true baseOffset: 4 lastOffset: 8 count: 5 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 229 CreateTime: 1636353061435 size: 141 magic: 2 compresscodec: none crc: 157376877 isvalid: true baseOffset: 9 lastOffset: 13 count: 5 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 370 CreateTime: 1636353204051 size: 146 magic: 2 compresscodec: none crc: 4058582827 isvalid: true3）index 文件和 log 文件详解 如何在log文件中定位到offset=600的Record？ 根据目标offset定位Segment文件找到小于等于目标offset的最大offset对应的索引项定位到log文件向下遍历找到目标Record
注意： index为稀疏索引，大约每往log文件写入4kb数据，会往index文件写入一条索引。参数log.index.interval.bytes默认4kb。2.Index文件中保存的offset为相对offset，这样能确保offset的值所占空间不会过大，因此能将offset的值控制在固定大小说明：日志存储参数配置 参数 描述 log.segment.bytes Kafka 中 log 日志是分成一块块存储的，此配置是指 log 日志划分成块的大小，默认值 1G。 log.segment.bytes 默认 4kb，kafka 里面每当写入了 4kb 大小的日志（.log），然后就往 index 文件里面记录一个索引。 稀疏索引。 4.4.2 文件清理策略Kafka 中默认的日志保存时间为 7 天，可以通过调整如下参数修改保存时间。 log.retention.hours，最低优先级小时，默认 7 天。log.retention.minutes，分钟。log.retention.ms，最高优先级毫秒。log.retention.check.interval.ms，负责设置检查周期，默认 5 分钟。那么日志一旦超过了设置的时间，怎么处理呢？Kafka 中提供的日志清理策略有 delete 和 compact 两种。 1）delete 日志删除：将过期数据删除 log.cleanup.policy = delete 所有数据启用删除策略基于时间：默认打开。以 segment 中所有记录中的最大时间戳作为该文件时间戳。基于大小：默认关闭。超过设置的所有日志总大小，删除最早的 segment log.retention.bytes，默认等于-1，表示无穷大。
2）compact 日志压缩 compact日志压缩：对于相同key的不同value值，只保留最后一个版本。 log.cleanup.policy = compact 所有数据启用压缩策略压缩之前的数据和压缩之后的数据:
压缩后的offset可能是不连续的，比如上图中没有6，当从这些offset消费消息时，将会拿到比这个offset大的offset对应的消息，实际上会拿到offset为7的消息，并从这个位置开始消费。 这种策略只适合特殊场景，比如消息的key是用户ID，value是用户的资料，通过这种压缩策略，整个消息集里就保存了所有用户最新的资料。 4.5 高效读写数据1）Kafka 本身是分布式集群，可以采用分区技术，并行度高 2）读数据采用稀疏索引，可以快速定位要消费的数据 3）顺序写磁盘 Kafka 的 producer 生产数据，要写入到 log 文件中，写的过程是一直追加到文件末端，为顺序写。官网有数据表明，同样的磁盘，顺序写能到 600M/s，而随机写只有 100K/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。
4）页缓存 + 零拷贝技术 零拷贝：Kafka的数据加工处理操作交由Kafka生产者和Kafka消费者处理。Kafka Broker应用层不关心存储的数据，所以就不用走应用层，传输效率高。 PageCache页缓存：Kafka重度依赖底层操作系统提供的PageCache功 能。当上层有写操作时，操作系统只是将数据写入PageCache。当读操作发生时，先从PageCache中查找，如果找不到，再去磁盘中读取。实际上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来使用。
参数 描述 log.flush.interval.messages 强制页缓存刷写到磁盘的条数，默认是 long 的最大值，9223372036854775807。一般不建议修改，交给系统自己管理。 log.flush.interval.ms 每隔多久，刷数据到磁盘，默认是 null。一般不建议修改，交给系统自己管理。 五、Kafka消费者 5.1 Kafka消费方式pull（拉）模 式： consumer采用从broker中主动拉取数据。Kafka采用这种方式。 push（推）模式： Kafka没有采用这种方式，因为由broker决定消息发送速率，很难适应所有消费者的消费速率。 例如推送的速度是50m/s，Consumer1、Consumer2就来不及处理消息。
pull模式不足之处是，如 果Kafka没有数据，消费者可能会陷入循环中，一直返回空数据。 5.2 Kafka消费者工作流程 5.2.1 消费者总体工作流程
Consumer Group（CG）：消费者组，由多个consumer组成。 形成一个消费者组的条件，是所有消费者的groupid相同。 消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费。 消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
如果向消费组中添加更多的消费者，超过主题分区数量，则有一部分消费者就会闲置，不会接收任何消息。 消费者组初始化流程 coordinator：辅助实现消费者组的初始化和分区的分配。 coordinator节点选择 = groupid的hashcode值 % 50（ __consumer_offsets的分区数量）。 例如： groupid的hashcode值 = 1，1% 50 = 1，那么__consumer_offsets 主题的1号分区，在哪个broker上，就选择这个节点的coordinator作为这个消费者组的老大。消费者组下的所有的消费者提交offset的时候就往这个分区去提交offset。
消费者组详细消费流程
参数名称 描述 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 （brokerconfig）or max.message.bytes （topic config）影响。 max.poll.records 一次 poll 拉取数据返回消息的最大条数，默认是 500 条。 5.3 消费者API 5.3.1 独立消费者案例（订阅主题）1）需求：创建一个独立消费者，消费 first 主题中数据。
注意：在消费者 API 代码中必须配置消费者组 id。命令行启动消费者不填写消费者组id 会被自动填写随机的消费者组 id。 2）实现步骤 public static void main(String[] args) { // 1.创建消费者的配置对象 Properties properties = new Properties(); // 2.给消费者配置对象添加参数 properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 配置序列化 必须 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"); // 注册要消费的主题（可以消费多个主题） ArrayList topics = new ArrayList(); topics.add("first"); // 创建消费者对象 KafkaConsumer kafkaConsumer = new KafkaConsumer(properties); kafkaConsumer.subscribe(topics); // 拉取数据打印 while (true) { // 设置 1s 中消费一批数据 ConsumerRecords consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1)); // 打印消费到的数据 for (ConsumerRecord consumerRecord : consumerRecords) { System.out.println(consumerRecord); } } } 5.3.2 独立消费者案例（订阅分区）1）需求：创建一个独立消费者，消费 first 主题 0 号分区的数据。
2）实现步骤 public static void main(String[] args) { // 1.创建消费者的配置对象 Properties properties = new Properties(); // 2.给消费者配置对象添加参数 properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 配置序列化 必须 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"); // 注册要消费的主题（可以消费多个主题） ArrayList topicPartitions = new ArrayList(); topicPartitions.add(new TopicPartition("first", 0)); // 创建消费者对象 KafkaConsumer kafkaConsumer = new KafkaConsumer(properties); kafkaConsumer.assign(topicPartitions); // 拉取数据打印 while (true) { // 设置 1s 中消费一批数据 ConsumerRecords consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1)); // 打印消费到的数据 for (ConsumerRecord consumerRecord : consumerRecords) { System.out.println(consumerRecord); } } } 5.3.3 消费者组案例1）需求：测试同一个主题的分区数据，只能由一个消费者组中的一个消费。
2）案例实操 复制一份基础消费者的代码，在 IDEA 中同时启动，即可启动同一个消费者组中的两个消费者。 public static void main(String[] args) { // 1.创建消费者的配置对象 Properties properties = new Properties(); // 2.给消费者配置对象添加参数 properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 配置序列化 必须 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"); // 注册要消费的主题（可以消费多个主题） ArrayList topics = new ArrayList(); topics.add("first"); // 创建消费者对象 KafkaConsumer kafkaConsumer = new KafkaConsumer(properties); kafkaConsumer.subscribe(topics); // 拉取数据打印 while (true) { // 设置 1s 中消费一批数据 ConsumerRecords consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1)); // 打印消费到的数据 for (ConsumerRecord consumerRecord : consumerRecords) { System.out.println(consumerRecord); } } } 5.4 生产经验-分区的分配以及再平衡1、一个consumer group中有多个consumer组成，一个 topic有多个partition组成，现在的问题是，到底由哪个consumer来消费哪个partition的数据。 2、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 5.4.1 Range以及再平衡1）Range 分区策略原理 Range是对每个topic而言的。 首先对同一个topic里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。 假如现在有7个分区，3个消费者，排序后的分区将会是0,1,2,3,4,5,6；消费者排序完之后将会是C0,C1,C2。 通过partitions数/consumer数来决定每个消费者应该消费几个分区。如果除不尽，那么前面几个消费者将会多消费1个分区。 例如，7/3=2余1，除不尽，那么消费者C0便会多消费1个分区。8/3=2余2，除不尽，那么C0和C1分别多消费一个。 注意：如果只是针对1个topic而言，C0消费者多消费1个分区影响不是很大。但是如果有N多个topic，那么针对每个topic，消费者C0都将多消费1个分区，topic越多，C0消费的分区会比其他消费者明显显多消费N个分区。 容易产生数据倾斜！
2）Range 分区分配策略案例 （1）修改主题 first 为 7 个分区。 [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-serverhadoop102:9092 --alter --topic first --partitions 7注意：分区数可以增加，但是不能减少。 （2）复制 CustomConsumer 类，创建 CustomConsumer2。这样可以由三个消费者CustomConsumer、CustomConsumer1、CustomConsumer2 组成消费者组，组名都为“test”，同时启动 3 个消费者。 （3）启动 CustomProducer 生产者，发送 500 条消息，随机发送到不同的分区。 说明：Kafka 默认的分区分配策略就是 Range + CooperativeSticky，所以不需要修改策略。 （4）观看 3 个消费者分别消费哪些分区的数据。
3）Range 分区分配再平衡案例 （1）停止掉 0 号消费者，快速重新发送消息观看结果（45s 以内，越快越好）。 1 号消费者：消费到 3、4 号分区数据。 2 号消费者：消费到 5、6 号分区数据。 0 号消费者的任务会整体被分配到 1 号消费者或者 2 号消费者。 说明：0 号消费者挂掉后，消费者组需要按照超时时间 45s 来判断它是否退出，所以需要等待，时间到了 45s 后，判断它真的退出就会把任务分配给其他 broker 执行。 （2）再次重新发送消息观看结果（45s 以后）。 1 号消费者：消费到 0、1、2、3 号分区数据。 2 号消费者：消费到 4、5、6 号分区数据。 说明：消费者 0 已经被踢出消费者组，所以重新按照 range 方式分配。 5.4.2 RoundRobin以及再平衡1）RoundRobin 分区策略原理 RoundRobin 针对集群中所有Topic而言。 RoundRobin 轮询分区策略，是把所有的 partition 和所有的consumer 都列出来，然后按照 hashcode 进行排序，最后通过轮询算法来分配 partition 给到各个消费者。
2）RoundRobin 分区分配策略案例 （1）依次在 CustomConsumer、CustomConsumer1、CustomConsumer2 三个消费者代码中修改分区分配策略为 RoundRobin。 // 修改分区分配策略 properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");（2）重启 3 个消费者，重复发送消息的步骤，观看分区结果。
3）RoundRobin 分区分配再平衡案例 （1）停止掉 0 号消费者，快速重新发送消息观看结果（45s 以内，越快越好）。 1 号消费者：消费到 2、5 号分区数据 2 号消费者：消费到 4、1 号分区数据 0 号消费者的任务会按照 RoundRobin 的方式，把数据轮询分成 0 、6 和 3 号分区数据，分别由 1 号消费者或者 2 号消费者消费。 说明：0 号消费者挂掉后，消费者组需要按照超时时间 45s 来判断它是否退出，所以需要等待，时间到了 45s 后，判断它真的退出就会把任务分配给其他 broker 执行。 （2）再次重新发送消息观看结果（45s 以后）。 1 号消费者：消费到 0、2、4、6 号分区数据 2 号消费者：消费到 1、3、5 号分区数据 说明：消费者 0 已经被踢出消费者组，所以重新按照 RoundRobin 方式分配。 5.4.3 Sticky以及再平衡粘性分区定义：可以理解为分配的结果带有“粘性的”。即在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可以节省大量的开销。 粘性分区是 Kafka 从 0.11.x 版本开始引入这种分配策略，首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。 1）需求 设置主题为 first，7 个分区；准备 3 个消费者，采用粘性分区策略，并进行消费，观察消费分配情况。然后再停止其中一个消费者，再次观察消费分配情况。 2）步骤 （1）修改分区分配策略为粘性。 注意：3 个消费者都应该注释掉，之后重启 3 个消费者，如果出现报错，全部停止等会再重启，或者修改为全新的消费者组 // 修改分区分配策略 ArrayList startegys = new ArrayList(); startegys.add("org.apache.kafka.clients.consumer.StickyAssignor"); properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, startegys);（2）使用同样的生产者发送 500 条消息。 可以看到会尽量保持分区的个数近似划分分区。
3）Sticky 分区分配再平衡案例 （1）停止掉 0 号消费者，快速重新发送消息观看结果（45s 以内，越快越好）。 1 号消费者：消费到 2、5、3 号分区数据。 2 号消费者：消费到 4、6 号分区数据。 0 号消费者的任务会按照粘性规则，尽可能均衡的随机分成 0 和 1 号分区数据，分别由 1 号消费者或者 2 号消费者消费。 说明：0 号消费者挂掉后，消费者组需要按照超时时间 45s 来判断它是否退出，所以需要等待，时间到了 45s 后，判断它真的退出就会把任务分配给其他 broker 执行。 （2）再次重新发送消息观看结果（45s 以后）。 1 号消费者：消费到 2、3、5 号分区数据。 2 号消费者：消费到 0、1、4、6 号分区数据。 说明：消费者 0 已经被踢出消费者组，所以重新按照粘性方式分配。 5.5 offset位移 5.5.1 offset的默认维护位置Kafka0.9版本之前，consumer默认将offset保存在Zookeeper中. 从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+分区号就保留最新数据。 1）消费 offset 案例 （0）思想：__consumer_offsets 为 Kafka 中的 topic，那就可以通过消费者进行消费。 （1）在配置文件 config/consumer.properties 中添加配置 exclude.internal.topics=false，默认是 true，表示不能消费系统主题。为了查看该系统主题数据，所以该参数修改为 false。 （2）采用命令行方式，创建一个新的 topic。 [atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-serverhadoop102:9092 --create --topic atguigu --partitions 2 --replication-factor 2（3）启动生产者往 atguigu 生产数据。 [atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --topicatguigu --bootstrap-server hadoop102:9092（4）启动消费者消费 atguigu 数据。 [atguigu@hadoop104 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic atguigu --group test注意：指定消费者组名称，更好观察数据存储位置（key 是 group.id+topic+分区号）。 （5）查看消费者消费主题__consumer_offsets。 [atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server hadoop102:9092 --consumer.config config/consumer.properties --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning [offset,atguigu,1]::OffsetAndMetadata(offset=7,leaderEpoch=Optional[0], metadata=, commitTimestamp=1622442520203,expireTimestamp=None) [offset,atguigu,0]::OffsetAndMetadata(offset=8,leaderEpoch=Optional[0], metadata=, commitTimestamp=1622442520203,expireTimestamp=None) 5.5.2 自动提交offset为了使我们能够专注于自己的业务逻辑，Kafka提供了自动提交offset的功能。 自动提交offset的相关参数： enable.auto.commit：是否开启自动提交offset功能，默认是trueauto.commit.interval.ms：自动提交offset的时间间隔，默认是5s
参数名称 描述 enable.auto.commit 默认值为 true，消费者会自动周期性地向服务器提交偏移量。 auto.commit.interval.ms 如果设置了 enable.auto.commit 的值为 true， 则该值定义了消费者偏移量向 Kafka 提交的频率，默认 5s。 1）消费者自动提交 offset public static void main(String[] args) { // 1. 创建 kafka 消费者配置类 Properties properties = new Properties(); // 2. 添加配置参数 // 添加连接 properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 配置序列化 必须 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"); // 是否自动提交 offset properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); // 提交 offset 的时间周期 1000ms，默认 5s properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000); //3. 创建 kafka 消费者 KafkaConsumer consumer = new KafkaConsumer(properties); //4. 设置消费主题 形参是列表 consumer.subscribe(Collections.singletonList("first")); //5. 消费数据 while (true) { // 读取消息 ConsumerRecords consumerRecords = consumer.poll(Duration.ofSeconds(1)); // 输出消息 for (ConsumerRecord consumerRecord : consumerRecords) { System.out.println(consumerRecord.value()); } } } 5.5.3 手动提交offset虽然自动提交offset十分简单便利，但由于其是基于时间提交的，开发人员难以把握offset提交的时机。因此Kafka还提供了手动提交offset的API。 手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。两者的相同点是，都会将本次提交的一批数据最高的偏移量提交；不同点是，同步提交阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；而异步提交则没有失败重试机制，故有可能提交失败。 commitSync（同步提交）：必须等待offset提交完毕，再去消费下一批数据。commitAsync（异步提交） ：发送完提交offset请求后，就开始消费下一批数据了。
1）同步提交 offset由于同步提交 offset 有失败重试机制，故更加可靠，但是由于一直等待提交结果，提交的效率比较低。以下为同步提交 offset 的示例。 public static void main(String[] args) { // 1. 创建 kafka 消费者配置类 Properties properties = new Properties(); // 2. 添加配置参数 // 添加连接 properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 配置序列化 必须 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"); // 是否自动提交 offset properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000); //3. 创建 kafka 消费者 KafkaConsumer consumer = new KafkaConsumer(properties); //4. 设置消费主题 形参是列表 consumer.subscribe(Collections.singletonList("first")); //5. 消费数据 while (true) { // 读取消息 ConsumerRecords consumerRecords = consumer.poll(Duration.ofSeconds(1)); // 输出消息 for (ConsumerRecord consumerRecord : consumerRecords) { System.out.println(consumerRecord.value()); } // 同步提交 offset consumer.commitSync(); } }2）异步提交 offset 虽然同步提交 offset 更可靠一些，但是由于其会阻塞当前线程，直到提交成功。因此吞吐量会受到很大的影响。因此更多的情况下，会选用异步提交 offset 的方式。以下为异步提交 offset 的示例： public static void main(String[] args) { // 1. 创建 kafka 消费者配置类 Properties properties = new Properties(); // 2. 添加配置参数 // 添加连接 properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 配置序列化 必须 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"); // 是否自动提交 offset properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000); //3. 创建 kafka 消费者 KafkaConsumer consumer = new KafkaConsumer(properties); //4. 设置消费主题 形参是列表 consumer.subscribe(Collections.singletonList("first")); //5. 消费数据 while (true) { // 读取消息 ConsumerRecords consumerRecords = consumer.poll(Duration.ofSeconds(1)); // 输出消息 for (ConsumerRecord consumerRecord : consumerRecords) { System.out.println(consumerRecord.value()); } // 同步提交 offset consumer.commitAsync(); } } 5.5.4 指定Offset消费auto.offset.reset = earliest | latest | none 默认是 latest。 当 Kafka 中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量时（例如该数据已被删除），该怎么办？ （1）earliest：自动将偏移量重置为最早的偏移量，--from-beginning。 （2）latest（默认值）：自动将偏移量重置为最新偏移量。 （3）none：如果未找到消费者组的先前偏移量，则向消费者抛出异常。
（4）任意指定 offset 位移开始消费 public static void main(String[] args) { // 0 配置信息 Properties properties = new Properties(); // 连接 properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 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 kafkaConsumer = new KafkaConsumer(properties); // 2 订阅一个主题 ArrayList topics = new ArrayList(); topics.add("first"); kafkaConsumer.subscribe(topics); Set 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 consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1)); for (ConsumerRecord consumerRecord : consumerRecords) { System.out.println(consumerRecord); } } } 5.5.5 指定时间消费需求：在生产环境中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。例如要求按照时间消费前一天的数据，怎么处理？ 操作步骤： public static void main(String[] args) { // 0 配置信息 Properties properties = new Properties(); // 连接 properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVER_CONFIG); // 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 kafkaConsumer = new KafkaConsumer(properties); // 2 订阅一个主题 ArrayList topics = new ArrayList(); topics.add("first"); kafkaConsumer.subscribe(topics); Set assignment = new HashSet(); while (assignment.size() == 0) { kafkaConsumer.poll(Duration.ofSeconds(1)); // 获取消费者分区分配信息（有了分区分配信息才能开始消费） assignment = kafkaConsumer.assignment(); } HashMap timestampToSearch = new HashMap(); // 封装集合存储，每个分区对应一天前的数据 for (TopicPartition topicPartition : assignment) { timestampToSearch.put(topicPartition, System.currentTimeMillis() - 24 * 3600 * 1000); } // 获取从 1 天前开始消费的每个分区的 offset Map 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 consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1)); for (ConsumerRecord consumerRecord : consumerRecords) { System.out.println(consumerRecord); } } } 5.5.6 漏消费和重复消费重复消费：已经消费了数据，但是 offset 没提交。 漏消费：先提交 offset 后消费，有可能会造成数据的漏消费。 （1）场景1：重复消费。自动提交offset引起。
（2）场景1：漏消费。设置offset为手动提交，当offset被提交时，数据还在内存中未落盘，此时刚好消费者线程被kill掉，那么offset已经提交，但是数据未处理，导致这部分内存中的数据丢失。
如果想完成Consumer端的精准一次性消费，那么需要Kafka消费端将消费过程和提交offset过程做原子绑定。此时我们需要将Kafka的offset保存到支持事务的自定义介质（比 如MySQL）。这部分知识会在后续项目部分涉及。
1）如果是Kafka消费能力不足，则可以考虑增加Topic的分区数，并且同时提升消费组的消费者数量，消费者数 = 分区数。（两者缺一不可）
2）如果是下游的数据处理不及时：提高每批次拉取的数量。批次拉取数据过少（拉取数据/处理时间 < 生产速度），使处理的数据小于生产的数据，也会造成数据积压。
参数名称 描述 fetch.max.bytes 默认 Default: 52428800（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受 message.max.bytes （brokerconfig）or max.message.bytes （topic config）影响。 max.poll.records 一次 poll 拉取数据返回消息的最大条数，默认是 500 条 六、Kafka-Eagle 监控Kafka-Eagle 框架可以监控 Kafka 集群的整体运行情况，在生产环境中经常使用。 6.1 Mysql环境准备Kafka-Eagle 的安装依赖于 MySQL，MySQL 主要用来存储可视化展示的数据。如果集群中之前安装过 MySQL 可以跨过该步。 安装参考：有道云笔记 6.2 Kafka环境准备1）关闭 Kafka 集群 [atguigu@hadoop102 kafka]$ kf.sh stop2）修改/opt/module/kafka/bin/kafka-server-start.sh 命令中 [atguigu@hadoop102 kafka]$ vim bin/kafka-server-start.sh修改如下参数值： if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G" fi为 if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then export KAFKA_HEAP_OPTS="-server -Xms2G -Xmx2G -XX:PermSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:ParallelGCThreads=8 -XX:ConcGCThreads=5 -XX:InitiatingHeapOccupancyPercent=70" export JMX_PORT="9999" #export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G" fi注意：修改之后在启动 Kafka 之前要分发之其他节点 6.3 Kafka-Eagle安装1）官网：EFAK 2）解压到本地 mbp:kafka xiang$ tar -zxvf kafka-eagle-bin3.0.1.tar.gz3）进入刚才解压的目录,将 efak-web-3.0.1-bin.tar.gz 解压,并重命名efak mbp:kafka-eagle xiang$ ll drwxr-xr-x@ 8 xiang staff 256 6 17 16:31 efak -rw-rw-r--@ 1 xiang staff 89947836 9 6 2022 efak-web-3.0.1-bin.tar.gz4）修改配置文件conf/system-config.properties ###################################### efak.zk.cluster.alias=cluster1 cluster1.zk.list=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka # offset 保存在 kafka cluster1.efak.offset.storage=kafka ###################################### # 配置 mysql 连接 efak.driver=com.mysql.jdbc.Driver efak.url=jdbc:mysql://hadoop102:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull efak.username=root efak.password=000000 ###################################### # kafka mysql jdbc driver address ###################################### #efak.driver=com.mysql.cj.jdbc.Driver #efak.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=U TF-8&zeroDateTimeBehavior=convertToNull #efak.username=root #efak.password=1234565)添加环境变量 vi /etc/profile # kafkaEFAK export KE_HOME=/Users/xiang/software/kafka/kafka-eagle/efak export PATH=$PATH:$KE_HOME/bin注意：source /etc/profile 6）启动 （1）注意：启动之前需要先启动 ZK 以及 KAFKA。 （2）启动 efak mbp:efak xiang$ 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 ***************************************************************** * ke.sh [start|status|stop|restart|stats] * https://www.kafka-eagle.org/ *****************************************************************
说明：如果停止 efak，执行命令。 [atguigu@hadoop102 efak]$ bin/ke.sh stop 6.4 Kafka-Eagle页面操作登录页面查看监控数据:http://127.0.0.1:8048/
左图为Kafka现有架构，元数据在zookeeper中，运行时动态选举controller，由controller进行Kafka集群管理。 右图为kraft模式架构（实验性），不再依赖zookeeper集群，而是用三台controller节点代替zookeeper，元数据保存在controller中，由controller直接进行Kafka集群管理。 这样做的好处有以下几个： Kafka不再依赖外部框架，而是能够独立运行；controller管理集群时，不再需要从zookeeper中先读取数据，集群性能上升；由于不依赖zookeeper，集群扩展时不再受到zookeeper读写能力限制；controller不再动态选举，而是由配置文件规定。这样我们可以有针对性的加强controller节点的配置，而不是像以前一样对随机controller节点的高负载束手无策。 7.2 Kafka-Kraft 集群部署1）再次解压一份 kafka 安装包 [atguigu@hadoop102 software]$ tar -zxvf kafka_2.12-3.0.0.tgz -C/opt/module/2）重命名为 kafka2 [atguigu@hadoop102 module]$ mv kafka_2.12-3.0.0/ kafka23）在 hadoop102 上修改/opt/module/kafka2/config/kraft/server.properties 配置文件 [atguigu@hadoop102 kraft]$ vim server.properties #kafka 的角色（controller 相当于主机、broker 节点相当于从机，主机类似 zk 功能） process.roles=broker,controller #节点 ID node.id=2 #controller 服务协议别名 controller.listener.names=CONTROLLER #全 Controller 列表 controller.quorum.voters=2@hadoop102:9093,3@hadoop103:9093,4@hadoop104:9093 #不同服务器绑定的端口 listeners=PLAINTEXT://:9092,CONTROLLER://:9093 #broker 服务协议别名 inter.broker.listener.name=PLAINTEXT #broker 对外暴露的地址 advertised.Listeners=PLAINTEXT://hadoop102:9092 #协议别名到安全协议的映射 listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL #kafka 数据存储目录 log.dirs=/opt/module/kafka2/data4）分发 kafka2 [atguigu@hadoop102 module]$ xsync kafka2/ 在hadoop103和hadoop104上需要对node.id相应改变，值需要和controller.quorum.voters对应。在hadoop103和hadoop104上需要根据各自的主机名称，修改相应的advertised.Listeners地址。5）初始化集群数据目录 （1）首先生成存储目录唯一 ID。 [atguigu@hadoop102 kafka2]$ bin/kafka-storage.sh random-uuidJ7s9e8PPTKOO47PxzI39VA（2）用该 ID 格式化 kafka 存储目录（三台节点）。 [atguigu@hadoop102 kafka2]$ bin/kafka-storage.sh format -t J7s9e8PPTKOO47PxzI39VA -c /opt/module/kafka2/config/kraft/server.properties [atguigu@hadoop103 kafka2]$ bin/kafka-storage.sh format -t J7s9e8PPTKOO47PxzI39VA -c /opt/module/kafka2/config/kraft/server.properties [atguigu@hadoop104 kafka2]$ bin/kafka-storage.sh format -t J7s9e8PPTKOO47PxzI39VA -c /opt/module/kafka2/config/kraft/server.properties6）启动 kafka 集群 [atguigu@hadoop102 kafka2]$ bin/kafka-server-start.sh -daemon config/kraft/server.properties [atguigu@hadoop103 kafka2]$ bin/kafka-server-start.sh -daemon config/kraft/server.properties [atguigu@hadoop104 kafka2]$ bin/kafka-server-start.sh -daemon config/kraft/server.properties7）停止 kafka 集群 [atguigu@hadoop102 kafka2]$ bin/kafka-server-stop.sh [atguigu@hadoop103 kafka2]$ bin/kafka-server-stop.sh [atguigu@hadoop104 kafka2]$ bin/kafka-server-stop.sh 7.3 Kafka-Kraft 集群启动停止脚本1）在/home/atguigu/bin 目录下创建文件 kf2.sh 脚本文件 [atguigu@hadoop102 bin]$ vim kf2.sh脚本如下： #! /bin/bash case $1 in "start"){ for i in hadoop102 hadoop103 hadoop104 do echo " --------启动 $i Kafka2-------" ssh $i "/opt/module/kafka2/bin/kafka-server-start.sh -daemon /opt/module/kafka2/config/kraft/server.properties" done };; "stop"){ for i in hadoop102 hadoop103 hadoop104 do echo " --------停止 $i Kafka2-------" ssh $i "/opt/module/kafka2/bin/kafka-server-stop.sh " done };; esac2）添加执行权限 [atguigu@hadoop102 bin]$ chmod +x kf2.sh3）启动集群命令 [atguigu@hadoop102 ~]$ kf2.sh start4）停止集群命令 [atguigu@hadoop102 ~]$ kf2.sh stop 八、集成FlumeFlume 是一个在大数据开发中非常常用的组件。可以用于 Kafka 的生产者，也可以用于Flume 的消费者。
（1）启动 kafka 集群 [atguigu@hadoop102 ~]$ zk.sh start [atguigu@hadoop102 ~]$ kf.sh start（2）启动 kafka 消费者 [atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first（3）Flume 安装 （4）配置 Flume 在 hadoop102 节点的 Flume 的 job 目录下创建 file_to_kafka.conf [atguigu@hadoop102 flume]$ mkdir jobs [atguigu@hadoop102 flume]$ 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（5）启动 Flume [atguigu@hadoop102 flume]$ bin/flume-ng agent -c conf/ -n a1 -fjobs/file_to_kafka.conf &（6）向/opt/module/applog/app.log 里追加数据，查看 kafka 消费者消费情况 [atguigu@hadoop102 module]$ mkdir applog [atguigu@hadoop102 applog]$ echo hello >>/opt/module/applog/app.log（7）观察 kafka 消费者，能够看到消费的 hello 数据 8.2 Flume消费者
（1）配置 Flume 在 hadoop102 节点的 Flume 的/opt/module/flume/jobs 目录下创建 kafka_to_file.conf [atguigu@hadoop102 jobs]$ 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（2）启动 Flume [atguigu@hadoop102 flume]$ bin/flume-ng agent -c conf/ -n a1 -fjobs/kafka_to_file.conf -Dflume.root.logger=INFO,console（3）启动 kafka 生产者 [atguigu@hadoop103 kafka]$ bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first并输入数据，例如：hello world （4）观察控制台输出的日志 九、集成FlinkFlink 是一个在大数据开发中非常常用的组件。可以用于 Kafka 的生产者，也可以用于Flink 的消费者。
1）Flink 环境准备 （1）创建一个 maven 项目 flink-kafka （2）添加配置文件 org.apache.flink flink-java 1.13.0 org.apache.flink flink-streaming-java_2.12 1.13.0 org.apache.flink flink-clients_2.12 1.13.0 org.apache.flink flink-connector-kafka_2.12 1.13.0（3）将 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%n log4j.appender.R=org.apache.log4j.RollingFileAppender log4j.appender.R.File=../log/agent.log log4j.appender.R.MaxFileSize=1024KB log4j.appender.R.MaxBackupIndex=1 log4j.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（4）在 java 文件夹下创建包名为 com.flink 9.1 Flink生产者（1）在 com.flink 包下创建 java 类：FlinkKafkaProducer1 package com.flink; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import java.util.ArrayList; import java.util.Properties; public class FlinkKafkaProducer1 { public static void main(String[] args) throws Exception { // 0 初始化 flink 环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(3); // 1 读取集合中数据 ArrayList wordsList = new ArrayList(); wordsList.add("hello"); wordsList.add("world"); DataStream stream = env.fromCollection(wordsList); // 2 kafka 生产者配置信息 Properties properties = new Properties(); properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // 3 创建 kafka 生产者 FlinkKafkaProducer kafkaProducer = new FlinkKafkaProducer("first", new SimpleStringSchema(), properties); // 4 生产者和 flink 流关联 stream.addSink(kafkaProducer); // 5 执行 env.execute(); } }（2）启动 Kafka 消费者 （3）执行 FlinkKafkaProducer1 程序，观察 kafka 消费者控制台情况
（1）在 com.flink 包下创建 java 类：FlinkKafkaConsumer1 package com.flink; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import java.util.Properties; public class FlinkKafkaConsumer1 { 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, "localhost:9092"); // 2 创建 kafka 消费者 FlinkKafkaConsumer kafkaConsumer = new FlinkKafkaConsumer("first", new SimpleStringSchema(), properties); // 3 消费者和 flink 流关联 env.addSource(kafkaConsumer).print(); // 4 执行 env.execute(); } }（2）启动 FlinkKafkaConsumer1 消费者 （3）启动 kafka 生产者 （4）观察 IDEA 控制台数据打印
SpringBoot 是一个在 JavaEE 开发中非常常用的组件。可以用于 Kafka 的生产者，也可以用于 SpringBoot 的消费者。
SpringBoot 环境准备 初始化时勾选： lombokspring webkafka org.springframework.boot spring-boot-starter-web org.springframework.kafka spring-kafka org.projectlombok lombok true org.springframework.boot spring-boot-starter-test test org.springframework.kafka spring-kafka-test test 10.1 SpringBoot生产者（1）修改 SpringBoot 核心配置文件 application.propeties, 添加生产者相关信息 # 应用名称 spring.application.name=atguigu_springboot_kafka # 指定 kafka 的地址 spring.kafka.bootstrapservers=localhost:9092 #指定 key 和 value 的序列化器 spring.kafka.producer.keyserializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.valueserializer=org.apache.kafka.common.serialization.StringSerializer（2）创建 controller 从浏览器接收数据, 并写入指定的 topic @RestController public class TestController { // Kafka 模板用来向 kafka 发送数据 @Resource KafkaTemplate kafka; @GetMapping("/testSend") public String data(String msg) { kafka.send("first", msg); return "ok"; } }（3）在浏览器中给调用接口发送数据 http://localhost:8080/testSend?msg=hello （4）观察消费者
（1） SpringBoot 核心配置文件 application.propeties增加消费者配置项 # 指定 kafka 的地址 spring.kafka.bootstrapservers=localhost:9092 # 指定 key 和 value 的反序列化器 spring.kafka.consumer.keydeserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.valuedeserializer=org.apache.kafka.common.serialization.StringDeserializer #指定消费者组的 group_id spring.kafka.consumer.group-id=xiang（2）创建类消费 Kafka 中指定 topic 的数据 public class Consumer { // 指定要监听的 topic @KafkaListener(topics = "first") public void consumeTopic(String msg) { // 参数: 收到的 value System.out.println("收到的信息: " + msg); } }（3）向 first 主题发送数据 （4）观察消费记录
Spark 是一个在大数据开发中非常常用的组件。可以用于 Kafka 的生产者，也可以用于Spark 的消费者。
1）Scala 环境安装 2）Spark 环境准备 （1）创建一个 maven 项目 spark-kafka （2）在项目 spark-kafka 上点击右键，Add Framework Support=》勾选 scala （3）在 main 下创建 scala 文件夹，并右键 Mark Directory as Sources Root=>在 scala 下创建包名为 com.atguigu.spark （4）添加配置文件 org.apache.spark spark-streaming-kafka-0-10_2.12 3.0.0（5）将 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%n log4j.appender.R=org.apache.log4j.RollingFileAppender log4j.appender.R.File=../log/agent.log log4j.appender.R.MaxFileSize=1024KB log4j.appender.R.MaxBackupIndex=1 log4j.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 11.1 Spark 生产者（1）在 com.atguigu.spark 包下创建 scala Object：SparkKafkaProducer package com.atguigu.spark import java.util.Properties import org.apache.kafka.clients.producer.{KafkaProducer,ProducerRecord} 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 "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 数据创建 DStream val 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.计算 WordCount valueDStream.print() //7.开启任务 ssc.start() ssc.awaitTermination() } }（3）启动 SparkKafkaConsumer 消费者 （4）启动 kafka 生产者 （5）观察 IDEA 控制台数据打印 |
【本文地址】
今日新闻 |
推荐新闻 |