标签：oti -- kafka atguigu hadoop102 properties Kafka

尚硅谷大数据技术之Kafka
（作者：尚硅谷研究院）

版本：V4.0

第1章 Kafka概述
1.1 定义

1.2 消息队列
目前企业中比较常见的消息队列产品主要有Kafka、ActiveMQ、RabbitMQ、RocketMQ等。
在大数据场景主要采用Kafka作为消息队列。在JavaEE开发中主要采用ActiveMQ、RabbitMQ、RocketMQ。可以关注尚硅谷教育公众号回复java，免费获取相关资料。
1.2.1 传统消息队列的应用场景
传统的消息队列的主要应用场景包括：缓存/消峰、解耦和异步通信。

1.2.2 消息队列的两种模式

1.3 Kafka基础架构
（1）Producer：消息生产者，就是向Kafka broker发消息的客户端。
（2）Consumer：消息消费者，向Kafka broker取消息的客户端。
（3）Consumer Group（CG）：消费者组，由多个consumer组成。消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费；消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
（4）Broker：一台Kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。
（5）Topic：可以理解为一个队列，生产者和消费者面向的都是一个topic。
（6）Partition：为了实现扩展性，一个非常大的topic可以分布到多个broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。
（7）Replica：副本。一个topic的每个分区都有若干个副本，一个Leader和若干个Follower。
（8）Leader：每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是Leader。
（9）Follower：每个分区多个副本中的“从”，实时从Leader中同步数据，保持和Leader数据的同步。Leader发生故障时，某个Follower会成为新的Leader。
第2章 Kafka快速入门
2.1 安装部署
2.1.1 集群规划
hadoop102 hadoop103 hadoop104
zk zk zk
kafka kafka kafka
2.1.2 集群部署
0）官方下载地址：http://kafka.apache.org/downloads.html
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/ kafka
3）进入到/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/kafka
4）分发安装包
[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=2
6）配置环境变量
（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/profile
7）启动集群
（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.properties
注意：配置文件的路径要能够到server.properties。
8）关闭集群
[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
};;
esac
2）添加执行权限
[atguigu@hadoop102 bin]$ chmod +x kf.sh
3）启动集群命令
[atguigu@hadoop102 ~]$ kf.sh start
4）停止集群命令
[atguigu@hadoop102 ~]$ kf.sh stop
注意：停止Kafka集群时，一定要等Kafka所有节点进程全部停止后再停止Zookeeper集群。因为Zookeeper集群当中记录着Kafka集群相关信息，Zookeeper集群一旦先停止，Kafka集群就没有办法再获取停止进程的信息，只能手动杀死Kafka进程了。
2.2 Kafka命令行操作

2.2.1 主题命令行操作
1）查看操作主题命令参数
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh

参数描述
--bootstrap-server <String: server toconnect to> 连接的Kafka Broker主机名称和端口号。
--topic <String: topic> 操作的topic名称。
--create 创建主题。
--delete 删除主题。
--alter 修改主题。
--list 查看所有主题。
--describe 查看主题详细描述。
--partitions <Integer: # of partitions> 设置分区数。
--replication-factor<Integer: replication factor> 设置分区副本。
--config <String: name=value> 更新系统默认的配置。
2）查看当前服务器中的所有topic
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --list
3）创建first topic
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 1 --replication-factor 3 --topic first
选项说明：
--topic 定义topic名
--replication-factor 定义副本数
--partitions 定义分区数
4）查看first主题的详情
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first
5）修改分区数（注意：分区数只能增加，不能减少）
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 3
6）再次查看first主题的详情
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first
7）删除topic（学生自己演示）
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --delete --topic first
2.2.2 生产者命令行操作
1）查看操作生产者命令参数
[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh

参数描述
--bootstrap-server <String: server toconnect to> 连接的Kafka Broker主机名称和端口号。
--topic <String: topic> 操作的topic名称。
2）发送消息
[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first

hello world
atguigu atguigu
2.2.3 消费者命令行操作
1）查看操作消费者命令参数
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh

参数描述
--bootstrap-server <String: server toconnect to> 连接的Kafka Broker主机名称和端口号。
--topic <String: topic> 操作的topic名称。
--from-beginning 从头开始消费。
--group <String: consumer group id> 指定消费者组名称。
2）消费消息
（1）消费first主题中的数据。
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
（2）把主题中所有的数据都读取出来（包括历史数据）。
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --from-beginning --topic first
第3章 Kafka生产者
3.1 生产者消息发送流程
3.1.1 发送原理
在消息发送的过程中，涉及到了两个线程——main线程和Sender线程。在main线程中创建了一个双端队列RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka Broker。

3.1.2 生产者重要参数列表
参数名称描述
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代码
package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class CustomProducer {

public static void main(String[] args) throws InterruptedException {

    // 1. 创建kafka生产者的配置对象
    Properties properties = new Properties();

    // 2. 给kafka配置对象添加配置信息：bootstrap.servers
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
    
    // key,value序列化（必须）：key.serializer，value.serializer
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

    // 3. 创建kafka生产者对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

    // 4. 调用send方法,发送消息
    for (int i = 0; i < 5; i++) {

        kafkaProducer.send(new ProducerRecord<>("first","atguigu " + i));

    }

    // 5. 关闭资源
    kafkaProducer.close();
}

}
测试：
①在hadoop102上开启Kafka消费者。
[atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
②在IDEA中执行代码，观察hadoop102控制台中是否接收到消息。
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

atguigu 0
atguigu 1
atguigu 2
atguigu 3
atguigu 4
3.2.2 带回调函数的异步发送
回调函数会在producer收到ack时调用，为异步调用，该方法有两个参数，分别是元数据信息（RecordMetadata）和异常信息（Exception），如果Exception为null，说明消息发送成功，如果Exception不为null，说明消息发送失败。

注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。
package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class CustomProducerCallback {

public static void main(String[] args) throws InterruptedException {

    // 1. 创建kafka生产者的配置对象
    Properties properties = new Properties();

    // 2. 给kafka配置对象添加配置信息
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");

    // key,value序列化（必须）：key.serializer，value.serializer
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    // 3. 创建kafka生产者对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

    // 4. 调用send方法,发送消息
    for (int i = 0; i < 5; i++) {

        // 添加回调
        kafkaProducer.send(new ProducerRecord<>("first", "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();
}

atguigu 0
atguigu 1
atguigu 2
atguigu 3
atguigu 4
③在IDEA控制台观察回调信息。
主题：first->分区：0
主题：first->分区：0
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
3.3 同步发送API

只需在异步发送的基础上，再调用一下get()方法即可。
package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class CustomProducerSync {

public static void main(String[] args) throws InterruptedException, ExecutionException {

    // 1. 创建kafka生产者的配置对象
    Properties properties = new Properties();

    // 2. 给kafka配置对象添加配置信息

properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");

    // key,value序列化（必须）：key.serializer，value.serializer
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    // 3. 创建kafka生产者对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

    // 4. 调用send方法,发送消息
    for (int i = 0; i < 10; i++) {

        // 异步发送 默认

// kafkaProducer.send(new ProducerRecord<>("first","kafka" + i));
// 同步发送
kafkaProducer.send(new ProducerRecord<>("first","kafka" + i)).get();

    }

    // 5. 关闭资源
    kafkaProducer.close();
}

atguigu 0
atguigu 1
atguigu 2
atguigu 3
atguigu 4
3.4 生产者分区
3.4.1 分区好处

3.4.2 生产者发送消息的分区策略
1）默认的分区器 DefaultPartitioner
在IDEA中ctrl + n，全局查找DefaultPartitioner。
/**

The default partitioning strategy:
- If a partition is specified in the record, use it
- If no partition is specified but a key is present choose a partition based on a hash of the key
- If no partition or key is present choose the sticky partition that changes when the batch is full.
- See KIP-480 for details about sticky partitioning.
  */
  public class DefaultPartitioner implements Partitioner {
  
  … …
  }
2）案例一
将数据发往指定partition的情况下，例如，将所有数据发往分区1中。
package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class CustomProducerCallbackPartitions {
```
public static void main(String[] args) {

    // 1. 创建kafka生产者的配置对象
    Properties properties = new Properties();

    // 2. 给kafka配置对象添加配置信息
  properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");

    // key,value序列化（必须）：key.serializer，value.serializer
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

    for (int i = 0; i < 5; i++) {
        // 指定数据发送到1号分区，key为空（IDEA中ctrl + p查看参数）
        kafkaProducer.send(new ProducerRecord<>("first", 1,"","atguigu " + i), new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception e) {
                if (e == null){
                    System.out.println("主题：" + metadata.topic() + "->"  + "分区：" + metadata.partition()
                    );
                }else {
                    e.printStackTrace();
                }
            }
        });
    }

    kafkaProducer.close();
}
```
}
测试：
①在hadoop102上开启Kafka消费者。
[atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
②在IDEA中执行代码，观察hadoop102控制台中是否接收到消息。
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

atguigu 0
atguigu 1
atguigu 2
atguigu 3
atguigu 4
③在IDEA控制台观察回调信息。
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
3）案例二
没有指明partition值但有key的情况下，将key的hash值与topic的partition数进行取余得到partition值。
package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class CustomProducerCallback {
```
public static void main(String[] args) {

    Properties properties = new Properties();

    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");

    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

    for (int i = 0; i < 5; i++) {
        // 依次指定key值为a,b,f ，数据key的hash值与3个分区求余，分别发往1、2、0
        kafkaProducer.send(new ProducerRecord<>("first", "a","atguigu " + i), new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception e) {
                if (e == null){
                    System.out.println("主题：" + metadata.topic() + "->" + "分区：" + metadata.partition()
                    );
                }else {
                    e.printStackTrace();
                }
            }
        });
    }

    kafkaProducer.close();
}
```
}
测试：
①key="a"时，在控制台查看结果。
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
②key="b"时，在控制台查看结果。
主题：first->分区：2
主题：first->分区：2
主题：first->分区：2
主题：first->分区：2
主题：first->分区：2
③key="f"时，在控制台查看结果。
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
3.4.3 自定义分区器
如果研发人员可以根据企业需求，自己重新实现分区器。
1）需求
例如我们实现一个分区器实现，发送过来的数据中如果包含atguigu，就发往0号分区，不包含atguigu，就发往1号分区。
2）实现步骤
（1）定义类实现Partitioner接口。
（2）重写partition()方法。
package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

/**
- 1. 实现接口Partitioner
- 1. 实现3个方法:partition,close,configure
- 1. 编写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）使用分区器的方法，在生产者的配置中添加分区器参数。
  package com.atguigu.kafka.producer;
import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class CustomProducerCallbackPartitions {
```
public static void main(String[] args) throws InterruptedException {

    Properties properties = new Properties();

    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");

    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    // 添加自定义分区器
```
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.atguigu.kafka.producer.MyPartitioner");
```
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

    for (int i = 0; i < 5; i++) {
        
        kafkaProducer.send(new ProducerRecord<>("first", "atguigu " + i), new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception e) {
                if (e == null){
                    System.out.println("主题：" + metadata.topic() + "->" + "分区：" + metadata.partition()
                    );
                }else {
                    e.printStackTrace();
                }
            }
        });
    }

    kafkaProducer.close();
}
```
}
（4）测试
①在hadoop102上开启Kafka消费者。
[atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
②在IDEA控制台观察回调信息。
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
3.5 生产经验——生产者如何提高吞吐量

package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class CustomProducerParameters {
```
public static void main(String[] args) throws InterruptedException {

    // 1. 创建kafka生产者的配置对象
    Properties properties = new Properties();

    // 2. 给kafka配置对象添加配置信息：bootstrap.servers
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
    
    // key,value序列化（必须）：key.serializer，value.serializer
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

    // batch.size：批次大小，默认16K
    properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);

    // linger.ms：等待时间，默认0
    properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);

    // RecordAccumulator：缓冲区大小，默认32M：buffer.memory
    properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

    // compression.type：压缩，默认none，可配置值gzip、snappy、lz4和zstd
```
properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");
```
    // 3. 创建kafka生产者对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

    // 4. 调用send方法,发送消息
    for (int i = 0; i < 5; i++) {

        kafkaProducer.send(new ProducerRecord<>("first","atguigu " + i));

    }

    // 5. 关闭资源
    kafkaProducer.close();
}
```
}
测试：
①在hadoop102上开启Kafka消费者。
[atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
②在IDEA中执行代码，观察hadoop102控制台中是否接收到消息。
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

atguigu 0
atguigu 1
atguigu 2
atguigu 3
atguigu 4
3.6 生产经验——数据可靠性
0）回顾发送流程

1）ack应答原理

2）代码配置
package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class CustomProducerAck {
```
public static void main(String[] args) throws InterruptedException {

    // 1. 创建kafka生产者的配置对象
    Properties properties = new Properties();

    // 2. 给kafka配置对象添加配置信息：bootstrap.servers
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
    
    // key,value序列化（必须）：key.serializer，value.serializer
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    // 设置acks
    properties.put(ProducerConfig.ACKS_CONFIG, "all");

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

    // 3. 创建kafka生产者对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

    // 4. 调用send方法,发送消息
    for (int i = 0; i < 5; i++) {

        kafkaProducer.send(new ProducerRecord<>("first","atguigu " + i));

    }

    // 5. 关闭资源
    kafkaProducer.close();
}
```
}
3.7 生产经验——数据去重
3.7.1 数据传递语义

3.7.2 幂等性
1）幂等性原理

2）如何使用幂等性
开启参数enable.idempotence 默认为true，false关闭。
3.7.3 生产者事务
1）Kafka事务介绍
0.11版本的Kafka同时引入了事务的特性，为了实现跨分区跨会话的事务，需要引入一个全局唯一的Transaction ID，并将Producer获得的PID和Transaction ID绑定。这样当Producer重启后就可以通过正在进行的Transaction ID获得原来的PID。
为了管理Transaction，Kafka引入了一个新的组件Transaction Coordinator。Producer就是通过和Transaction Coordinator交互获得Transaction ID对应的任务状态。Transaction Coordinator还负责将事务所有写入Kafka的一个内部Topic，这样即使整个服务重启，由于事务状态得到保存，进行中的事务状态可以得到恢复，从而继续进行。
注意：提前开启幂等性!!!
3.8 生产经验——数据有序

3.9 生产经验——数据乱序

第4章 Kafka Broker
4.1 Kafka Broker工作流程
4.1.1 Zookeeper存储的Kafka信息
（1）启动Zookeeper客户端。
[atguigu@hadoop102 zookeeper-3.5.7]$ bin/zkCli.sh
（2）通过ls命令可以查看kafka相关信息。
[zk: localhost:2181(CONNECTED) 2] ls /kafka

4.1.2 Kafka Broker总体工作流程

1）模拟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 Kafka 副本
4.2.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.2.2 Leader选举流程
Kafka集群中有一个broker的Controller会被选举为Controller Leader，负责管理集群broker的上下线，所有topic的分区副本分配和Leader选举等工作。
Controller的信息同步工作是依赖于Zookeeper的。

假设有hadoop102、hadoop103、hadoop104、hadoop105等4台服务器。
（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.2.3 Leader和Follower故障处理细节

4.3 文件存储
4.3.1 文件存储机制
1）Topic数据的存储机制

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: true
3）index文件和log文件详解

说明：日志存储参数配置
参数描述
log.segment.bytes Kafka中log日志是分成一块块存储的，此配置是指log日志划分成块的大小，默认值1G。
log.index.interval.bytes 默认4kb，kafka里面每当写入了4kb大小的日志（.log），然后就往index文件里面记录一个索引。稀疏索引。
4.3.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 所有数据启用删除策略
（1）基于时间：默认打开。以segment中所有记录中的最大时间戳作为该文件时间戳。
（2）基于大小：默认关闭。超过设置的所有日志总大小，删除最早的segment。
log.retention.bytes，默认等于-1，表示无穷大。
思考：如果一个segment中有一部分数据过期，一部分没有过期，怎么处理？

2）compact日志压缩

4.4 高效读写数据
1）Kafka本身是分布式集群，可以采用分区技术，并行度高
2）读数据采用稀疏索引，可以快速定位要消费的数据
3）顺序写磁盘
Kafka的producer生产数据，要写入到log文件中，写的过程是一直追加到文件末端，为顺序写。官网有数据表明，同样的磁盘，顺序写能到600M/s，而随机写只有100K/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。

4）页缓存 + 零拷贝技术

参数描述
log.flush.interval.messages 强制页缓存刷写到磁盘的条数，默认是long的最大值，9223372036854775807。一般不建议修改，交给系统自己管理。
log.flush.interval.ms 每隔多久，刷数据到磁盘，默认是null。一般不建议修改，交给系统自己管理。
第5章 Kafka消费者
5.1 Kafka消费方式

5.2 Kafka消费者工作流程
5.2.1 消费者总体工作流程

5.2.2 消费者组原理

5.2.3 消费者重要参数
参数名称描述
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条。
5.3 消费者API
5.3.1 独立消费者案例（订阅主题）
1）需求：
创建一个独立消费者，消费first主题中数据。

注意：在消费者API代码中必须配置消费者组id。命令行启动消费者不填写消费者组id会被自动填写随机的消费者组id。
2）实现步骤
（1）创建包名：com.atguigu.kafka.consumer
（2）编写代码
package com.atguigu.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

public class CustomConsumer {
```
public static void main(String[] args) {

    // 1.创建消费者的配置对象
    Properties properties = new Properties();

    // 2.给消费者配置对象添加参数
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");

    // 配置序列化 必须
```
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
```
    // 配置消费者组（组名任意起名） 必须
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");

    // 创建消费者对象
    KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);

    // 注册要消费的主题（可以消费多个主题）
    ArrayList<String> topics = new ArrayList<>();
    topics.add("first");
    kafkaConsumer.subscribe(topics);

    // 拉取数据打印
    while (true) {
        // 设置1s中消费一批数据
        ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

        // 打印消费到的数据
        for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
            System.out.println(consumerRecord);
        }
    }
}
```
}
3）测试
（1）在IDEA中执行消费者程序。
（2）在Kafka集群控制台，创建Kafka生产者，并输入数据。
[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first

hello
（3）在IDEA控制台观察接收到的数据。
ConsumerRecord(topic = first, partition = 1, leaderEpoch = 3, offset = 0, CreateTime = 1629160841112, serialized key size = -1, serialized value size = 5, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello)
5.3.2 消费者组案例
1）需求：测试同一个主题的分区数据，只能由一个消费者组中的一个消费。

2）案例实操
（1）复制一份基础消费者的代码，在IDEA中同时启动，即可启动同一个消费者组中的两个消费者。
package com.atguigu.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

public class CustomConsumer1 {
```
public static void main(String[] args) {

    // 1.创建消费者的配置对象
    Properties properties = new Properties();

    // 2.给消费者配置对象添加参数
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");

    // 配置序列化 必须
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    // 配置消费者组 必须
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");

    // 创建消费者对象
    KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);

    // 注册主题
    ArrayList<String> topics = new ArrayList<>();
    topics.add("first");
    kafkaConsumer.subscribe(topics);

    // 拉取数据打印
    while (true) {
        // 设置1s中消费一批数据
        ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

        // 打印消费到的数据
        for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
            System.out.println(consumerRecord);
        }
    }
}
```
}
（2）启动代码中的生产者发送消息，在IDEA控制台即可看到两个消费者在消费不同分区的数据（如果只发生到一个分区，可以在发送时增加延迟代码Thread.sleep(2);）。
ConsumerRecord(topic = first, partition = 0, leaderEpoch = 2, offset = 3, CreateTime = 1629169606820, serialized key size = -1, serialized value size = 8, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello1)

ConsumerRecord(topic = first, partition = 1, leaderEpoch = 3, offset = 2, CreateTime = 1629169609524, serialized key size = -1, serialized value size = 6, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello2)

ConsumerRecord(topic = first, partition = 2, leaderEpoch = 3, offset = 21, CreateTime = 1629169611884, serialized key size = -1, serialized value size = 6, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello3)
（3）重新发送到一个全新的主题中，由于默认创建的主题分区数为1，可以看到只能有一个消费者消费到数据。

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

参数名称描述
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分区策略原理

2）Range分区分配策略案例
（1）修改主题first为7个分区。
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 7
注意：分区数可以增加，但是不能减少。
（2）复制CustomConsumer类，创建CustomConsumer2。这样可以由三个消费者CustomConsumer、CustomConsumer1、CustomConsumer2组成消费者组，组名都为“test”，同时启动3个消费者。

（3）启动CustomProducer生产者，发送500条消息，随机发送到不同的分区。
package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class CustomProducer {
```
public static void main(String[] args) throws InterruptedException {

    Properties properties = new Properties();

    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");

    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

    for (int i = 0; i < 7; i++) {
        kafkaProducer.send(new ProducerRecord<>("first", i, "test", "atguigu"));
    }

    kafkaProducer.close();
}
```
}
说明：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分区策略原理

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的默认维护位置

__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-server hadoop102:9092 --create --topic atguigu --partitions 2 --replication-factor 2
（3）启动生产者往atguigu生产数据。
[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --topic atguigu --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

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

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class CustomConsumerAutoOffset {
```
public static void main(String[] args) {

    // 1. 创建kafka消费者配置类
    Properties properties = new Properties();

    // 2. 添加配置参数
    // 添加连接
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
   
    // 配置序列化 必须
```
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
```
    // 配置消费者组
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");

    // 是否自动提交offset
    properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
    // 提交offset的时间周期1000ms，默认5s
```
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
```
    //3. 创建kafka消费者
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

    //4. 设置消费主题  形参是列表
    consumer.subscribe(Arrays.asList("first"));

    //5. 消费数据
    while (true){
        // 读取消息
        ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));

        // 输出消息
        for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
            System.out.println(consumerRecord.value());
        }
    }
}
```
}
5.5.3 手动提交offset

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

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class CustomConsumerByHandSync {
```
public static void main(String[] args) {
    // 1. 创建kafka消费者配置类
    Properties properties = new Properties();
    // 2. 添加配置参数
    // 添加连接
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");

    // 配置序列化 必须
```
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
```
    // 配置消费者组
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");

    // 是否自动提交offset
    properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

    //3. 创建kafka消费者
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

    //4. 设置消费主题  形参是列表
    consumer.subscribe(Arrays.asList("first"));

    //5. 消费数据
    while (true){

        // 读取消息
        ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));

        // 输出消息
        for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
            System.out.println(consumerRecord.value());
        } 

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

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.Arrays;
import java.util.Map;
import java.util.Properties;

public class CustomConsumerByHandAsync {
```
public static void main(String[] args) {

    // 1. 创建kafka消费者配置类
    Properties properties = new Properties();

    // 2. 添加配置参数
    // 添加连接
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");

    // 配置序列化 必须
```
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
```
    // 配置消费者组
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");

    // 是否自动提交offset
    properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

    //3. 创建Kafka消费者
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

    //4. 设置消费主题  形参是列表
    consumer.subscribe(Arrays.asList("first"));

    //5. 消费数据
    while (true){

        // 读取消息
        ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));

        // 输出消息
        for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
            System.out.println(consumerRecord.value());
        }

        // 异步提交offset
        consumer.commitAsync();
    }
}
```
}
5.5.4 指定Offset消费
auto.offset.reset = earliest | latest | none 默认是latest。
当Kafka中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量时（例如该数据已被删除），该怎么办？
（1）earliest：自动将偏移量重置为最早的偏移量，--from-beginning。
（2）latest（默认值）：自动将偏移量重置为最新偏移量。
（3）none：如果未找到消费者组的先前偏移量，则向消费者抛出异常。

5.5.5 漏消费和重复消费
重复消费：已经消费了数据，但是offset没提交。
漏消费：先提交offset后消费，有可能会造成数据的漏消费。

思考：怎么能做到既不漏消费也不重复消费呢？详看消费者事务。
5.6 生产经验——消费者事务

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

参数名称描述
fetch.max.bytes 默认Default: 52428800（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受message.max.bytes （broker config）or max.message.bytes （topic config）影响。
max.poll.records 一次poll拉取数据返回消息的最大条数，默认是500条
第6章 Kafka-Eagle监控
Kafka-Eagle框架可以监控Kafka集群的整体运行情况，在生产环境中经常使用。
6.1 MySQL环境准备
Kafka-Eagle的安装依赖于MySQL，MySQL主要用来存储可视化展示的数据。如果集群中之前安装过MySQL可以跨过该步。

6.2 Kafka环境准备
1）关闭Kafka集群
[atguigu@hadoop102 kafka]$ kf.sh stop
2）修改/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之前要分发之其他节点。
[atguigu@hadoop102 bin]$ xsync kafka-server-start.sh
6.3 Kafka-Eagle安装
0）官网：https://www.kafka-eagle.org/
1）上传压缩包kafka-eagle-bin-2.0.8.tar.gz到集群/opt/software目录
2）解压到本地
[atguigu@hadoop102 software]$ tar -zxvf kafka-eagle-bin-2.0.8.tar.gz
3）进入刚才解压的目录
[atguigu@hadoop102 kafka-eagle-bin-2.0.8]$ ll
总用量 79164
-rw-rw-r--. 1 atguigu atguigu 81062577 10月 13 00:00 efak-web-2.0.8-bin.tar.gz
4）将efak-web-2.0.8-bin.tar.gz解压至/opt/module
[atguigu@hadoop102 kafka-eagle-bin-2.0.8]$ tar -zxvf efak-web-2.0.8-bin.tar.gz -C /opt/module/
5）修改名称
[atguigu@hadoop102 module]$ mv efak-web-2.0.8/ efak
6）修改配置文件 /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.truststore
cluster1.efak.jmx.truststore.password=ke123456

######################################

kafka offset storage

######################################

offset保存在kafka

cluster1.efak.offset.storage=kafka

######################################

kafka jmx uri

######################################
cluster1.efak.jmx.uri=service:jmx:rmi:///jndi/rmi://%s/jmxrmi

######################################

kafka metrics, 15 days by default

######################################
efak.metrics.charts=true
efak.metrics.retain=15

######################################

kafka sql topic records max

######################################
efak.sql.topic.records.max=5000
efak.sql.topic.preview.records.max=10

######################################

delete kafka topic token

######################################
efak.topic.token=keadmin

######################################

kafka sasl authenticate

######################################
cluster1.efak.sasl.enable=false
cluster1.efak.sasl.protocol=SASL_PLAINTEXT
cluster1.efak.sasl.mechanism=SCRAM-SHA-256
cluster1.efak.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="kafka" password="kafka-eagle";
cluster1.efak.sasl.client.id=
cluster1.efak.blacklist.topics=
cluster1.efak.sasl.cgroup.enable=false
cluster1.efak.sasl.cgroup.topics=
cluster2.efak.sasl.enable=false
cluster2.efak.sasl.protocol=SASL_PLAINTEXT
cluster2.efak.sasl.mechanism=PLAIN
cluster2.efak.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="kafka-eagle";
cluster2.efak.sasl.client.id=
cluster2.efak.blacklist.topics=
cluster2.efak.sasl.cgroup.enable=false
cluster2.efak.sasl.cgroup.topics=

######################################

kafka ssl authenticate

######################################
cluster3.efak.ssl.enable=false
cluster3.efak.ssl.protocol=SSL
cluster3.efak.ssl.truststore.location=
cluster3.efak.ssl.truststore.password=
cluster3.efak.ssl.keystore.location=
cluster3.efak.ssl.keystore.password=
cluster3.efak.ssl.key.password=
cluster3.efak.ssl.endpoint.identification.algorithm=https
cluster3.efak.blacklist.topics=
cluster3.efak.ssl.cgroup.enable=false
cluster3.efak.ssl.cgroup.topics=

######################################

kafka sqlite jdbc driver address

######################################

配置mysql连接

efak.driver=com.mysql.jdbc.Driver
efak.url=jdbc:mysql://hadoop102:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
efak.username=root
efak.password=root

######################################

kafka mysql jdbc driver address

######################################

efak.driver=com.mysql.cj.jdbc.Driver

efak.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull

efak.username=root

efak.password=123456

7）添加环境变量
[atguigu@hadoop102 conf]$ sudo vim /etc/profile.d/my_env.sh

kafkaEFAK

export KE_HOME=/opt/module/efak
export PATH=$PATH:$KE_HOME/bin
注意：source /etc/profile
[atguigu@hadoop102 conf]$ source /etc/profile
8）启动
（1）注意：启动之前需要先启动ZK以及KAFKA。
[atguigu@hadoop102 kafka]$ kf.sh start
（2）启动efak
[atguigu@hadoop102 efak]$ bin/ke.sh start
Version 2.0.8 -- Copyright 2016-2021
- EFAK Service has started success.
- Welcome, Now you can visit 'http://192.168.10.102:8048'
- Account:admin ,Password:123456
- ke.sh [start|status|stop|restart|stats]
- https://www.kafka-eagle.org/
说明：如果停止efak，执行命令。
[atguigu@hadoop102 efak]$ bin/ke.sh stop
6.4 Kafka-Eagle页面操作
1）登录页面查看监控数据
http://192.168.10.102:8048/

第7章 Kafka-Kraft模式
7.1 Kafka-Kraft架构

左图为Kafka现有架构，元数据在zookeeper中，运行时动态选举controller，由controller进行Kafka集群管理。右图为kraft模式架构（实验性），不再依赖zookeeper集群，而是用三台controller节点代替zookeeper，元数据保存在controller中，由controller直接进行Kafka集群管理。
这样做的好处有以下几个：
 Kafka不再依赖外部框架，而是能够独立运行；
 controller管理集群时，不再需要从zookeeper中先读取数据，集群性能上升；
 由于不依赖zookeeper，集群扩展时不再受到zookeeper读写能力限制；
 controller不再动态选举，而是由配置文件规定。这样我们可以有针对性的加强controller节点的配置，而不是像以前一样对随机controller节点的高负载束手无策。
7.2 Kafka-Kraft集群部署
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/ kafka2
3）在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/data
4）分发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-uuid
J7s9e8PPTKOO47PxzI39VA
（2）用该ID格式化kafka存储目录（三台节点）。
[atguigu@hadoop102 kafka2]$ bin/kafka-storage.sh format -t J7s9e8PPTKOO47PxzI39VA -c /opt/module/kafka_2.12-3.0.0/config/kraft/server.properties

[atguigu@hadoop103 kafka2]$ bin/kafka-storage.sh format -t J7s9e8PPTKOO47PxzI39VA -c /opt/module/kafka_2.12-3.0.0/config/kraft/server.properties

[atguigu@hadoop104 kafka2]$ bin/kafka-storage.sh format -t J7s9e8PPTKOO47PxzI39VA -c /opt/module/kafka_2.12-3.0.0/config/kraft/server.properties
6）启动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.properties
7）停止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
};;
esac
2）添加执行权限
[atguigu@hadoop102 bin]$ chmod +x kf2.sh
3）启动集群命令
[atguigu@hadoop102 ~]$ kf2.sh start
4）停止集群命令
[atguigu@hadoop102 ~]$ kf2.sh stop

标签：oti,--,kafka,atguigu,hadoop102,properties,Kafka
From： https://www.cnblogs.com/seron/p/18008971

kafka-oti

broker的全局唯一编号，不能重复，只能是数字。

处理网络请求的线程数量

用来处理磁盘IO的线程数量

发送套接字的缓冲区大小

接收套接字的缓冲区大小

请求套接字的缓冲区大小

kafka运行日志(数据)存放的路径，路径不需要提前创建，kafka自动帮你创建，可以配置多个磁盘路径，路径与路径之间可以用"，"分隔

topic在当前broker上的分区个数

用来恢复和清理data下数据的线程数量

每个topic创建时的副本数，默认时1个副本

segment文件保留的最长时间，超时将被删除

每个segment文件的大小，默认最大1G

检查过期数据的时间，默认5分钟检查一次是否数据过期

配置连接Zookeeper集群地址（在zk根目录下创建/kafka，方便管理）

The id of the broker. This must be set to a unique integer for each broker.

The id of the broker. This must be set to a unique integer for each broker.

KAFKA_HOME

! /bin/bash

multi zookeeper & kafka cluster list

Settings prefixed with 'kafka.eagle.' will be deprecated, use 'efak.' instead

zookeeper enable acl

broker size online list

zk client thread limit

EFAK webui port

kafka jmx acl and ssl authenticate

kafka offset storage

offset保存在kafka

kafka jmx uri

kafka metrics, 15 days by default

kafka sql topic records max

delete kafka topic token

kafka sasl authenticate

kafka ssl authenticate

kafka sqlite jdbc driver address

配置mysql连接

kafka mysql jdbc driver address

efak.driver=com.mysql.cj.jdbc.Driver

efak.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull

efak.username=root

efak.password=123456

kafkaEFAK

kafka的角色（controller相当于主机、broker节点相当于从机，主机类似zk功能）

节点ID

controller服务协议别名

全Controller列表

不同服务器绑定的端口

broker服务协议别名

broker对外暴露的地址

协议别名到安全协议的映射

kafka数据存储目录

! /bin/bash

相关文章

赞助商

阅读排行