Kafka 分区和主题 - 深入

2025年1月23日 | 阅读 9 分钟

Kafka 简介

Apache Kafka 是一个分布式流处理平台，用于构建实时数据管道和流处理应用程序。它最初由 LinkedIn 开发，后来成为 Apache 软件基金会下的一个开源项目。Kafka 旨在处理海量数据，具有高吞吐量和低延迟的特点，使其适用于日志聚合、实时分析、事件采集等场景。

Kafka 架构的核心是主题（topic）的概念。本篇深入探讨将深入研究 Kafka 主题的细节、它们的结构、功能以及最佳使用实践。

Kafka 主题：核心概念

Kafka 主题 是生产者发送数据和消费者接收数据的逻辑通道。主题是 Kafka 数据组织和分发机制的基础。

Kafka 主题的特征

逻辑分组： 主题将相同类型或类别的消息（记录）进行逻辑分组。例如，一个名为 user-signups 的主题将包含与用户注册相关的消息。
可扩展性： 主题可以被分割成多个分区，以便在多个代理（broker）之间进行并行处理和分发。
解耦： 主题在生产者和消费者之间提供了解耦层。生产者无需知道谁将消费它们，即可将消息写入主题；消费者也可以独立地从主题读取消息。

主题的结构

一个主题被划分为多个分区，每个分区是一系列有序的消息。

分区： 每个分区可以包含一个或多个日志段（segment）。每个分区是一个仅追加（append-only）的日志，而段内的消息则按其偏移量（offset）排序。
偏移量（Offsets）： 分区中的每条消息都有一个唯一的偏移量，这是一个整数，用于标识消息在分区中的位置。

主题配置

Kafka 主题是高度可配置的，允许对性能、持久性和数据保留进行精细调整。一些关键配置包括：

副本因子（Replication Factor）： 指定了在不同代理之间保留的每个分区的副本数量。较高的副本因子可提高容错能力，但需要更多的存储空间。
保留策略（Retention Policy）： 决定了消息在主题中保留多长时间。Kafka 支持两种主要的保留策略：
- 基于时间的保留： 消息将保留指定的时长（例如，7 天）。
- 基于大小的保留： 消息将保留，直到主题达到指定的大小限制（例如，100 GB）。
清理策略（Cleanup Policy）： 定义了旧消息如何被丢弃或压缩。主要的清理策略有两种：
- 删除（Delete）： 默认策略，当旧消息超出保留期限或大小限制时，它们将被删除。
- 压缩（Compact）： 只保留每个键（key）的最新消息，这对于维护实体最新状态等场景非常有用。
段大小（Segment Size）： 控制分区内日志段的大小。较小的段更容易管理，但可能会增加开销。

示例：创建主题

您可以使用 kafka-topics.sh 脚本来创建 Kafka 主题：

kafka-topics.sh --create --topic user-signups --partitions 3 --replication-factor 2 --bootstrap-server localhost:9092

此命令创建一个名为 user-signups 的主题，包含 3 个分区和 2 的副本因子。

Kafka 主题中的分区

基于键的分区

Kafka 使用分区器（partitioner）来确定消息将被发送到哪个分区。默认情况下，Kafka 使用消息键的哈希值来决定分区。这确保了具有相同键的消息始终发送到同一分区，从而保持顺序。

示例：基于键的分区

考虑一个生产者发送带有用户 ID 作为键的消息：

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

def send_user_signup(user_id, user_data):
    producer.send('user-signups', key=user_id.encode('utf-8'), value=user_data)
    producer.flush()

# Send messages
send_user_signup('user1', {'action': 'signup', 'timestamp': '2024-08-07T10:00:00Z'})
send_user_signup('user2', {'action': 'signup', 'timestamp': '2024-08-07T10:01:00Z'})

在此示例中，具有相同 user_id 的消息将被发送到同一分区。

消费者和消费者组

消费者组（Consumer Groups）： 消费者被组织成消费者组。每个消费者组充当主题的单个逻辑订阅者。Kafka 确保每个分区在同一时间只被组内的一个消费者消费。

