轩辕李的博客
轩辕李
2023-05-15
目录

Kafka实战

# 一、Kafka基础介绍

Kafka是一种高吞吐量、可扩展的分布式消息队列系统，用于处理大规模的实时数据流。

与Rabbitmq对比，Kafka的优势主要是高吞吐量、可扩展性、实时数据处理能力。

# 1、Kafka的历史和演变

Kafka最初是由LinkedIn开发的，并于2011年开源。它的设计目标是为了解决LinkedIn在处理大规模实时数据流时遇到的问题。随着时间的推移，Kafka逐渐成为了一个独立的项目，并在业界广泛应用。

# 2、Kafka的核心组件和架构

Kafka的核心组件包括生产者（Producer）、消费者（Consumer）和代理（Broker）。生产者负责将消息发布到Kafka集群，消费者负责从Kafka集群中读取消息，而代理是Kafka集群中的中间层，负责管理消息的存储和分发。

Kafka的架构采用了分布式的设计，它可以将数据分散到多个代理（Broker）上，以实现数据的高可靠性和可扩展性。Kafka使用主题（Topic）来组织和分类消息，每个主题可以有多个分区（Partition），而每个分区可以在不同的代理上进行复制和分布。这种分区和复制的机制使得Kafka可以处理大规模的数据流，并提供了容错和高可用性。

# 3、Kafka与其他消息队列系统的对比

Kafka与其他消息队列系统相比具有一些独特的特点和优势。与传统的消息队列系统相比，Kafka在以下几个方面有所突出：

  • 高吞吐量：Kafka能够处理每秒数百万条消息的高吞吐量，适用于处理大规模的实时数据流。
  • 持久性：Kafka将消息持久化到磁盘，确保消息不会丢失。消费者可以按需从任意位置读取消息。
  • 可扩展性：Kafka的分布式设计和分区机制使得它可以在集群中水平扩展，以处理更多的数据和请求。
  • 多样的数据处理方式：Kafka不仅可以用作消息队列，还可以用作事件日志、流处理平台和存储系统等多种用途。
  • 生态系统丰富：Kafka有一个庞大的生态系统，包括与流处理、数据集成和监控等相关的工具和技术。

与Rabbitmq对比，Kafka的劣势在于灵活性差（不支持复杂的消息路由和交换机类型）、较小的生态系统。

RabbitMQ适用于更灵活的消息传递场景，注重消息的可靠性和一致性，而Kafka则适用于需要处理大规模实时数据流、高吞吐量和流处理的场景。选择使用哪种消息队列系统取决于应用程序的需求和特定的使用情况。

# 二、Kafka环境搭建

在了解了Kafka的基础知识之后，接下来将介绍如何搭建Kafka的环境。

# 1、Kafka的安装和配置

要安装Kafka，您可以按照以下步骤进行操作：

步骤 1：下载Kafka 前往Kafka官方网站（https://kafka.apache.org/downloads (opens new window)）下载适用于您操作系统的最新版本的Kafka。

您的本地环境必须安装Java 8+。

步骤 2：解压文件 解压下载的Kafka压缩文件到您选择的目录。

步骤 3：配置Kafka 在Kafka的配置文件（通常是server.properties）中进行必要的配置。您可以指定ZooKeeper连接信息、Kafka监听地址、日志目录等。

步骤 4：启动Kafka 使用命令行或终端进入Kafka安装目录，并执行以下命令启动Kafka服务器：

bin/kafka-server-start.sh config/server.properties

Kafka将会启动并监听配置中指定的地址和端口。

config/server.properties 配置文件中的主要配置项说明：

  1. broker.id: Kafka 代理节点的唯一标识符。每个节点在集群中必须具有唯一的 ID。

  2. listeners: 监听器的地址和端口配置。指定 Kafka 监听连接请求的地址和端口，例如 PLAINTEXT://localhost:9092。

  3. log.dirs: Kafka 存储消息日志的目录路径。可以指定一个或多个目录，用逗号分隔。

  4. num.network.threads: 处理网络请求的线程数。指定用于处理客户端和复制网络请求的线程数。

  5. num.io.threads: 处理磁盘 I/O 的线程数。用于处理磁盘读写请求的线程数。

  6. socket.send.buffer.bytes 和 socket.receive.buffer.bytes: 定义套接字的发送缓冲区和接收缓冲区的大小。

  7. socket.request.max.bytes: 定义单个网络请求的最大字节数。超过该大小的请求将被拒绝。

  8. log.retention.hours: 定义消息日志保留的时间。超过指定时间的消息将被删除。

  9. log.segment.bytes: 定义每个消息日志段（segment）的大小。当一个日志段达到指定大小时，会被关闭并创建新的日志段。

  10. num.partitions: 每个主题的默认分区数。可以通过创建主题时进行覆盖。

  11. offsets.topic.replication.factor: 存储消费者偏移量的主题的副本因子。指定该主题的副本数量。

  12. transaction.state.log.replication.factor: 存储事务状态的主题的副本因子。指定该主题的副本数量。

# 2、Kafka with KRaft

在介绍 Kafka with KRaft 之前，我们先了解一下 Kafka 的传统架构。

传统的 Kafka 架构中，Kafka 使用 ZooKeeper 作为其元数据存储和协调服务。ZooKeeper 负责管理 Kafka 代理节点、主题和分区的元数据信息，并协调生产者和消费者之间的消息传递。

然而，随着 Kafka 的规模不断增长和需求的变化，使用 ZooKeeper 作为协调服务面临一些挑战，例如可靠性、性能和可扩展性。为了克服这些问题，Kafka 社区引入了 Kafka with KRaft。

