第02篇_Kafka

第01章_Kafka快速入门

第一节 Kafka简介

1. 什么是Kafka？

Kafka是一个基于 Scala 语言开发的分布式消息队列（Message Queue），也称作分布式事件流平台（Event Streaming Platform）。

它支持如下两种模式：

• 点对点模式：一对一，消费者主动拉取数据，消息收到后消息清除。
• 发布/订阅模式：一对多，数据生产后，分类推送给所有订阅者。

主要的功能如下：

• 数据存储：持久化队列中的数据，直到被消费完成。
• 业务解耦：解耦生产者和消费者，两者只需通过消息队列进行通信和协作。
• 削峰填谷：缓和突发流量，不会因为突发的超负荷的请求而全线崩溃。
• 异步处理：消息先放在消息队列，在需要的时候或通过其它线程进行处理。
• 速率控制：控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。
• 顺序保证：同一 Partition 内，消息可以保证有序被消费。
• 可扩展性：方便的增加或减少生产者或消费者的数量。
• 可恢复性：未处理的消息会暂存在消息队列，组件崩溃恢复后仍可继续处理，不会影响整个系统。

2. 基础架构

Kafka的基础架构图如下：

• 服务器(Broker)：一台 kafka 服务器就是一个 broker，一个集群由多个 broker 组成。
• 主题(Topic) ：一个 broker 可以容纳多个 topic（主题可以理解为消息队列）。
• 分区(Partition)：一个 topic 可以细分为多个 partition，不同 partition 可以分布在不同 broker 上( num.partitions)。
• 副本(Replication)：一个 partition 可能有多个 replication，以保证分区数据的高可用(default.replication.factor)。
• 偏移量(Offset)：每个 partition 是一个有序的队列，其中的每条消息都会被分配一个 offset。
• 生产者(Producer)：向 broker 发送消息的客户端，可以指定 topic 和 partition 进行发送。
• 消费者(Consumer)：接收 broker 消息的客户端，在同一 partition 的消息可以按序接收。
• 消费者组(Consumer Group)：用于实现广播(消息发送给关注该topic的所有消费者)和单播(只发给某一消费者)的一种手段。

第二节 安装部署

1. 集群部署

​x
1
# 1. 安装JDK
2
https://www.oracle.com/java/technologies/javase/jdk11-archive-downloads.html
3

4
# 2. 下载Kafka
5
http://kafka.apache.org/downloads.html
6

7
# 3. 上传和解压安装包
8
mkdir /usr/local/kafka
9
tar -zxvf kafka_2.12-3.9.0.tgz
10

11
# 4. 创建logs目录
12
mkdir /usr/local/kafka/kafka_2.12-3.9.0/logs
13

14
# 5. 修改Broker配置文件
15
vi config/server.properties
16

17
# 6. 配置环境变量
18
vim /etc/profile
19
export KAFKA_HOME=/usr/local/kafka/kafka_2.12-3.9.0
20
export PATH=$PATH:$KAFKA_HOME/bin
21
source /etc/profile
22

2. 配置 server.properties

33
1
# 【重要】broker编号（不同服务器编号必须不同）
2
broker.id=0
3

4
# 【重要】监听端口
5
listeners=PLAINTEXT://主机名:9092
6

7
# 支持删除主题
8
delete.topic.enable=true
9

10
# 处理网络请求的线程数量
11
num.network.threads=3
12
# 用来处理磁盘 IO 的线程数量
13
num.io.threads=8
14
# 发送套接字的缓冲区大小
15
socket.send.buffer.bytes=102400
16
# 接收套接字的缓冲区大小
17
socket.receive.buffer.bytes=102400
18
# 请求套接字的缓冲区大小
19
socket.request.max.bytes=104857600
20

21
# 【重要】运行日志存放的路径 
22
log.dirs=/usr/local/kafka/kafka_2.12-3.9.0/logs
23

24
# 主题在当前机器上的分区个数
25
num.partitions=1
26
# 用来恢复和清理 data 下数据的线程数量
27
num.recovery.threads.per.data.dir=1
28
# segment 文件保留的最长时间，超时将被删除
29
log.retention.hours=168
30