偏移量（Offsets）： 消费者使用偏移量来跟踪它们在分区中的位置。偏移量可以自动或手动提交。
示例：消费者代码

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'user-signups',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='user-signups-group',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

for message in consumer:
    user_id = message.key.decode('utf-8')
    user_data = message.value
    print(f"Received message for user {user_id}: {user_data}")

此消费者作为 user-signups-group 消费者组的一部分，从 user-signups 主题读取消息。

主题管理和监控

Kafka 提供了各种工具和指标来管理和监控主题。主要工具包括：

kafka-topics.sh： 一个用于创建、列出、描述和删除主题的命令行工具。
Kafka Manager： 一个用于管理 Kafka 集群的 Web 工具，包括主题管理和监控。
JMX 指标： Kafka 通过 JMX (Java Management Extensions) 暴露许多指标，用于监控主题和分区的健康状况和性能。

示例：描述主题

kafka-topics.sh --describe --topic user-signups --bootstrap-server localhost:9092

此命令提供有关 user-signups 主题的详细信息，包括分区详细信息、副本状态和配置。

高级主题功能

日志压缩

日志压缩（Log compaction）是一项功能，它只保留主题中每个键的最新值。这对于维护实体最新状态等场景非常有用。

示例：创建日志压缩主题

kafka-topics.sh --create --topic user-profiles --partitions 3 --replication-factor 2 --config cleanup.policy=compact --bootstrap-server localhost:9092

此命令创建一个名为 user-profiles 的主题，并启用了日志压缩。

主题删除

当主题不再需要时，Kafka 允许删除主题。这可以释放资源并简化管理。

示例：删除主题

此命令删除 user-signups 主题。

动态配置

Kafka 主题的配置可以动态更改，无需停机。这允许根据需要调整性能和保留策略。

示例：修改主题配置

kafka-configs.sh --alter --topic user-signups --add-config retention.ms=604800000 --bootstrap-server localhost:9092

此命令将 user-signups 主题的保留期设置为 7 天（604800000 毫秒）。

Kafka 主题的最佳实践

主题命名： 为主题使用描述性和一致的命名约定，以便轻松识别。
分区策略： 根据数据分发和处理需求选择合适的分区策略。使用基于键的分区来保证相关消息的顺序。
副本因子： 设置足够的副本因子来平衡容错能力和资源使用。
监控： 定期使用 Kafka 的指标和工具监控主题和分区，以确保最佳性能和可用性。

Kafka 分区

分区策略

在向 Kafka 主题生产消息时，分区策略决定了记录如何分发到各个分区。Kafka 根据用例提供了灵活的消息分区方式：

轮询（Round-Robin）： 如果未提供键，Kafka 会以轮询的方式将消息分发到分区，以确保负载均衡。
基于键的分区： 如果提供了键，Kafka 会使用键的哈希值来决定分区。这确保了具有相同键的所有消息都进入同一分区，从而保留该键的消息顺序。

示例：基于键的分区

考虑一个具有多个分区的 Kafka 主题 user-activity。我们可以使用代表用户 ID 的键来生产消息，确保特定用户的所有消息都被路由到同一分区。

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

topic = 'user-activity'

def send_message(key, value):
    producer.send(topic, key=key.encode('utf-8'), value=value)
    producer.flush()
    print(f"Sent message with key={key} value={value}")

messages = [
    ('user1', {'action': 'login', 'timestamp': '2024-08-07T10:00:00Z'}),
    ('user2', {'action': 'purchase', 'timestamp': '2024-08-07T10:01:00Z'}),
    ('user3', {'action': 'logout', 'timestamp': '2024-08-07T10:02:00Z'}),
]

for key, value in messages:
    send_message(key, value)

producer.close()

在此示例中，send_message 函数将消息作为键发送到 user-activity 主题。Kafka 将哈希键来决定分区，确保同一个用户的每个请求都进入同一个分区。

创建带分区的主题

您可以使用 kafka-topics.sh 脚本创建具有指定分区数的主题：

消费者与分区的交互

