轻松了解Kafka：入门与架构深度解析指南

最编程 2024-02-19 21:54:36

...

kafka 工作流程及文件存储机制

在这里插入图片描述

一个 topic 下的每一个分区都单独维护一个 offset，所以分发到不同分区中的数据是不同的数据。消费者的分区维护的是一个消费者组一个主题的一个分区维护一个 offset

同一个消费者组能够支持断点续传：

$ ./kafka-console-consumer.sh --bootstrap-server u22a:9092 --topic first --group kafka1
$ ./kafka-console-producer.sh --bootstrap-server u22a:9092 --topic first

在这里插入图片描述
文件存储机制：

在这里插入图片描述

index 和 log 文件以当前 segment 的第一条消息的 offset 命名

在这里插入图片描述

index 文件存储索引信息，索引信息按照数组逻辑排列。log 文件存储数据，数据直接紧密排列，索引文件中的元数据指向对应数据文件中的 message 的物理偏移地址

Kafka 生产者

消息发送流程

在这里插入图片描述

异步发送 API

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) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "u22a:9092");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0; i < 10; i++) {
        	//ProducerRecord<K, V>(totpic, value);
            producer.send(new ProducerRecord<String, String>("first", "atguigu " + i));
        }
        producer.close();
    }
}

在终端监视：

$ ./kafka-console-consumer.sh --bootstrap-server u22a:9092 --topic first --group kafka1
atguigu 0
atguigu 1
atguigu 2
atguigu 3
atguigu 4
atguigu 5
atguigu 6
atguigu 7
atguigu 8
atguigu 9

producer 在关闭之前会 flush 缓冲区

public class CustomProducer {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "u22a:9092");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 非必要参数
        properties.put("batch.size", 16384);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 33554432);

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<String, String>("first", "atguigu " + i));
        }
        producer.close();   // flush
    }
}

生产者有回调函数的 API：

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

public class CustomProducerWithCallBack {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "u22a:9092");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 非必要参数
        properties.put("batch.size", 16384);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 33554432);

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<String, String>("first", "atguigu " + i), new Callback() {
                // 匿名子类：直接重写接口中的方法
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    // 发送消息成功，收到 ack 时调用
                    // 发送消息遇到异常，也会调用
                    if (e != null) {
                        e.printStackTrace();
                    } else {
                        System.out.println("get ack from " + recordMetadata.topic() + ": "
                                + recordMetadata.partition() + ": " + recordMetadata.offset());
                    }
                }
            });
        }
        producer.close();   // flush
    }
}

同步发送 API

send() 函数返回一个 Future 对象，直接对其调用 get() 方法即可同步调用

public class CustomProducerWithCallBackSync {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "u22a:9092");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 非必要参数
        properties.put("batch.size", 16384);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 33554432);

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<String, String>("first", "atguigu " + i), new Callback() {
                // 匿名子类：直接重写接口中的方法
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    // 发送消息成功，收到 ack 时调用
                    // 发送消息遇到异常，也会调用
                    if (e != null) {
                        e.printStackTrace();
                    } else {
                        System.out.println("get ack from " + recordMetadata.topic() + ": "
                                + recordMetadata.partition() + ": " + recordMetadata.offset());
                    }
                }
            }).get();
            System.out.println("send " + i);
        }
        producer.close();   // flush
    }
}

分区策略

指明 partition 的情况下，直接将指明的值作为 partition 的值
没有指明 partition 值但有 key 的情况下，将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值
既没有 partition 值也没有 key 值的情况下，kafka 采用 Sticky Partition，随机选择一个分区，并尽可能一直使用该分区，待该分区的 batch 已满或者已完成，kafka 再随机选择一个分区使用

指定分区：

for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<String, String>("first", 0, "", "atguigu " + i), new Callback() {
                // 匿名子类：直接重写接口中的方法
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    // 发送消息成功，收到 ack 时调用
                    // 发送消息遇到异常，也会调用
                    if (e != null) {
                        e.printStackTrace();
                    } else {
                        System.out.println("get ack from " + recordMetadata.topic() + ": "
                                + recordMetadata.partition() + ": " + recordMetadata.offset());
                    }
                }
            });
        }

自定义分区器

// CustomPartitioner.java
public class CustomPartitioner implements Partitioner {
    public static void main(String[] args) throws InterruptedException {
    }

    @Override
    public int partition(String s, Object o, byte[] bytes, Object o1, byte[] bytes1, Cluster cluster) {
        String s1 = o1.toString();
        if (s1.contains("atguigu")) {
            return 1;
        }
        return 0;
    }

    @Override
    public void close() {
    }

    @Override
    public void configure(Map<String, ?> map) {
    }
}

在生产者中注册分区器即可：

// 注册使用自定义分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.atguigu.kafka.partition.CustomPartitioner");

数据可靠性

数据可靠性：ack + 全同步机制

为了保证 producer 发送的数据能可靠的发送到指定的 topic，topic 的每个 partition 收到 producer 发送的数据后，都需要向 producer 发送 ack，如果 producer 收到 ack，就会进行下一轮发送，否则重新发送

在这里插入图片描述

kafka 选用了第二种方案，虽然它受到网络延迟的影响，但是由于集群一般位于同一个局域网，网速对 kafka 的影响比较小

第二种方案带来一个问题：如果有一个 follower 单点故障，迟迟不能与 leader 进行同步，那 leader 就要一直等下去。为此，kafka 引入了 ISR：in-sync replica set

在这里插入图片描述

在不同的时间点回复 ack 会影响速度和数据可靠性，这个级别可以通过参数 acks 配置：