Kafka with KRaft 是一个新的 Kafka 元数据管理和协调的实现方式，它不再依赖于 ZooKeeper。相反，它使用了 KRaft（Kafka Raft）算法，以实现高可用性和一致性。

KRaft 是一种领导者选举算法，它基于 Raft 算法，并针对 Kafka 进行了优化。它将 Kafka 的元数据分片复制到多个 KRaft 服务器节点上，确保高可用性和容错性。每个分片都有一个领导者节点，负责处理元数据更新和协调请求。如果领导者节点失效，KRaft 算法会从备份节点中选举出新的领导者。

使用 Kafka with KRaft，Kafka 获得了以下优势：

  1. 去中心化：Kafka with KRaft 不再依赖于外部的 ZooKeeper，从而减少了单点故障的风险，提高了整个系统的可靠性。
  2. 简化架构：Kafka with KRaft 简化了 Kafka 的架构，减少了依赖项和管理成本。
  3. 高性能：KRaft 算法针对 Kafka 进行了优化，提供了更高的性能和吞吐量。
  4. 可扩展性：Kafka with KRaft 具备良好的可扩展性，可以轻松地根据需求添加或删除节点。
  5. 自动化故障转移：Kafka with KRaft 通过自动进行领导者选举，实现了故障转移的自动化。

它的安装步骤如下： 生成集群UUID

KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

格式化日志目录

bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties

启动服务

bin/kafka-server-start.sh config/kraft/server.properties

一旦 Kafka 服务器成功启动，您将拥有一个基本的 Kafka 环境可供使用。

# 3、Docker安装

你也可以使用Docker安转 (opens new window)，就不赘述了。

# 三、Kafka生产者（Producer）

Kafka生产者是将消息发布到Kafka集群的组件。

# 1、生产者的角色和职责

Kafka生产者的主要角色是将消息发布到Kafka集群的指定主题中。其职责包括：

  • 构建消息：生产者负责构建要发送的消息。消息可以是任何格式的数据，例如字符串、JSON、二进制等。
  • 发送消息：生产者使用Kafka提供的API将消息发送到指定的主题中。发送消息的过程是异步的，生产者将消息批量发送到Kafka集群。
  • 处理发送结果：生产者可以选择处理消息发送的结果，包括确认消息是否已成功发送、处理发送失败等情况。

# 2、生产者的API使用

Kafka提供了丰富的API来使用生产者发送消息。以下是使用Java语言的示例代码，展示了如何使用Kafka生产者API发送消息：

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

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        String topic = "my-topic";
        String key = "my-key";
        String value = "Hello, Kafka!";

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

        try {
            RecordMetadata metadata = producer.send(record).get();
            System.out.println("Message sent successfully. Offset: " + metadata.offset());
        } catch (InterruptedException | ExecutionException e) {
            System.err.println("Error sending message: " + e.getMessage());
        }

        producer.close();
    }
}

在上面的代码示例中，我们首先设置了Kafka集群的连接信息，并配置了消息的键（key）和值（value）的序列化器。

然后，创建了一个Kafka生产者实例，并指定要发送消息的主题、键和值。

最后，通过调用producer.send()方法发送消息，同时提供一个回调函数以处理发送结果。

# 3、生产者的性能优化

为了提高Kafka生产者的性能，可以采取以下一些优化策略：

  • 批量发送：将多条消息批量发送到Kafka集群，而不是逐条发送。可以通过配置batch.size和linger.ms参数来调整批量发送的大小和等待时间。

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

import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("batch.size", 16384); // 设置批量大小
        props.put("linger.ms", 10); // 设置等待时间

        Producer<String, String> producer = new KafkaProducer<>(props);
        String topic = "my-topic";

        for (int i = 0; i < 1000; i++) {
            String key = "key-" + i;
            String value = "Message-" + i;
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

            producer.send(record);
        }

        producer.close();
    }
}

    在上述示例中，我们通过设置batch.size为16384字节，表示当消息的累计大小达到该值时，会将批量消息发送到Kafka集群。同时，通过设置linger.ms为10毫秒，表示如果消息在指定的等待时间内未达到批量大小，则也会触发批量发送。

    在for循环中，我们生成了1000条消息，并通过producer.send()方法逐条发送。由于配置了批量发送，当累计的消息大小达到批量大小或等待时间到达时，批量消息将被发送到Kafka集群。

    使用批量发送可以显著提高生产者的吞吐量，因为减少了单条消息的发送开销。

    请根据实际需求调整batch.size和linger.ms的值，以获得最佳性能和吞吐量。

  • 异步发送：将发送消息的过程设置为异步操作，以避免阻塞主线程。

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

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        String topic = "my-topic";
        String key = "my-key";
        String value = "Hello, Kafka!";

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

        producer.send(record, new Callback() {
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception == null) {
                    System.out.println("Message sent successfully. Offset: " + metadata.offset());
                } else {
                    System.err.println("Error sending message: " + exception.getMessage());
                }
            }
        });

        // 继续执行其他操作...

        producer.close();
    }
}

    在上述代码中，我们通过调用producer.send()方法发送消息时，传入了一个回调函数。这样可以在消息发送完成后异步处理结果，而不需要等待发送的确认。这种异步发送方式可以提高生产者的吞吐量。

  • buffer.memory：该参数控制生产者可用于缓冲等待发送到服务器的记录的总内存大小。当生产者发送消息时，首先会将消息添加到发送缓冲区。如果发送缓冲区的空间不足以容纳新的消息，并且buffer.memory参数指定的总内存大小已经耗尽，生产者将无法继续发送消息，直到有足够的空间可用。

    props.put("buffer.memory", "16777216"); // 设置为16MB，默认也是32MB

  • acks：该参数指定了需要生产者收到服务器确认的消息发送模式。它有三个可选值：acks=0（不需要任何确认）、acks=1（在领导者副本收到消息后确认）和acks=all（在所有副本都收到消息后确认）。较高的确认级别可以提供更高的可靠性，但也会增加延迟。以下是示例代码展示如何设置acks参数：

    props.put("acks", "all");  // 默认值为1