31
# 【重要】ZK地址（以逗号分隔）
32
zookeeper.connect=127.0.0.1:2181,127.0.0.2:2181,127.0.0.3:2181
33

3. 启停命令

13
1
# 授权
2
cd /usr/local/kafka/kafka_2.12-3.9.0/bin
3
chmod +x *.sh
4

5
# ZK启动（可能需要）
6
./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties
7

8
# Kafka启动
9
./bin/kafka-server-start.sh -daemon ./config/server.properties
10

11
# 关闭
12
./bin/kafka-server-stop.sh stop
13

4. 配置开机自启

新建vim /etc/systemd/system/kafka.service文件，内容如下：

15
1
[Unit]
2
Description=Apache Kafka Server
3
After=network.target
4

5
[Service]
6
Type=simple
7
User=kafka
8
Group=kafka
9
ExecStart=/usr/local/kafka/kafka_2.12-3.9.0/bin/kafka-server-start.sh /usr/local/kafka/kafka_2.12-3.9.0/server.properties
10
ExecStop=/usr/local/kafka/kafka_2.12-3.9.0/bin/kafka-server-stop.sh /usr/local/kafka/kafka_2.12-3.9.0/server.properties
11
Restart=on-failure
12

13
[Install]
14
WantedBy=multi-user.target
15

设置开机自启：

9
1
# 开启服务
2
sudo systemctl start kafka
3

4
# 开机自启
5
sudo systemctl enable kafka
6

7
# 查看服务状态
8
sudo systemctl status kafka
9

5. 常用命令行操作

1) 主题相关操作

25
1
# 查看主题操作文档
2
# --bootstrap-server                  ip/host:port       连接到哪个Broker
3
# --list                                                 查看所有主题   
4
# --describe/create/delete/alter/                        查看/创建/删除/修改主题
5
# --topic                             topic_name         操作的主题
6
# --partitions                        partition_num      设置分区数
7
# --replication-factor                replication_num    设置分区副本数
8
# --config                            name=value         修改系统默认的配置
9
bin/kafka-topics.sh
10

11
# 查看所有主题
12
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --list
13

14
# 查看主题详情
15
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first
16

17
# 创建主题
18
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 1 --replication-factor 3 --topic first
19

20
# 删除主题
21
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --delete --topic first
22

23
# 增加主题分区数(注意：分区数只能增加，不能减少)
24
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 3
25

注意：

1. 在低版本中，查看所有主题需使用 ./kafka-topics.sh --zookeeper <ZOOKEEPER_ADDRESS> --list命令，其它同理。

2) 生产者相关操作

9
1
# 查看生产者操作文档
2
# --bootstrap-server                  ip/host:port       连接到哪个Broker
3
# --topic                             topic_name         操作的主题
4
bin/kafka-console-producer.sh
5

6
# 向指定主题发送消息
7
bin/kafka-console-producer.sh  --bootstrap-server hadoop102:9092 --topic first
8
>hello world
9

3) 消费者相关操作

13
1
# 查看消费者操作文档
2
# --bootstrap-server                  ip/host:port       连接到哪个Broker
3
# --topic                             topic_name         操作的主题
4
# --from-beginning                                       从头开始消费
5
# --group                             consumer_group_id  指定消费者组名称
6
bin/kafka-console-consumer.sh
7

8
# 消费指定主题中的数据
9
bin/kafka-console-consumer.sh  --bootstrap-server hadoop102:9092 --topic first
10

11
# 从头消费指定主题中的数据(包括之前已消费过的历史数据)
12
bin/kafka-console-consumer.sh  --bootstrap-server hadoop102:9092 --from-beginning --topic first
13

4) 副本相关操作

25
1
# 增加副本因子
2
vim increase-replication-factor.json
3
输入：
4
{
5
    "version":1,
6
    "partitions":[
7
        {"topic":"four","partition":0,"replicas":[0,1,2]},
8
        {"topic":"four","partition":1,"replicas":[0,1,2]},
9
        {"topic":"four","partition":2,"replicas":[0,1,2]}]
10
}
11
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute
12

13
# 调整分区副本存储
14
vim increase-replication-factor.json
15
输入：
16
{
17
    "version":1,
18
    "partitions":[{"topic":"three","partition":0,"replicas":[0,1]},
19
        {"topic":"three","partition":1,"replicas":[0,1]},
20
        {"topic":"three","partition":2,"replicas":[1,0]},
21
        {"topic":"three","partition":3,"replicas":[1,0]}]
22
}
23
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute
24
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify
25