消费者从分区读取数据，Kafka 确保每个分区在同一时间只被一个消费者组中的一个消费者读取。这允许并行处理，同时保持每个分区内的消息顺序。

示例：消费者代码

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'user-activity',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='user-activity-group',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

for message in consumer:
    key = message.key.decode('utf-8')
    value = message.value
    print(f"Received message with key={key} value={value}")

此消费者作为 user-activity-group 消费者组的一部分，从 user-activity 主题读取消息。Kafka 将分区分配给组中的消费者，以确保负载均衡和容错。

分区的容错性

Kafka 通过复制来实现容错。每个分区可以有多个副本，分布在不同的代理上。一个副本是领导者（leader），负责处理所有读写操作；其他副本是跟随者（followers），负责复制领导者的数据。

如果领导者发生故障，其中一个跟随者将接任新的领导者，从而保证高可用性。

示例：配置副本

您可以在创建主题时配置主题的副本因子。更高的副本因子可提高容错能力，但需要更多的存储空间。

kafka-topics.sh --create --topic user-activity --partitions 3 --replication-factor 3 --bootstrap-server localhost:9092

此命令创建了一个 user-activity 主题，包含 3 个分区和 3 的副本因子，确保每个分区都有三个副本以实现容错。

高级分区技术

1. 自定义分区器

from kafka import KafkaProducer
import json

class CustomPartitioner:
    def __call__(self, key_bytes, all_partitions, available_partitions):
        key = key_bytes.decode('utf-8')
        if key == 'user1':
            return 0
        elif key == 'user2':
            return 1
        else:
            return 2

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    partitioner=CustomPartitioner()
)

def send_custom_partitioned_message(key, value):
    producer.send('user-activity', key=key.encode('utf-8'), value=value)
    producer.flush()
    print(f"Sent message with key={key} value={value}")

messages = [
    ('user1', {'action': 'login', 'timestamp': '2024-08-07T10:00:00Z'}),
    ('user2', {'action': 'purchase', 'timestamp': '2024-08-07T10:01:00Z'}),
    ('user3', {'action': 'logout', 'timestamp': '2024-08-07T10:02:00Z'}),
]

for key, value in messages:
    send_custom_partitioned_message(key, value)

producer.close()

在此示例中，CustomPartitioner 类根据键来确定每条消息的分区。

2. 重平衡（Rebalancing）

当消费者加入或离开消费者组时，Kafka 会执行重平衡以重新分配分区给消费者。这确保了负载的均匀分配和容错。重平衡是确保负载在消费者之间均匀分布并保持高可用性的重要机制。

监控和管理分区

Kafka 提供了各种工具和指标来监控和管理分区。这些工具有助于确保分区性能最佳，并能及时发现和解决任何问题。

kafka-topics.sh： 一个用于创建、列出、描述和删除主题和分区的命令行工具。

示例：描述主题

kafka-topics.sh --describe --topic user-activity --bootstrap-server localhost:9092

此命令提供有关 user-activity 主题的详细信息，包括分区详细信息和副本状态。

Kafka Manager： 一个用于管理 Kafka 集群的 Web 工具，包括监控分区。Kafka Manager 提供了一个用户友好的界面来管理和监控 Kafka 集群。
JMX 指标： Kafka 通过 JMX (Java Management Extensions) 暴露许多指标，用于监控分区的健康状况和性能。JMX 指标提供了对 Kafka 性能各个方面的洞察，例如吞吐量、延迟和资源利用率。

监控和管理 Apache Kafka 中的分区对于确保数据在 Kafka 代理之间的有效和均衡分发至关重要。下面是一个 Java 示例程序，演示了如何使用 Kafka AdminClient API 监控和管理 Kafka 分区。

前提条件

请确保您已运行 Kafka，并且项目中已包含必要的依赖项，例如 Maven 或 Gradle 项目中的 kafka-clients。

Maven 依赖项

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.5.0</version> <!-- Check for the latest version -->
</dependency>