# 四、Kafka消费者（Consumer）

Kafka消费者扮演着从Kafka集群中读取消息的角色，它们负责订阅和消费特定主题的消息。

# 1、消费者的角色和职责

Kafka消费者的主要职责是从Kafka集群中获取消息并进行处理。消费者可以根据自己的需求订阅一个或多个主题，并从订阅的主题中拉取消息。消费者可以以不同的方式处理消息，如打印消息、将消息写入数据库等。

# 2、消费者的API使用

Kafka提供了丰富的API来使用和管理消费者。

public class KafkaConsumerExample {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my-consumer-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("topic1", "topic2"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Received message: " + record.value());
                // 处理消息的逻辑
            }
        }
    }
}

上述代码中，我们使用Properties对象来设置消费者的配置，包括连接的Kafka服务器地址、消费者组ID以及消息的键值反序列化器。

然后，我们创建一个KafkaConsumer实例。使用subscribe方法，我们可以让消费者订阅一个或多个主题。在示例中，我们订阅了名为"topic1"和"topic2"的两个主题。

使用poll方法从Kafka集群中拉取消息。然后，我们遍历接收到的消息并进行处理。这里的处理逻辑可以根据实际需求进行自定义。

# 3、消费者的性能优化

为了优化Kafka消费者的性能，我们可以考虑以下几个方面：

# 3.1、批量拉取

通过调整poll方法的参数来一次拉取多条消息，减少与Kafka集群的交互次数，提高效率。

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

consumer.poll(Duration.ofMillis(100))中的参数Duration.ofMillis(100)指定了消费者在等待可用消息时的最大时间。它表示消费者在一次poll调用中最多等待100毫秒来获取消息。

# 3.2、多线程消费

可以使用多个消费者线程来并行处理消息，提高吞吐量和并发性能。

int numThreads = 5;
ExecutorService executor = Executors.newFixedThreadPool(numThreads);

for (int i = 0; i < numThreads; i++) {
    executor.submit(() -> {
        KafkaConsumer<String, String> threadConsumer = new KafkaConsumer<>(props);
        threadConsumer.subscribe(Arrays.asList("topic1", "topic2"));

        while (true) {
            ConsumerRecords<String, String> records = threadConsumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Received message: " + record.value());
                // 处理消息的逻辑
            }
        }
    });
}

executor.shutdown();

在这个示例中，我们创建了一个固定大小的线程池，并为每个线程创建一个独立的消费者。每个消费者都可以订阅相同或不同的主题，并在独立的线程中进行消息的消费和处理。

# 3.3、消费者配置的优化

通过调整消费者的配置参数，如批量处理的大小、最大拉取的等待时间等，可以进一步优化消费者的性能。

下面详细介绍一些常用的配置参数及其作用：

1. fetch.min.bytes和fetch.max.wait.ms

  • fetch.min.bytes参数指定了消费者从Kafka服务器获取消息的最小字节数。如果可用的消息总字节数小于该值，消费者将等待更多的消息到达，以满足最小字节数的要求。这可以减少频繁的网络往返次数，提高性能。
  • fetch.max.wait.ms参数定义了消费者等待服务器返回响应的最大时间。如果在指定的时间内没有可用的消息，消费者将立即返回当前可用的消息。通过适当设置等待时间，可以在延迟和吞吐量之间进行权衡。

2. fetch.max.bytes

  • fetch.max.bytes参数规定了单个拉取请求返回的最大字节数。如果某个分区的消息量很大，可以适当增加该参数的值，以提高单次拉取的效率。但需要注意的是，过大的值可能导致单个请求的内存占用增加。

3. max.poll.records

  • max.poll.records参数定义了每次poll调用返回的最大记录数。通过调整该参数，可以一次处理更多的消息，提高消费者的吞吐量。但需要注意的是，过大的值可能会增加单次处理消息的时间，导致较高的延迟。

4. session.timeout.ms和heartbeat.interval.ms

  • session.timeout.ms参数定义了消费者与消费者组协调器之间的会话超时时间。如果消费者在该时间内没有发送心跳给协调器，协调器将认为消费者失效，并将其剔除出消费者组。通过适当调整该参数，可以平衡消费者的故障检测时间和响应速度。
  • heartbeat.interval.ms参数规定了消费者发送心跳的频率。较短的心跳间隔可以更快地检测到消费者故障，但也会增加网络开销。根据网络状况和消费者组的大小，可以调整该参数的值。

5. 其他参数

除了上述参数之外，还有其他一些可供配置的参数，如max.partition.fetch.bytes（单个分区一次拉取的最大字节数）、enable.auto.commit（是否启用自动提交位移）等。

# 五、Kafka主题和分区

Kafka的主题（Topic）和分区（Partition）是构成其核心架构的重要组成部分。

# 1、主题（Topic）和分区（Partition）的概念

# 1.1、主题（Topic）

在Kafka中，主题是消息的逻辑分类单位。它类似于传统消息系统中的队列或主题的概念。每个主题可以包含一个或多个分区。

# 1.2、分区（Partition）

分区是主题的物理存储单位。每个分区都是一个有序且不可变的消息序列。分区中的每条消息都会被分配一个唯一的偏移量（Offset），用于标识消息在分区中的位置。