5) Kafka压测操作

17
1
# 生产者压测
2
bin/kafka-producer-perf-test.sh 
3
    --topic test 
4
    --record-size 1024        # 一条信息有多大，单位是字节，本次测试设置为 1k
5
    --num-records 1000000     # 总共发送多少条信息，本次测试设置为 100 万条
6
    --throughput 10000        # 每秒多少条信息，设成-1，表示不限流，尽可能快的生产数据
7
    --producer-props          # 生产者参数设置
8
        bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
9
        batch.size=16384 linger.ms=0
10

11
# 消费者压测
12
bin/kafka-consumer-perf-test.sh 
13
    --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 
14
    --topic  test  
15
    --messages  1000000   # 要消费的消息个数
16
    --consumer.config config/consumer.properties
17

第三节 生产者

1. 消息发送流程

生产者采用推（push）模式将消息发送到 broker ，具体存储于哪个分区可由生产者指定（指定分区或指定Hash Key）或轮询选出。

注意：

1. 分区数据始终先由 Leader 写入磁盘，然后复制给其它 follower 副本。

2. 生产者参数配置

3. Java API

1) 导入依赖

7
1
<dependencies>
2
    <dependency>
3
        <groupId>org.apache.kafka</groupId>
4
        <artifactId>kafka-clients</artifactId>
5
        <version>3.0.0</version>
6
    </dependency>
7
</dependencies>

2) 同步发送

33
1
public class DemoProducerBySync {
2
    public static void main(String[] args) throws ExecutionException, InterruptedException {
3
        // 1. 连接配置
4
        Properties properties = new Properties();
5
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.202.82.87:9092");
6
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
7
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
8

9
        // 2. 创建生产者
10
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
11

12
        // 3. 发送数据
13
        for (int i = 0; i < 5; i++) {
14
            // 发送
15
            Future<RecordMetadata> future = kafkaProducer.send(new ProducerRecord<>("topic_name", "data" + i));
16

17
            try {
18
                // 同步阻塞获取发送结果
19
                RecordMetadata metadata = future.get();
20

21
                // 消息发送成功
22
                System.out.println(" 主 题 ： " + metadata.topic() + "->" + "分区：" + metadata.partition());
23
            } catch (ExecutionException exception) {
24
                // 消息发送异常
25
                exception.printStackTrace();
26
            }
27

28
        }
29

30
        // 4. 关闭生产者
31
        kafkaProducer.close();
32
    }
33
}

3) 异步发送

75
1
// 异步发送消息
2
public class DemoProducer {
3
    public static void main(String[] args) throws InterruptedException {
4
        // 1. 连接配置
5
        Properties properties = new Properties();
6
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.202.82.87:9092");
7
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
8
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
9
        
10
        // 自定义分区器
11
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.xxx.xxx.DemoPartitioner"); 
12
        
13
        // 平衡吞吐量和传输延时
14
        properties.put(ProducerConfig.ACKS_CONFIG, "1"); // 只等待Leader应答
15
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); // 缓冲区大小
16
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); // 批次大小
17
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1); // 最大缓冲时间(ms)
18
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy"); // 压缩格式
19

20
        // 2. 创建生产者
21
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
22

23
        // 3. 发送数据
24
        for (int i = 0; i < 5; i++) {
25
            kafkaProducer.send(new ProducerRecord<>("topic_name", "data" + i),
26
                    // 回调函数(可选)：在收到 ack 时触发
27
                    new Callback() {
28
                        @Override
29
                        public void onCompletion(RecordMetadata metadata, Exception exception) {
30

31
                            // 消息发送异常
32
                            if (exception != null) {
33
                                exception.printStackTrace();
34
                                return;
35
                            }
36

37
                            // 消息发送成功
38
                            System.out.println(" 主 题 ： " + metadata.topic() + "->" + "分区：" + metadata.partition());
39
                        }
40
                    });
41

42
            // 模拟业务延时
43
            Thread.sleep(2);
44
        }
45

46
        // 4. 关闭生产者