用于监控和管理分区的 Java 程序

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.admin.NewPartitions;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaPartitionManager {

    private final AdminClient adminClient;

    public KafkaPartitionManager(String bootstrapServers) {
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        this.adminClient = AdminClient.create(properties);
    }

    public void listTopicsAndPartitions() throws ExecutionException, InterruptedException {
        ListTopicsResult topics = adminClient.listTopics();
        for (String topic : topics.names().get()) {
            TopicDescription description = adminClient.describeTopics(Collections.singletonList(topic)).all().get().get(topic);
            System.out.printf("Topic: %s, Partitions: %d%n", topic, description.partitions().size());
            description.partitions().forEach(partition -> {
                System.out.printf("  Partition: %d, Leader: %s, Replicas: %s, ISR: %s%n",
                        partition.partition(),
                        partition.leader().id(),
                        partition.replicas(),
                        partition.isr());
            });
        }
    }

    public void addPartitions(String topicName, int newPartitionCount) throws ExecutionException, InterruptedException {
        Map<String, NewPartitions> newPartitions = Collections.singletonMap(topicName, NewPartitions.increaseTo(newPartitionCount));
        adminClient.createPartitions(newPartitions).all().get();
        System.out.printf("Increased partitions for topic %s to %d%n", topicName, newPartitionCount);
    }

    public void close() {
        adminClient.close();
    }

    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092";
        KafkaPartitionManager manager = new KafkaPartitionManager(bootstrapServers);

        try {
            // List all topics and partitions
            manager.listTopicsAndPartitions();

            // Add partitions to a topic
            String topicName = "example-topic";
            int newPartitionCount = 5;
            manager.addPartitions(topicName, newPartitionCount);

            // List topics and partitions again to verify changes
            manager.listTopicsAndPartitions();

        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            manager.close();
        }
    }
}

说明

AdminClient 初始化： KafkaPartitionManager 类使用提供的 Kafka bootstrap servers 初始化 Kafka AdminClient。
列出主题和分区： listTopicsAndPartitions 方法列出 Kafka 集群中的所有主题，以及它们的分区、领导者、副本和同步副本 (ISR)。
添加分区： addPartitions 方法允许增加指定主题的分区数量。
主方法： main 方法演示了如何使用 KafkaPartitionManager 来列出分区、为主题添加更多分区，然后再次列出分区以验证更改。

下一主题Kafka-retention

Kafka 分区和主题 - 深入

Kafka 简介

Kafka 主题：核心概念

Kafka 主题的特征

主题的结构

主题配置

示例：创建主题

Kafka 主题中的分区

基于键的分区

消费者和消费者组

主题管理和监控

高级主题功能

日志压缩

示例：创建日志压缩主题

主题删除

动态配置

示例：修改主题配置

Kafka 主题的最佳实践

Kafka 分区

分区策略

示例：基于键的分区

创建带分区的主题

消费者与分区的交互

示例：消费者代码

分区的容错性

示例：配置副本

高级分区技术

监控和管理分区

前提条件

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Kafka 教程

Kafka 安装

Kafka CLI

Kafka 编程

实时示例

Kafka 监控

Kafka Connect

Kafka Streams

杂项

Kafka 分区和主题 - 深入

Kafka 简介

Kafka 主题：核心概念

Kafka 主题的特征

主题的结构

主题配置

示例：创建主题

Kafka 主题中的分区

基于键的分区

消费者和消费者组

主题管理和监控

高级主题功能

日志压缩

示例：创建日志压缩主题

主题删除

动态配置

示例：修改主题配置

Kafka 主题的最佳实践

Kafka 分区

分区策略

示例：基于键的分区

创建带分区的主题

消费者与分区的交互

示例：消费者代码

分区的容错性

示例：配置副本

高级分区技术

监控和管理分区

前提条件

相关帖子

Apache Kafka 幂等生产者

Kafka 消息压缩

Kafka 与区块链系统集成

Kafka 用于跨数据中心复制的挑战

使用 Kafka 进行零停机升级

Kafka 事件驱动架构

事件驱动架构：Kafka 到 Cosmos DB 集成

Kafka 消费者重新平衡

Kafka 超越发布-订阅：高级用法模式

Kafka 窗口化

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器