Kafka的分区具有以下特性：

  • 每个分区都有一个领导者（Leader）和零个或多个副本（Replica）。
  • 分区中的消息按照写入顺序进行存储，并且保留一定时间。
  • 消费者可以以任意顺序从分区中读取消息，并且可以根据偏移量进行灵活的消息消费。

# 2、如何创建和管理主题

在Kafka中，可以使用命令行工具或Kafka提供的API来创建和管理主题。

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import java.util.Properties;
import java.util.Collections;

public class KafkaTopicManager {
    public static void createTopic(String bootstrapServers, String topicName, int numPartitions, short replicationFactor) {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        try (AdminClient adminClient = AdminClient.create(props)) {
            NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);
            adminClient.createTopics(Collections.singletonList(newTopic)).all().get();
            System.out.println("Topic created successfully.");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092";
        String topicName = "my-topic";
        int numPartitions = 3;
        short replicationFactor = 1;

        createTopic(bootstrapServers, topicName, numPartitions, replicationFactor);
    }
}

上述代码使用Kafka的AdminClient来创建一个新主题。需要指定Kafka集群的引导服务器地址（bootstrapServers），主题名称（topicName），分区数量（numPartitions）和副本因子（replicationFactor）。执行该代码将创建一个名为"my-topic"的主题，包含3个分区和1个副本。

# 3、分区策略和数据平衡

Kafka使用分区策略来决定将消息分配到每个分区的方式。默认情况下，Kafka使用Round-robin（轮询）策略，将消息均匀地分配到每个分区。但是，您也可以自定义分区策略来根据特定的业务需求进行分区。

在Kafka集群中，分区的分布和数据平衡是非常重要的。数据平衡确保每个代理节点上的分区数量和数据量相对均衡，以充分利用集群的资源并提高整体性能。Kafka通过重新分配分区和副本来实现数据平衡。

以下是一些常用的分区策略：

  • Round-robin策略：默认的分区分配策略。它将消息依次分配到每个分区，实现相对均匀的负载。
  • Key-based策略：基于消息的键（Key）进行分区，确保具有相同键的消息被分配到同一个分区。这样可以确保具有相关性的消息被顺序处理。
  • 自定义策略：根据业务需求编写自定义的分区策略，例如根据某个特定字段进行分区。

对于数据平衡，Kafka提供了自动的分区重新分配和副本平衡机制。当集群中的代理节点发生故障或新增节点时，Kafka会自动将分区重新分配到可用的节点上，并确保每个节点上的分区数相对均衡。

您可以使用Kafka提供的工具和监控功能来监控分区的分布和数据平衡情况，并根据需要采取相应的调整措施。

# 4、消息顺序性

Kafka保证了特定条件下的消息顺序性。具体来说，对于每个分区，Kafka保证了消息的有序性。也就是说，对于同一个分区内的消息，它们将按照被写入的顺序进行存储和读取。

然而，对于多个分区的情况，Kafka不能保证全局顺序性。这是因为Kafka的设计目标之一是实现高吞吐量和可扩展性，通过并行地处理多个分区来提高消息处理能力。因此，多个分区之间的消息可能会以并发方式进行处理，导致消息在消费者端的到达顺序与发送顺序不完全一致。

尽管如此，Kafka提供了一些机制来实现特定场景下的顺序性要求：

  1. 单分区顺序性：对于只使用单个分区的主题，Kafka可以保证消息的严格顺序性。在此情况下，生产者将所有相关消息发送到同一个分区，消费者从该分区读取消息，确保消息按照发送顺序进行处理。

  2. 分区键（Key）：在发送消息时，生产者可以为每条消息指定一个键。Kafka使用键来确定消息应该被分配到哪个分区。通过选择恰当的键，生产者可以确保具有相同键的消息被分配到同一个分区，从而实现相关消息的顺序处理。

  3. 仅单个消费者：如果只有一个消费者订阅了特定分区的消息，那么该消费者将按照消息的顺序进行处理。这是因为Kafka将分区内的消息按照顺序传递给单个消费者。

总的来说，Kafka提供了一些机制来实现特定条件下的消息顺序性。在单分区场景下，Kafka可以提供严格的顺序性。对于多分区场景，可以通过选择合适的分区键和消费者订阅策略来实现相关消息的顺序处理。

# 六、Kafka的数据持久性和可靠性

Kafka作为一种分布式消息队列系统，具备数据的持久性和可靠性。

# 1、Kafka如何确保数据的持久性

Kafka通过将消息持久化到磁盘来确保数据的持久性。一旦消息被写入Kafka，它将被保存在磁盘上，即使在写入后出现故障，消息也不会丢失。

Kafka使用了顺序写和批量提交的方式来提高写入性能。生产者将消息追加到日志（Log）文件的末尾，并记录消息在日志文件中的偏移量（Offset）。消费者可以根据偏移量来读取消息，从而实现消息的顺序访问。

# 2、Kafka的副本（Replica）管理

为了提高可靠性和容错性，Kafka使用了副本机制。副本是指在Kafka集群中对分区的复制。每个分区可以有多个副本，其中一个副本被指定为领导者（Leader），其他副本被称为追随者（Follower）。

领导者负责处理生产者和消费者的请求，并保持与追随者之间的同步。追随者从领导者处复制消息，并保持与领导者的同步。如果领导者发生故障，某个追随者会被选举为新的领导者。

Kafka的副本管理涉及到以下几个关键概念：

  • ISR（In-Sync Replica）：ISR是指与领导者保持同步的副本集合。只有处于ISR中的副本才能成为新的领导者。如果某个副本与领导者的同步滞后超过一定程度，它将被移除出ISR。
  • AR（Assigned Replica）：AR是指被分配到某个代理节点的副本，无论它是领导者还是追随者。
  • UNCLEAN LEADER ELECTION：UNCLEAN LEADER ELECTION是指在发生故障时，允许不完全同步的副本成为新的领导者。这种情况下可能会导致数据的丢失，因此不推荐使用UNCLEAN LEADER ELECTION。