47
        kafkaProducer.close();
48
    }
49
}
50

51
// 自定义分区器
52
public class DemoPartitioner implements Partitioner {
53
    @Override
54
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
55

56
        // 获取消息值
57
        String msgValues = value.toString();
58

59
        // 将vip开头的消息发送到 1 分区
60
        if (msgValues != null && msgValues.startsWith("vip:")) {
61
            return 1;
62
        }
63

64
        return 0;
65
    }
66

67
    @Override
68
    public void close() {
69
    }
70

71
    @Override
72
    public void configure(Map<String, ?> configs) {
73
    }
74
}
75

注意：

1. 除了自定义分区器外，ProducerRecord类可以直接指定消息发送的分区或消息key。
2. 默认的DefaultPartitioner 会优先使用 消息Key 的 hash 值对主题分区数进行取余发送。
3. 如果既未设置自定义分区器，又未设置消息 Key，则会使用随机粘性分区器(随机分区，但会优先填满目前分区的当前批次)。

4) 事务控制

38
1
public class DemoProducerByTranaction {
2
    public static void main(String[] args) {
3
        // 1. 连接配置
4
        Properties properties = new Properties();
5
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");
6
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
7
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
8
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tranactional_id_01"); // 【重要】指定事务id
9

10
        // 2. 创建生产者
11
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
12

13
        // 3. 开启事务
14
        kafkaProducer.initTransactions();
15
        kafkaProducer.beginTransaction();
16

17
        try {
18
            // 4. 发送数据
19
            for (int i = 0; i < 5; i++) {
20
                kafkaProducer.send(new ProducerRecord<>("first", "atguigu" + i));
21
            }
22

23
            // 模拟业务异常
24
            int i = 1 / 0;
25

26
            // 5. 提交事务
27
            kafkaProducer.commitTransaction();
28
        } catch (Exception e) {
29

30
            // 5. 回滚事务
31
            kafkaProducer.abortTransaction();
32
        } finally {
33

34
            // 6. 关闭生产者
35
            kafkaProducer.close();
36
        }
37
    }
38
}

注意：

1. 进行事务控制时， 必须开启幂等性，即 enable.idempotence 设置为 true。
2. 必须定义一个唯一的 transactional.id，这样即使生产者客户端挂掉了，重启后也能继续处理未完成的事务。

4. Spring Boot API

1) 导入依赖

4
1
<dependency>
2
    <groupId>org.springframework.kafka</groupId>
3
    <artifactId>spring-kafka</artifactId>
4
</dependency>

2) 生产者配置

9
1
# application.propeties
2

3
# 指定 kafka 的地址
4
spring.kafka.bootstrap-servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
5

6
# 【生产者】指定 key 和 value 的序列化器
7
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
8
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
9

3) 发送消息

13
1
@RestController
2
public class ProducerController {
3
    // Kafka 模板用来向 kafka 发送数据
4
    @Autowired
5
    KafkaTemplate<String, String> kafka;
6
    
7
    @RequestMapping("/atguigu")
8
    public String data(String msg) {
9
        kafka.send("first", msg);
10
        return "ok";
11
    }
12
}
13

第四节 Broker

1. ZK节点信息

注意：

1. 消费者在 zk 中注册，但生产者不在 zk 中注册。

2. Broker参数配置

3. 核心机制

1) Leader 选举机制

Controller Leader 负责管理集群 broker的上下线，所有 topic 的分区副本分配和 Leader 选举等工作。

2) Follower故障处理

3) Leader故障处理

4) 文件存储机制

提示：

1. 使用 kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.index 可以查看日志和索引文件。
2. Broker相关参数可设置文件清理策略，按时间清理或按日志大小清理等。

第五节 消费者

1. 消费流程

消费者以消费者组为单位采用 拉（pull）模式从 broker 中读取数据，每个分区在同一时间只能由组中的一个消费者进行读取，但是不同组可以同时消费这个分区。

注意：

1. 拉模式可以由消费者控制消息消费的速率，但是消费者可能会在等待消息的“长轮询”中被阻塞。

2. 消费者参数配置

3. Java API

1) 导入依赖

7
1
<dependencies>
2
    <dependency>
3
        <groupId>org.apache.kafka</groupId>