0：partition 的 leader 接收到消息还没写入磁盘就返回 ack，当 leader 故障就会丢失数据，但是这样延迟最低
1：partition 的 leader 接收到消息落盘成功后回复 ack，如果在 follower 同步成功之前 leader 故障，会丢失数据
-1：全部落盘成功才回复 ack。但是如果在 follower 同步完成后，broker 发送 ack 之前，leader 发生故障，那么会造成数据重复

注意 acks == 1 的情况，数据还存在原 leader 的磁盘里没有丢失，但是因为选举机制，新的 leader 无法感知原数据的存在，从整个系统来看，数据丢失了：

在这里插入图片描述

数据重复的情况：

在这里插入图片描述

将各自 log 文件高于 HW 的部分截掉，然后从新的 leader 同步数据

Eaxctly Once

在这里插入图片描述

Producer 事务

在这里插入图片描述

Kafka 消费者

消费方式

在这里插入图片描述

基础消费者

public class CustomConsumer {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "u22a: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");
        // 必须设置消费者组： --group kafka2
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "kafka2");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);

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

        while(true) {
            // 设置超时等待时长
            ConsumerRecords<String, String> res = kafkaConsumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> r : res) {
                System.out.println(r.toString());
            }
        }
    }
}

消费者组

同一个主题的分区，同一时刻只能有一个消费者消费

重新发送到一个全新的主题中，由于默认创建的主题分区数为 1，可以看到只有一个消费者消费到数据

分区分配策略

一个消费者组中有多个消费者，一个主题下有多个分区，所以必然会涉及到分区的分配问题，即确定哪个分区由哪个消费者消费

kafka 有 3 种分配策略：RoundRobin，Range 和 Sticky。默认使用 Range 分区器

在这里插入图片描述

更改分区分配策略：

properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RoundRobinAssignor");

在这里插入图片描述

粘性分区分配策略（StickyAssignor），首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化

offset 的维护

由于 consumer 在消费过程中可能会出现断电宕机等故障，consumer 恢复后，需要从故障前的位置的继续消费，所以consumer需要实时记录自己消费到了哪个 ofset，以便故障恢复后继续消费

Kafka 0.9 版本之前，consumer 默认将 offset 保存在 Zookeeper 中。从 0.9 版本开始，consumer 默认将 offset 保存在 Kafka一个内置的 topic 中，该topic为 __consumer_offsets

查看该主题：

先修改配置文件，增加配置项 exclude.internal.topics=false

./bin/kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server u22b:9092 --consumer.config config/consumer.properties --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning

自动提交 offset

enable.auto.commit：是否开启自动提交 offset
auto.commit.interval.ms：自动提交 offset 时间间隔

properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

重置 offset

auto.offset.reset = earliest | latest | none

当 kafka 中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量时

如果一直使用同一个消费者组，会触发断点续传，能够消费到之前的数据

如果使用一个新的消费者组来消费，会触发 offset 重置，相当于 from beginning

// 新的消费者组
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "kafka3");
// 一旦使用新的消费者组，重置 offset
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

手动提交 offset

如果自动提交 offset，会在内存中拉取到数据的时候就完成 offset 的提交

ConsumerRecords<String, String> res = kafkaConsumer.poll(Duration.ofSeconds(1));

手动提交 offset 的方法有 2 种，分别是 commitSync（同步提交）和 commitAsync （异步提交）

二者的相同点是，都会将本次 poll 的一批数据最高的偏移量提交

不同点是，commitSync 会阻塞当前线程，一直到提交成功，并且失败后会自动重试

commitAsync 没有失败重试机制，故有可能提交失败

首先关闭自动提交的配置参数：

properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
// properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

ConsumerRecords<String, String> res = kafkaConsumer.poll(Duration.ofSeconds(1));
 // kafkaConsumer.commitSync();     // 同步提交
 kafkaConsumer.commitAsync(new OffsetCommitCallback() {
     @Override
     public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
         if (e != null) {
             e.printStackTrace();
         } else {
             System.out.println(map);
         }
     }
 }); // 异步提交，更高效

Consumer 事务（精准一次性消费）

kafka 消费端将消费过程和提交 offset 过程做原子绑定

Kafka 高效读写数据

顺序写磁盘：写的过程是一直追加到文件末端，为顺序写

在这里插入图片描述
零拷贝技术：

在这里插入图片描述

Zookeeper 在 Kafka 中的作用

上一篇： pyspark.mllib.feature module-Sample Code:

下一篇： JoyoshareMac版视频切割工具 (一款针对Mac的简易视频剪辑软件)

轻松了解Kafka：入门与架构深度解析指南

kafka 工作流程及文件存储机制

Kafka 生产者

消息发送流程

异步发送 API

同步发送 API

分区策略

自定义分区器

数据可靠性

Eaxctly Once

Producer 事务

Kafka 消费者

消费方式

基础消费者

消费者组

分区分配策略

offset 的维护

自动提交 offset

重置 offset

手动提交 offset

Consumer 事务（精准一次性消费）

Kafka 高效读写数据

Zookeeper 在 Kafka 中的作用

初探X86与ARM处理器架构：轻松入门指南

玩转Kotlin性能测试：JMH入门指南一 - 测试基础" "深入理解JMH在Kotlin中的应用：基准测试实战解析" "轻松实践Kotlin基准测试：JMH工具详解与实例总结

速成指南：轻松掌握Python中的机器学习与深度学习之Numpy基础全面解析

轻松了解Kafka：入门与架构深度解析指南

实战指南：从入门到精通--强化学习实例解析与面试必备知识(10讲)：模仿学习、行为复制、反向强化学习、旁观者视角模仿学习、序列生成及聊天机器人的深度剖析

立体匹配入门指南（8）：视差图、深度图与点云解析