# 3、Kafka的故障恢复策略

Kafka具备强大的故障恢复能力，它可以在节点故障的情况下保持服务的可用性和数据的一致性。

当一个代理节点发生故障时，Kafka采用以下故障恢复策略：

  • 故障检测：Kafka通过心跳机制和元数据的定期更新来检测代理节点的故障。如果一个节点长时间未发送心跳或更新元数据，Kafka将认为该节点已宕机。
  • 自动副本重新分配：一旦发现节点故障，Kafka会自动触发副本重新分配。它会将故障节点上的分区副本重新分配到其他健康的节点上，以保持副本的复制和数据的可用性。
  • 领导者选举：如果领导者节点发生故障，Kafka会自动进行领导者选举，从ISR中选举一个副本作为新的领导者。这确保了在节点故障的情况下，Kafka仍然能够提供读写服务。
  • 数据复制：Kafka使用异步的方式将消息从领导者复制到追随者。这种异步复制的机制可以提高写入性能，但也可能导致部分数据的丢失。为了确保数据的可靠性，可以使用配置参数来调整复制的策略，例如增加追随者的数量或减小复制的延迟。

下面是一个Java示例，展示了如何使用Kafka的Java客户端API来创建一个副本为1的主题：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.errors.TopicExistsException;
import java.util.Collections;
import java.util.Properties;

public class KafkaTopicExample {

    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String TOPIC_NAME = "my-topic";
    private static final int NUM_PARTITIONS = 1;
    private static final short REPLICATION_FACTOR = 1;

    public static void main(String[] args) {
        createTopic();
    }

    public static void createTopic() {
        Properties props = new Properties();
        props.put("bootstrap.servers", BOOTSTRAP_SERVERS);

        try (AdminClient adminClient = AdminClient.create(props)) {
            NewTopic newTopic = new NewTopic(TOPIC_NAME, NUM_PARTITIONS, REPLICATION_FACTOR);
            adminClient.createTopics(Collections.singleton(newTopic)).all().get();
            System.out.println("Topic created successfully");
        } catch (Exception e) {
            if (e.getCause() instanceof TopicExistsException) {
                System.out.println("Topic already exists");
            } else {
                e.printStackTrace();
            }
        }
    }
}

通过以上代码，您可以使用Kafka的Java客户端API创建一个名为"my-topic"的主题，设置分区数量为1，副本数量为1。

# 4、按需消费和回溯消费

对于单个分区内的数据，消费者可以根据偏移量来读取消息，从而实现消息的顺序访问。

当使用Kafka的Java消费者API时，可以实现按需消费和回溯消费的场景。

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.*;

public class KafkaConsumerExample {

    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String TOPIC_NAME = "my-topic";
    private static final String GROUP_ID = "my-group";
    private static final int MAX_RECORDS = 10;

    public static void main(String[] args) {
        consumeMessages();
    }

    public static void consumeMessages() {
        Properties props = new Properties();
        props.put("bootstrap.servers", BOOTSTRAP_SERVERS);
        props.put("group.id", GROUP_ID);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Collections.singletonList(TOPIC_NAME));

        // 按需消费
        int consumedRecords = 0;
        while (consumedRecords < MAX_RECORDS) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Offset: %d, Key: %s, Value: %s\n",
                        record.offset(), record.key(), record.value());
                consumedRecords++;
            }
        }

        // 回溯消费
        Set<TopicPartition> assignedPartitions = consumer.assignment();
        Map<TopicPartition, Long> offsetsToReset = new HashMap<>();
        for (TopicPartition partition : assignedPartitions) {
            offsetsToReset.put(partition, 0L);
        }
        consumer.seekToBeginning(offsetsToReset);

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Offset: %d, Key: %s, Value: %s\n",
                        record.offset(), record.key(), record.value());
            }
        }
    }
}

在上述示例中，我们首先订阅了一个主题（Topic）。然后，我们使用一个计数器变量consumedRecords来实现按需消费，当达到指定的消费记录数量MAX_RECORDS时停止消费。

接下来，我们演示了回溯消费的场景。通过seekToBeginning()方法，我们将消费者的位置回溯到每个分区的起始偏移量，从而重新开始消费所有消息。这样可以实现回溯消费的需求，即消费之前的所有消息。

请注意，上述示例中的回溯消费是一个无限循环，它会持续地从Kafka集群中拉取并消费消息。您可以根据实际需求进行相应的停止条件设置。

# 七、Kafka流（Kafka Streams）

Kafka流是Kafka提供的一个强大的流处理库，它能够实时地对输入数据进行处理和转换。

# 1、Kafka流的概念和使用场景

Kafka流是一种轻量级的流处理库，它充分利用了Kafka的分布式和容错特性，提供了一种简单而高效的方式来处理实时数据流。Kafka流适用于以下场景：

  • 数据转换和处理：Kafka流可以将输入数据进行转换、过滤、聚合等操作，并将结果输出到新的Kafka主题中。
  • 实时计算：Kafka流支持基于事件时间的窗口操作，例如时间窗口、滑动窗口等，可以进行实时计算和统计。
  • 数据集成：Kafka流能够从多个输入流中读取数据，进行合并、连接等操作，实现数据集成和数据流的交互。
  • 实时监控和反馈：Kafka流可以实时处理来自多个数据源的数据，并提供实时的监控和反馈机制。