4
        <artifactId>kafka-clients</artifactId>
5
        <version>3.0.0</version>
6
    </dependency>
7
</dependencies>

2) 从指定主题消费消息

42
1
public class DemoConsumer {
2
    public static void main(String[] args) {
3

4
        // 1. 连接配置
5
        Properties properties = new Properties();
6
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");
7
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
8
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
9

10
        // 消费者组ID(必须配置)
11
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test5");
12

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

16
        // 手动提交offset
17
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
18

19
        // 2. 创建消费者
20
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
21

22
        // 3. 订阅主题
23
        kafkaConsumer.subscribe(new ArrayList<String>() {{
24
            add("first");
25
        }});
26

27
        // 4. 消费数据
28
        while (true) {
29
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
30

31
            // 执行业务
32
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
33
                System.out.println(consumerRecord);
34
            }
35

36
            // 异步提交offset
37
            kafkaConsumer.commitAsync(); // 可能会失败
38
            // kafkaConsumer.commitSync(); // 同步提交，失败重试
39
        }
40
    }
41
}
42

3) 从指定分区消费消息

33
1
public class DemoConsumerPartition {
2
    public static void main(String[] args) {
3
        // 1. 连接配置
4
        Properties properties = new Properties();
5
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");
6
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
7
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
8

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

12
        // 自动提交offset，默认为true
13
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
14
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000); // 自动提交 offset 时间间隔，默认5s
15

16
        // 2. 创建消费者
17
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
18

19
        // 3. 订阅主题指定分区
20
        kafkaConsumer.assign(new ArrayList<TopicPartition>() {{
21
            add(new TopicPartition("first", 0)); // first主题的 0 分区
22
        }});
23

24
        // 4. 消费数据
25
        while (true) {
26
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
27
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
28
                System.out.println(consumerRecord);
29
            }
30
        }
31
    }
32
}
33

4) 从指定offset消费消息

43
1

2
public class DemoConsumerSeek {
3
    public static void main(String[] args) {
4

5
        // 1. 连接配置
6
        Properties properties = new Properties();
7
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");
8
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
9
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
10

11
        // 消费者组ID(必须配置)
12
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test3");
13

14
        // 2. 创建消费者
15
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
16

17
        // 3. 订阅主题
18
        kafkaConsumer.subscribe(new ArrayList<String>() {{
19
            add("first");
20
        }});
21

22
        // 4. 查询消费者分区分配信息（有了分区分配信息才能开始消费）
23
        Set<TopicPartition> assignment = kafkaConsumer.assignment();
24
        while (assignment.size() == 0) {
25
            kafkaConsumer.poll(Duration.ofSeconds(1));
26
            assignment = kafkaConsumer.assignment();
27
        }
28

29
        // 5. 设置消从指定 offset 开始消费(对所有分区设置)
30
        for (TopicPartition topicPartition : assignment) {
31
            kafkaConsumer.seek(topicPartition, 600);
32
        }
33

34
        // 6. 消费数据
35
        while (true) {
36
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
37
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
38
                System.out.println(consumerRecord);
39
            }
40
        }
41
    }
42
}
43

5) 从指定时间消费消息

52
1
public class DemoConsumerSeekTime {
2
    public static void main(String[] args) {
3
        // 1. 连接配置
4
        Properties properties = new Properties();
5
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");
6
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
7
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
8

9
        // 消费者组ID(必须配置)
10
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test3");
11

12
        // 2. 创建消费者
13
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
14

15
        // 3. 订阅主题
16
        kafkaConsumer.subscribe(new ArrayList<String>() {{
17
            add("first");
18
        }});
19

20
        // 4. 查询消费者分区分配信息（有了分区分配信息才能开始消费）
21
        Set<TopicPartition> assignment = kafkaConsumer.assignment();
22
        while (assignment.size() == 0) {
23
            kafkaConsumer.poll(Duration.ofSeconds(1));
24
            assignment = kafkaConsumer.assignment();
25
        }
26

27
        // 5. 查询各分区 1 天前的 offset
28
        HashMap<TopicPartition, Long> timeMap = new HashMap<>();
29
        for (TopicPartition topicPartition : assignment) {
30
            timeMap.put(topicPartition, System.currentTimeMillis() - 1 * 24 * 3600 * 1000); // 每个分区都从前 1 天开始消费
31
        }
32
        Map<TopicPartition, OffsetAndTimestamp> offsetMap = kafkaConsumer.offsetsForTimes(timeMap); // 获取每个分区前 1 天时的 offset
33

34
        // 6. 设置消从指定 offset 开始消费(对所有分区设置)
35
        for (TopicPartition topicPartition : assignment) {
36
            // 获取 1 天前的分区 offset 并设置
37
            OffsetAndTimestamp offsetAndTimestamp = offsetMap.get(topicPartition);
38
            if (offsetAndTimestamp != null) {
39
                kafkaConsumer.seek(topicPartition, offsetAndTimestamp.offset());
40
            }
41
        }
42

43
        // 7. 消费数据
44
        while (true) {
45
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
46
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
47
                System.out.println(consumerRecord);
48
            }
49
        }
50
    }