# 2、Kafka流的API使用

Kafka流提供了简洁而强大的API，使得开发人员能够方便地编写流处理应用程序。

要在Java项目中使用Kafka流，您需要引入以下依赖包（Maven）：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>3.4.0</version>
</dependency>

实例：

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

public class KafkaStreamsExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-streams-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> input = builder.stream("input-topic");
        KStream<String, String> transformed = input.mapValues(value -> value.toUpperCase());
        transformed.to("output-topic");

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}

上述示例中，我们创建了一个StreamsBuilder对象，并通过builder.stream("input-topic")方法获取了一个输入流。接着，使用mapValues()方法对输入流进行转换操作，将输入数据的值转换为大写字母。最后，使用to("output-topic")方法将转换后的数据输出到另一个Kafka主题中。

通过这个简单的示例，您可以看到使用Kafka流进行流处理非常直观和易于理解。

# 3、Kafka流的性能优化和高级特性

为了进一步优化Kafka流的性能并实现一些高级功能，

当提到Kafka流的高级特性时，以下是每个特性的示例代码：

1. 状态管理： Kafka流提供了内置的状态存储机制，例如KeyValueStore，可以在处理过程中维护和查询状态。以下是一个简单示例，展示如何在Kafka流中使用状态存储：

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> input = builder.stream("input-topic");

KTable<String, Long> wordCountTable = input
    .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
    .groupBy((key, word) -> word)
    .count("WordCountStore");

wordCountTable.toStream().to("output-topic",Produced.with(Serdes.String(), Serdes.Long()));

在上述示例中，我们通过groupBy()方法按单词对输入数据进行分组，并使用count()方法对每个单词进行计数。结果将存储在名为"WordCountStore"的状态存储中。

2. 窗口操作： Kafka流支持窗口操作，可以基于事件时间或处理时间对数据流进行窗口化处理。以下是一个示例，演示如何使用时间窗口对数据进行聚合：

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Long> input = builder.stream("input-topic");

KStream<Windowed<String>, Long> windowedStream = input
    .groupByKey()
    .count(TimeWindows.of(Duration.ofMinutes(5)));

windowedStream..toStream().to("output-topic",Produced.with(Serdes.String(), Serdes.Long()));

在上述示例中，我们使用groupByKey()对输入流进行分组，然后使用count()方法对每个时间窗口（5分钟）内的数据进行计数。结果将输出到名为"output-topic"的Kafka主题中。

3. 容错和恢复： Kafka流具备容错和故障恢复机制，可以自动处理节点故障和数据丢失。以下是一个示例，展示如何设置容错和恢复机制：

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-streams-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);

StreamsBuilder builder = new StreamsBuilder();
// 构建流处理拓扑

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

在上述示例中，我们通过将REPLICATION_FACTOR_CONFIG属性设置为3，为Kafka流应用程序配置副本因子，以确保数据的可靠性和容错性。

4. 拓扑（Topology）： 在Kafka Streams中，拓扑（Topology）是指用于构建流处理应用程序的组件和操作的有向无环图（DAG）。拓扑定义了数据流的来源、转换和目的地，以及数据流之间的连接关系和处理逻辑。

使用拓扑，您可以定义和配置流处理应用程序的数据流路径，包括输入主题、输出主题，以及在数据流上执行的转换和处理操作。通过将这些操作连接在一起，构成了一个有向无环图，其中每个节点表示一个流处理操作。

public Topology buildTopology(List<ProcessorSupplier<String, MetricDataBO>> suppliers) {
    Topology build = new Topology();
    build.addSource("Source", metricDataTopic)
            .addProcessor("Process0", suppliers.get(0), "Source")
            .addStateStore(
                    getWindowsStore(STATE_STORE_NAME + "0")
                    ,"Process0");
    for (int i = 1; i < suppliers.size(); i++) {
        build.addProcessor("Process" + i, suppliers.get(i), "Process" + (i - 1))
                .addStateStore(
                        getWindowsStore(STATE_STORE_NAME + i)
                        ,"Process" + i);
    }
    build.addSink("Sink", alarmProcTopic, "Process" + (suppliers.size() - 1), "Process0");
    return build;
}

这段代码构建了一个拓扑（Topology），其中包含了一系列的处理器（Processor）和状态存储（State Store）。以下是对这段代码的总结：

  1. 首先，创建了一个空的拓扑对象：Topology build = new Topology();

  2. 使用addSource()方法向拓扑中添加一个数据源节点，命名为"Source"，该节点连接到metricDataTopic主题。

  3. 使用addProcessor()方法向拓扑中添加第一个处理器节点，命名为"Process0"，并将其连接到前一个节点"Source"。处理器供应者（ProcessorSupplier）来自suppliers列表的第一个元素。

  4. 使用addStateStore()方法向拓扑中的"Process0"节点添加状态存储，命名为STATE_STORE_NAME + "0"，该存储与处理器节点"Process0"相关联。

  5. 针对剩余的处理器供应者，使用循环将它们依次添加到拓扑中。每个处理器节点都连接到前一个处理器节点，并添加相应的状态存储。

  6. 最后，使用addSink()方法向拓扑中添加一个下沉节点（Sink），命名为"Sink"，将其连接到最后一个处理器节点"Process" + (suppliers.size() - 1)，并将数据发送到alarmProcTopic主题。同时，还将下沉节点连接到第一个处理器节点"Process0"，实现环形连接。

  7. 最后，返回构建好的拓扑对象。

总之，这段代码通过循环构建了一个具有环形连接的拓扑结构，每个处理器节点通过状态存储与前一个和后一个节点连接起来，最后的下沉节点将数据发送到指定的主题。拓扑的具体形状和流向取决于处理器供应者（suppliers列表中的元素）的个数和顺序。

5. 流-表连接： Kafka流支持流-表连接操作，可以将流数据与数据库或外部存储进行关联和查询。以下是一个示例，展示如何在Kafka流中进行流-表连接：

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> ordersStream = builder.stream("orders-topic");
// KTable是Kafka Streams提供的一个抽象概念，代表了一个可查询和可更新的关键-值存储
KTable<String, String> customerTable = builder.table("customer-table");

KStream<String, String> joinedStream = ordersStream
    .leftJoin(customerTable, (orderKey, orderValue, customerValue) ->
        "Order: " + orderValue + ", Customer: " + customerValue);

joinedStream.to("output-topic");

在上述示例中，我们通过leftJoin()方法将订单流(ordersStream)与客户表(customerTable)进行连接。根据订单和客户信息，我们可以对数据进行处理和关联，并将结果输出到"output-topic"中。

# 八、Connect

Kafka Connect是一个用于可靠地将数据从外部系统连接到Kafka和从Kafka连接到外部系统的工具。它提供了一种简单而可扩展的方式来管理数据流的传输，使得数据的导入和导出变得更加容易和灵活。下面将介绍Kafka Connect的概念和作用，以及如何使用和配置它。同时，还将介绍Kafka Connect的源连接器和接收连接器。

# 1、Kafka Connect的概念和作用

Kafka Connect是Kafka生态系统中的一部分，它旨在简化数据集成的过程。通过Kafka Connect，您可以轻松地从各种数据源（如数据库、消息队列、日志文件等）将数据导入到Kafka集群中，也可以将数据从Kafka导出到其他目标系统中。

Kafka Connect的主要作用包括：

  • 简化数据集成：Kafka Connect提供了一种标准化的方式来连接和管理数据源和目标系统，使数据集成变得更加简单和统一。
  • 可扩展性和容错性：Kafka Connect可以水平扩展，支持分布式部署，并具有高可用性和容错性，以处理大规模的数据传输。
  • 可插拔的连接器：Kafka Connect采用了插件化的架构，可以轻松地添加新的连接器来支持不同的数据源和目标系统。

# 2、Kafka Connect的使用和配置

Kafka Connect的使用涉及以下几个重要的组件：

  • 连接器（Connectors）：连接器是Kafka Connect的核心概念，它负责定义数据源和目标系统之间的连接规则和配置。Kafka Connect提供了一些内置的连接器，同时也支持自定义连接器。
  • 任务（Tasks）：任务是连接器的实际工作单元。每个连接器可以有一个或多个任务，每个任务负责从数据源读取数据或向目标系统写入数据。
  • 转换器（Converters）：转换器用于在数据源和Kafka之间进行数据格式的转换。Kafka Connect提供了一些内置的转换器，如JSON转换器、Avro转换器等。

下面是一个使用Kafka Connect的示例代码（使用Java语言）：

import org.apache.kafka.connect.runtime.Connect;
import org.apache.kafka.connect.runtime.ConnectConfig;
import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
import org.apache.kafka.connect.runtime.standalone.StandaloneConfig;

import java.util.Properties;

public class KafkaConnectExample {

    public static void main(String[] args) {
        // 配置Kafka Connect
        Properties props = new Properties();
        props.put(ConnectConfig.CONNECTOR_CLASS_CONFIG, "com.example.MyConnector");
        props.put(ConnectConfig.TOPICS_CONFIG, "my-topic");
        // 其他配置参数...

        // 创建并启动Kafka Connect
        Connect connect = new Connect(new StandaloneConfig(props));
        connect.start();
    }
}

通过上述代码，您可以创建并配置一个Kafka Connect实例，并启动它。这个示例中使用了StandaloneConfig来启动一个独立的Kafka Connect节点，您还可以使用DistributedConfig来启动一个分布式的Kafka Connect集群。

# 3、Kafka Connect的源连接器和接收连接器

Kafka Connect提供了两种类型的连接器：源连接器和接收连接器。

  • 源连接器（Source Connectors）：源连接器负责将数据从外部系统导入到Kafka中。它会从数据源中读取数据，并将其写入Kafka的一个或多个主题中，使得数据可以在Kafka集群中进行处理和分发。常见的源连接器包括JDBC连接器、文件连接器等。

  • 接收连接器（Sink Connectors）：接收连接器负责将数据从Kafka导出到外部系统中。它会从一个或多个Kafka主题中读取数据，并将其写入目标系统中，如数据库、消息队列等。接收连接器可以实现将Kafka作为数据流的入口，从而实现实时数据分析、流式处理等场景。

示例代码中的com.example.MyConnector即为自定义连接器的类名，根据需要可以替换为其他连接器。

Kafka Connect的源连接器和接收连接器提供了丰富的功能和配置选项，您可以根据具体需求来选择合适的连接器，并根据文档进行配置和使用。

更多信息参考：KAFKA CONNECT (opens new window)

# 九、Kafka的监控和运维

Kafka的监控和运维是确保Kafka集群正常运行和高效工作的关键方面。本章将介绍Kafka的性能监控、日志管理以及故障排查和常见问题解决方法。

# 1、Kafka的性能监控

了解Kafka集群的性能状况对于及时发现问题和做出调整至关重要。下面介绍几种常用的Kafka性能监控方法。

# 1.1、 JMX监控

Kafka通过Java Management Extensions（JMX）提供了丰富的监控指标。您可以使用JConsole、JVisualVM或命令行工具如jcmd来连接到Kafka代理并查看关键指标，如吞吐量、存储大小、请求延迟等。

以下是使用Java代码通过JMX获取Kafka指标的示例：