51
}
52

6) 事务控制

13
1
// 自动提交offset，默认为true
2
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
3
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000); // 自动提交 offset 时间间隔，默认5s
4

5
// 手动提交offset
6
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
7

8
// 异步提交offset
9
// 没有失败重试机制，故有可能提交失败
10
kafkaConsumer.commitAsync();
11
// 同步提交offset
12
// 同步提交阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）
13
kafkaConsumer.commitSync();

提示：

1. 消费 offset 保存在内置主题__consumer_offsets中，key 是 group.id+topic+分区号，value 就是当前 offset 的值。

4. Spring Boot API

1) 导入依赖

4
1
<dependency>
2
    <groupId>org.springframework.kafka</groupId>
3
    <artifactId>spring-kafka</artifactId>
4
</dependency>

2) 消费者配置

11
1
# application.propeties
2

3
# 指定 kafka 的地址
4
spring.kafka.bootstrap-servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
5

6
# 【消费者】指定 key 和 value 的反序列化器
7
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
8
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
9

10
# 【消费者】指定消费者组的 group_id
11
spring.kafka.consumer.group-id=atguigu

3) 消费消息

8
1
@Configuration
2
public class KafkaConsumer {
3
    //  指定要监听的 topic
4
    @KafkaListener(topics = "first")
5
    public void consumeTopic(String msg) {
6
        System.out.println(" 收到的信息: " + msg);
7
    }
8
}

第六节 扩展知识

1. 生产经验

1) 生产者精确发送一次

• 保证数据至少发送一次 ：ACK 级别设置为-1 + 分区副本大于等于2 + ISR队列里应答的最小副本数量大于等于2。
• 保证数据至多发送一次 ： ACK 级别设置为0。
• 保证数据精确发送一次 ： 满足数据至少发送一次的条件，并且开启幂等性支持，使用事务控制发送。

2) 保证分区内数据有序

• 不同分区无法保证有序：不同分区可能在不同机器，无法保证有序，只能保证分区内数据有序。
• 保证分区内数据有序 ： 需设置 max.in.flight.requests.per.connection 为 1，如果未设置重试次数且开启幂等性时，允许 <= 5。

3) 消费者精确消费一次

• 重复消费：如果已经消费了数据，但是 offset 没提交，此时消费者挂掉了再起来，就会造成重复消费；
• 漏消费：相应的，如果先提交了 offset ，但是还没消费完就挂掉了，会造成漏消费。
• 精确消费一次：如果需要实现精确消费一次，那么必须将消费过程和手动提交offset 过程做原子绑定。

4) 如何提高Kafka吞吐量

• 生产者：合理设置缓冲区大小(buffer.memory)、批次大小( batch.size)、等待时间(linger.ms)、压缩类型(compression.type)等。
• 消费者：增加主题分区数，并使组下消费者数量等于分区数量；提升每批次拉取数量(max.poll.records)和大小(fetch.max.bytes)。

2. Kafka-Eagle 监控

Kafka-Eagle 框架(https://www.kafka-eagle.org/)可以监控 Kafka 集群的整体运行情况，在生产环境中经常使用。

3. Kafka-Kraft 模式

在 Kafka 现有架构中（左图），元数据在 zookeeper 中，运行时动态选举 controller，由 controller 进行 Kafka 集群管理。

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

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