import javax.management.*;
import java.lang.management.ManagementFactory;

public class KafkaJMXMonitor {
    public static void main(String[] args) throws Exception {
        MBeanServer server = ManagementFactory.getPlatformMBeanServer();
        
        ObjectName brokerMBean = new ObjectName("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec");
        Double messagesInPerSec = (Double) server.getAttribute(brokerMBean, "OneMinuteRate");
        
        System.out.println("Messages In Per Second: " + messagesInPerSec);
    }
}

# 1.2、 Kafka Exporter

Kafka Exporter是一种将Kafka指标暴露给Prometheus监控系统的工具。它允许您在Prometheus中配置和查询Kafka指标，并通过Grafana等工具进行可视化。

您可以通过以下步骤使用Kafka Exporter：

  1. 下载和配置Kafka Exporter。
  2. 将Kafka Exporter与Prometheus集成。
  3. 配置和启动Grafana以查看和分析Kafka指标。

# 2、Kafka的日志管理

Kafka的日志是消息持久化的关键部分，了解如何管理和维护Kafka的日志是运维工作中的重要任务。

# 2.1、 日志保留策略

Kafka允许您根据时间或日志大小来配置日志的保留策略。您可以通过配置文件中的log.retention.hours和log.retention.bytes参数来设置。

以下是示例配置，保留最近7天的日志和最大1TB的日志大小：

log.retention.hours=168
log.retention.bytes=1099511627776

# 2.2、 日志压缩

Kafka提供了多种压缩方式来减小存储占用和网络传输。您可以使用配置文件中的compression.type参数来设置压缩类型，如gzip、snappy或lz4。

以下是示例配置，启用Gzip压缩：

compression.type=gzip

# 十、Kafka的高级特性

Kafka作为一种高性能的消息队列系统，除了基本的功能之外，还提供了一些高级特性。本节将介绍Kafka的事务管理、安全性特性，并展望Kafka的未来发展趋势和应用场景。

# 1、Kafka的事务管理

Kafka引入了事务管理功能，使得在生产者和消费者之间实现原子性的消息处理成为可能。在Kafka中，事务由生产者发起并跨越多个写操作，确保这些操作要么全部成功，要么全部失败。这对于需要确保数据一致性的应用程序非常重要。

以下是使用Java客户端实现Kafka事务的示例代码：

// 创建Kafka生产者
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("transactional.id", "my-transactional-id");
Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());

// 初始化事务
producer.initTransactions();

try {
    // 开启事务
    producer.beginTransaction();

    // 发送消息
    producer.send(new ProducerRecord<>("my-topic", "key", "value"));

    // 执行其他操作（如数据库更新等）

    // 提交事务
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    // 处理异常情况
    producer.close();
} catch (KafkaException e) {
    // 中止事务
    producer.abortTransaction();
    producer.close();
}

使用事务管理功能，Kafka能够提供可靠的消息传递保证，并确保消息的完整性和一致性。

# 2、Kafka的安全性特性

随着数据的重要性不断增加，数据安全成为企业的首要关注点。Kafka提供了一系列的安全性特性来保护数据的机密性和完整性。

身份认证：Kafka支持基于SSL/TLS的身份认证，通过证书验证客户端和服务器的身份。

授权访问：Kafka提供了基于ACL（Access Control List）的授权机制，可以限制特定用户或组对主题的读写访问权限。

加密传输：Kafka支持消息的端到端加密传输，确保消息在传输过程中不会被窃取或篡改。

日志审计：Kafka的服务器端和客户端都能够记录详细的操作日志，便于跟踪和审计数据访问和操作。

通过这些安全性特性，Kafka可以满足企业对数据安全性的需求，并保护敏感数据免受未经授权的访问。

# 3、Kafka的未来发展趋势和应用场景

Kafka作为一种强大的消息队列系统，在未来有着广阔的发展前景和应用场景。

边缘计算：随着物联网和边缘计算的兴起，Kafka可以作为边缘设备与中心服务器之间的消息传递桥梁，实现实时的数据处理和分发。

流处理：Kafka与流处理框架（如Apache Flink和Apache Spark）的集成，使得实时数据流处理变得更加简单和高效。Kafka作为事件日志的中间存储，可以提供强大的事件溯源和流式处理能力。

机器学习：Kafka的高吞吐量和低延迟特性使其成为大规模机器学习训练和推理的理想平台。Kafka可以作为数据交换的中心，连接数据源和机器学习模型，并支持实时的模型推断和反馈。

可观测性：Kafka提供了丰富的监控指标和工具，可以对消息的生产、传递和消费进行全面的监控和分析。这对于运维团队来说是非常有价值的，能够实时检测和解决潜在的问题。

# 十一、总结

Kafka是一种高吞吐量、可扩展的分布式消息队列系统，在大规模实时数据处理和流式处理中发挥着重要的作用。

本文从Kafka的基本概念和架构开始，介绍了Kafka的基本功能和特性，包括生产者、消费者、主题、分区、副本、流、连机器、事务管理、安全管理等。

通过学习和应用Kafka，希望你能够构建一个高可靠性、高吞吐量的消息队列，满足当今数据处理领域的挑战和需求。

祝你变得更强!

编辑 (opens new window)
#Kafka
上次更新: 2025/04/01
Rabbitmq介绍与应用
ELK实践之采集并分析Nginx与应用程序日志

← ELK实践之采集并分析Nginx与应用程序日志→

最近更新
01
Spring Boot版本新特性
09-15
02
Spring框架版本新特性
09-01
03
Spring Boot开发初体验
08-15
更多文章>
Theme by Vdoing | Copyright © 2018-2025 京ICP备2021021832号-2 | MIT License