Kafka 代理 - 详细信息

2025年5月14日 | 阅读 13 分钟

Kafka 简介

Apache Kafka 旨在高效处理实时数据流，提供高性能、高可靠性以及可扩展性。它作为一个分布式消息系统，数据由生产者发送，由消费者接收。Kafka 能够处理海量消息，使其非常适合各种应用，例如日志收集、流处理和实时分析。

什么是 Kafka Broker？

A Kafka broker 是 Apache Kafka 系统的一个关键组成部分。它充当生产者（发送消息）和消费者（接收消息）之间的中间人。Broker 管理数据存储，通过复制确保数据的可用性，并处理数据检索请求。

在 Kafka 集群中，多个 Broker 协同工作，提供可靠性、可伸缩性和有效的数据管理。每个 Broker 都有一个唯一的 ID，并负责 Kafka 主题的一个或多个分区。Kafka Broker 对于保持 Kafka 集群平稳高效运行至关重要。

Kafka 架构中的 Kafka Broker

在 Kafka 集群中，其架构由以下主要组件构成：

生产者 (Producers): 生产者将消息发布到 Kafka 主题。它们可以向集群中的一个或多个 Broker 发送数据，然后 Broker 将数据存储在分区中。
消费者 (Consumers): 消费者订阅 Kafka 主题以接收消息。它们从一个或多个 Broker 获取数据并按需处理。
Kafka Broker: Kafka Broker 是一个服务器，它接收来自生产者的消息，将其存储在相应的分区中，并将其传递给消费者。Broker 还负责数据复制，并确保系统的可靠性。
ZooKeeper: 尽管 Kafka 现在支持无 ZooKeeper 模式（KRaft 模式），但在传统的 Kafka 设置中，ZooKeeper 用于管理和协调 Kafka Broker，处理领导者选举和配置管理等任务。
分区和主题 (Partitions and Topics): 在 Kafka 中，一个主题被划分为多个分区。每个分区包含一个有序的记录列表，并由一个或多个 Broker 管理。领导者 Broker 负责分区的全部读写操作，而跟随者 Broker 则复制数据以确保可靠性。

Kafka Broker 的作用

让我们通过一个示例程序来探讨 Kafka Broker 的作用，该程序中，一个生产者向 Kafka 主题发送消息，一个消费者消费这些消息。Broker 负责处理生产者和消费者之间的通信，确保消息得到正确存储和提供。

设置 Kafka 和 ZooKeeper

在运行生产者和消费者程序之前，我们需要设置一个 Kafka Broker 和 ZooKeeper（如果未使用 KRaft 模式）。下载并解压 Kafka：

wget https://downloads.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz
tar -xzf kafka_2.13-3.0.0.tgz
cd kafka_2.13-3.0.0

1. 启动 ZooKeeper（如果使用 ZooKeeper 模式）

2. 启动 Kafka Broker

生产者程序

生产者程序将消息发送到 Kafka 主题。Broker 处理这些消息，将其存储在正确的分区中，并确保它们可供消费。

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

import java.util.Properties;

public class SimpleProducer {
    public static void main(String[] args) {
        // Set up properties for the Kafka producer
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");  // Kafka broker address
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // Create a Kafka producer
        Producer<String, String> producer = new KafkaProducer<>(props);

        // Send messages to the Kafka topic "test-topic"
        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<>("test-topic", Integer.toString(i), "Message " + i));
        }

        // Close the producer
        producer.close();
    }
}

输出

在此程序中

生产者将消息发送到 localhost:9092 的 Kafka Broker。
Broker 接收这些消息并将它们存储在主题 test-topic 的适当分区中。

消费者程序

消费者程序订阅同一个 Kafka 主题，并消费 Broker 存储的消息。

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

import java.util.Collections;
import java.util.Properties;

public class SimpleConsumer {
    public static void main(String[] args) {
        // Set up properties for the Kafka consumer
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");  // Kafka broker address
        props.put("group.id", "test-group");  // Consumer group ID
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // Create a Kafka consumer
        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        
        // Subscribe to the topic "test-topic"
        consumer.subscribe(Collections.singletonList("test-topic"));

        // Continuously poll for new messages
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Offset: %d, Key: %s, Value: %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
}

输出

在此程序中

消费者连接到 localhost:9092 的 Kafka Broker。
它订阅 test-topic 主题并不断轮询新消息。
Broker 将主题中存储的消息提供给消费者。

Kafka Broker 如何处理通信

生产者到 Broker: 生产者将消息发送到 Broker，然后 Broker 将其存储在主题的适当分区中。Broker 确保数据安全地保存到磁盘以防止丢失。
Broker 到消费者: 消费者向 Broker 请求消息。Broker 根据消费者的偏移量提供请求的消息，确保消费者获得正确的消息序列。
数据复制: 如果主题的复制因子大于 1，Broker 会确保数据跨多个 Broker 进行复制。这种复制提供了可靠性，因此如果一个 Broker 失败，数据仍然可以从副本中访问。
领导者和跟随者角色: 对于每个分区，领导者 Broker 负责处理所有读写请求。跟随者 Broker 复制领导者的数据以保持高可用性。如果领导者失败，其中一个跟随者将自动升级为新的领导者。
消息存储和检索: Broker 将消息存储在日志段中，这些日志段是磁盘上的文件。每条消息都会获得一个偏移量，该偏移量有助于消费者跟踪其进度。
Broker 负责从日志段检索消息并将其提供给消费者。

Kafka Broker 的关键职责

Kafka Broker 是 Kafka 集群的核心，在管理生产者和消费者之间的数据流方面发挥着至关重要的作用。以下是 Kafka Broker 的关键职责，以及说明这些职责的示例程序：

消息存储和检索
分区和复制管理
领导者选举
容错和高可用性
处理消费者偏移量
负载均衡和可伸缩性
监控和指标
安全和访问控制

让我们通过解释和相应的代码示例来深入探讨这些职责。

1. 消息存储和检索

职责: Kafka Broker 将生产者传入的消息存储在分区中，并在请求时提供给消费者。每条消息都有一个偏移量，使其在分区内唯一。

程序示例: 以下是一个简单的生产者-消费者示例，演示了 Kafka Broker 如何存储和检索消息。

生产者程序（存储消息）

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

import java.util.Properties;

public class MessageProducer {
    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);

        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<>("example-topic", Integer.toString(i), "Message " + i));
        }

        producer.close();
    }
}

消费者程序（检索消息）

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

import java.util.Collections;
import java.util.Properties;

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

        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("example-topic"));

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

2. 分区和复制管理

职责: Broker 负责管理主题的分区。它还负责在不同 Broker 之间复制这些分区，以确保数据冗余和容错。

程序示例: 此程序演示了 Kafka Broker 如何管理分区和复制。

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;

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

public class TopicManagement {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        AdminClient adminClient = AdminClient.create(props);

        NewTopic newTopic = new NewTopic("replicated-topic", 3, (short) 2); // 3 partitions, replication factor of 2
        adminClient.createTopics(Collections.singletonList(newTopic)).all().get();

        adminClient.close();
    }
}

3. 领导者选举

职责: 对于每个分区，将选举一个 Broker 作为领导者，负责处理所有读写请求。其他 Broker 是跟随者，它们复制领导者的数据。如果领导者 Broker 出现故障，将自动选出一个新的 Broker 来担任新的领导者。

程序示例: 此示例模拟了 Broker 故障触发领导者选举的场景。但是，选举过程是 Kafka 内部的，无法通过客户端代码直接控制。以下管理客户端代码会检查分区的当前领导者。

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.TopicDescription;

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

public class LeaderElectionCheck {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        AdminClient adminClient = AdminClient.create(props);

        DescribeTopicsResult result = adminClient.describeTopics(Collections.singletonList("replicated-topic"));
        TopicDescription description = result.all().get().get("replicated-topic");

        description.partitions().forEach(partitionInfo ->
                System.out.printf("Partition: %d, Leader: %d%n", partitionInfo.partition(), partitionInfo.leader().id()));

        adminClient.close();
    }
}

4. 容错和高可用性

职责: Kafka Broker 通过将分区复制到不同的 Broker 来提供高可用性和可靠性。如果一个 Broker 出现故障，其分区可以从其他 Broker 上的副本中恢复。

程序示例: 通过停止一个 Broker 并观察分区领导者的重新分配来演示 Kafka 的容错能力。

# Stop a broker (assuming broker ID 1)
bin/kafka-server-stop.sh config/server-1.properties

# Check partition leadership reassignment
bin/kafka-topics.sh --describe --topic replicated-topic --bootstrap-server localhost:9092

5. 处理消费者偏移量

职责: Kafka Broker 管理每个消费者组的偏移量，这些偏移量跟踪消费者在主题消息中的进度。这使得消费者在发生故障时可以从中断处继续。

程序示例: 此示例显示了一个提交偏移量的消费者，允许 Broker 管理消费者在主题中的位置。

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

import java.util.Collections;
import java.util.Properties;

public class OffsetManagementConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "offset-group");
        props.put("enable.auto.commit", "false"); // Disable auto-commit
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("example-topic"));

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

            // Manually commit offsets
            consumer.commitSync();
        }
    }
}

6. 负载均衡和可伸缩性

职责: Kafka Broker 通过将分区分布到多个 Broker 来处理负载均衡。这可以防止任何单个 Broker 成为瓶颈，并通过添加更多 Broker 来实现系统的扩展。

程序示例: 这更多地涉及 Kafka 的内部机制，但您可以使用管理客户端观察分区分布。

bin/kafka-topics.sh --describe --topic example-topic --bootstrap-server localhost:9092

7. 监控和指标

职责: Kafka Broker 提供一系列可供跟踪的指标，以确保集群保持健康并运行良好。这些指标包括消息速率、延迟、分区状态等。

程序示例: 您可以使用 JMX 或 Prometheus 等工具来监控 Kafka 指标。以下是如何为 Kafka Broker 启用 JMX：

export JMX_PORT=9999
bin/kafka-server-start.sh config/server.properties &

8. 安全和访问控制

职责: Kafka Broker 强制执行安全策略，包括身份验证、授权和加密，以确保只有授权的客户端才能生产和消费消息。

程序示例: 此示例显示了如何设置一个带有 SSL 加密的 Broker。

# Kafka broker SSL configuration
ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
ssl.keystore.password=your_password
ssl.key.password=your_key_password
ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
ssl.truststore.password=your_truststore_password

Broker 配置和参数

Kafka Broker 是高度可配置的，允许管理员根据特定需求微调其操作。一些关键配置参数包括：

broker.id: 集群中每个 Broker 的唯一标识符。
log.dirs: Kafka 存储日志数据的目录。
num.network.threads: 处理网络请求的线程数。
num.io.threads: 处理 I/O 操作的线程数。
num.replica.fetchers: 用于获取副本的线程数。
socket.send.buffer.bytes: 通过网络发送数据的缓冲区大小。
auto.create.topics.enable: 启用自动主题创建。

Kafka Broker 操作

Kafka Broker 是 Kafka 分布式系统架构中的关键组件，负责管理生产者和消费者之间的数据流，处理元数据，并确保系统的可靠性和可用性。理解 Kafka Broker 执行的关键操作对于高效管理 Kafka 集群至关重要。

关键 Kafka Broker 操作

主题的创建和删除
分区重新分配
领导者选举
数据复制
日志清理
监控和管理

让我们通过演示这些操作的示例程序或命令来详细探讨每个操作。

1. 主题的创建和删除

操作: Kafka Broker 管理主题的创建和删除，主题是 Kafka 中数据的主要组织单位。

2. 分区重新分配

操作: Kafka Broker 处理跨 Broker 的分区重新分配，这在进行负载均衡或向集群添加/删除 Broker 时可能需要。

程序示例

重新分配分区

1. 生成重新分配计划

bin/kafka-reassign-partitions.sh --generate --zookeeper localhost:2181 --topics-to-move-json-file topics-to-move.json --broker-list "1,2,3"

2. 执行重新分配计划

bin/kafka-reassign-partitions.sh --execute --zookeeper localhost:2181 --reassignment-json-file reassignment.json

重新分配示例 JSON 文件（topics-to-move.json）

{
  "version": 1,
  "partitions": [
    {"topic": "my-topic", "partition": 0, "replicas": [2, 3]},
    {"topic": "my-topic", "partition": 1, "replicas": [1, 2]}
  ]
}

3. 领导者选举

操作: Kafka Broker 参与领导者选举过程，确定哪个 Broker 负责处理分区的读写请求。

程序示例

领导者选举由 Kafka 自动处理。但是，您可以使用上述分区重新分配工具来强制进行重新分配（包括领导者选举）。

检查分区的当前领导者

bin/kafka-topics.sh --describe --topic my-topic --bootstrap-server localhost:9092

4. 数据复制

程序示例

复制通常在创建主题时通过设置复制因子来配置。Broker 会自动处理复制。

您可以使用以下命令监视复制状态：

bin/kafka-topics.sh --describe --topic my-topic --bootstrap-server localhost:9092

此命令将显示 ISR（In-Sync Replicas），指示哪些副本与领导者同步。

5. 日志清理

操作: Kafka Broker 执行日志清理以管理磁盘空间和维护数据完整性，通过删除旧的日志段或压缩它们。

程序示例

配置日志清理

在 server.properties 文件中：

log.retention.hours=168
log.cleanup.policy=delete
log.segment.bytes=1073741824

手动触发日志清理: Kafka 没有提供直接的手动日志清理命令。但是，调整配置并重启 Broker 将触发清理过程。

6. 监控和管理

程序示例

启用 JMX 监控

使用以下命令启动启用了 JMX 监控的 Kafka Broker：

export JMX_PORT=9999
bin/kafka-server-start.sh config/server.properties &

然后，您可以使用 JConsole、Prometheus 或 Grafana 等工具来监控 Kafka 指标，例如消息吞吐量、复制延迟和磁盘使用情况。

示例：使用 Prometheus 监控 Kafka Broker 指标

1. 安装 Prometheus JMX Exporter: 下载 JMX exporter jar 文件并将其放在 Kafka 的 libs 目录中。

2. 配置 Kafka Broker 使用 JMX Exporter: 更新您的 Kafka Broker 启动命令以包含 JMX exporter。

export KAFKA_OPTS="-javaagent:/path/to/jmx_prometheus_javaagent-0.14.0.jar=7071:/path/to/config.yaml"
bin/kafka-server-start.sh config/server.properties &

3. Prometheus 配置: 将 Kafka Broker 的 JMX exporter 端点添加到您的 Prometheus 配置中。

scrape_configs:
  - job_name: 'kafka'
    static_configs:
      - targets: ['localhost:7071']

4. 查看指标: 启动 Prometheus 并查看 Prometheus UI 中的 Kafka Broker 指标。

Kafka Broker 示例程序

设置多 Broker 的 Kafka 集群

步骤 1：安装 Kafka

首先，下载并解压 Kafka 二进制文件。

wget https://downloads.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz
tar -xzf kafka_2.13-3.0.0.tgz
cd kafka_2.13-3.0.0

步骤 2：配置 Broker

编辑每个 Broker 的配置文件，分配唯一的 broker.id 并指定不同的 log.dirs 路径。

# config/server.properties for broker 1
broker.id=1
log.dirs=/tmp/kafka-logs-1
port=9092

# config/server.properties for broker 2
broker.id=2
log.dirs=/tmp/kafka-logs-2
port=9093

步骤 3：启动 Broker

使用以下命令启动每个 Broker：

对每个 Broker 配置重复此命令。

步骤 4：创建主题

在运行多个 Broker 后，创建一个具有复制因子的主题：

bin/kafka-topics.sh --create --topic my-topic --partitions 3 --replication-factor 2 --bootstrap-server localhost:9092

简单的生产者-消费者示例

生产者程序

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

import java.util.Properties;

public class SimpleProducer {
    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);

        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<>("my-topic", Integer.toString(i), "Message " + i));
        }

        producer.close();
    }
}

输出

消费者程序

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

import java.util.Collections;
import java.util.Properties;

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

        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

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

输出

Broker 故障转移和恢复示例

此示例演示了 Kafka 如何自动从 Broker 故障中恢复。关键步骤包括：

1. 停止一个 Broker: 使用以下命令停止集群中的一个 Broker：

2. 验证故障转移: 使用 Kafka 日志或命令行工具来验证已为主机上分区的 Broker 选举出新的领导者。

3. 重启 Broker: 重新启动 Broker，观察它如何重新加入集群并恢复其角色。

高级 Kafka Broker 配置

Broker 机架感知

Kafka 可以配置为确保分区的副本分布在不同的机架或数据中心。这通过减少机架或数据中心故障的影响来提高容错能力。

日志保留策略

使用 log.retention.hours 参数控制 Kafka 保留日志的时间：

监控和管理 Kafka Broker

Kafka Broker 监控工具

有几种工具和方法可以监控 Kafka Broker：

JMX (Java Management Extensions): Kafka 通过 JMX 公开指标，这些指标可以使用 Prometheus 等工具收集。
Kafka Manager: 一个免费工具，用于管理和监控 Kafka 集群。
Confluent Control Center: 一个付费工具，提供 Kafka 的高级监控和管理选项。

性能调优和优化

优化网络设置: 调整 socket.send.buffer.bytes 和 socket.receive.buffer.bytes 等参数以优化网络性能。
调整日志段大小: 配置 log.segment.bytes 以控制段大小，平衡 I/O 性能和存储效率。
调整复制: 确保 num.replica.fetchers 的设置与您集群中的复制负载相匹配。

处理 Broker 故障

Kafka 设计用于优雅地处理 Broker 故障：

ISR (In-Sync Replica) 机制: 确保在当前领导者失败时，跟随者可以接替成为领导者。
分区重新分配: 如果一个 Broker 发生故障且未恢复，Kafka 可以将受影响的分区重新分配给其他 Broker。
数据恢复: 当故障的 Broker 重新启动时，它会自动与领导者同步以恢复丢失的数据。

示例程序脚本

以下是一个自动化上述步骤的脚本。将其保存为 handle_broker_failure.sh 并执行。

#!/bin/bash

# Start Kafka brokers
echo "Starting Kafka brokers..."
bin/kafka-server-start.sh config/server-1.properties &
bin/kafka-server-start.sh config/server-2.properties &
bin/kafka-server-start.sh config/server-3.properties &

# Create a topic with replication
echo "Creating topic 'my-topic'..."
bin/kafka-topics.sh --create --topic my-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 3

# Produce messages to the topic
echo "Producing messages to 'my-topic'..."
echo -e "message 1\nmessage 2\nmessage 3" | bin/kafka-console-producer.sh --topic my-topic --bootstrap-server localhost:9092

# Simulate a broker failure
echo "Stopping Broker 1..."
bin/kafka-server-stop.sh config/server-1.properties

# Check partition leadership
echo "Checking partition leadership..."
bin/kafka-topics.sh --describe --topic my-topic --bootstrap-server localhost:9092

# Restart the failed broker
echo "Restarting Broker 1..."
bin/kafka-server-start.sh config/server-1.properties &

# Verify recovery
echo "Verifying recovery..."
bin/kafka-topics.sh --describe --topic my-topic --bootstrap-server localhost:9092

echo "Broker failure handling complete."

身份验证和授权

Kafka 支持各种身份验证机制，包括：

SASL (Simple Authentication and Security Layer): 用于使用 Kafka Broker 对客户端进行身份验证。
OAuth: Kafka 可以与 OAuth 集成以进行基于令牌的身份验证。

Kafka 还具有授权机制，该机制控制哪些客户端（生产者或消费者）可以访问特定主题。

加密和 SSL 配置

要配置 Kafka Broker 和客户端之间的 SSL 加密，请使用必要的 SSL 设置更新 Broker 配置：

ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
ssl.keystore.password=your_password
ssl.key.password=your_key_password
ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
ssl.truststore.password=your_truststore_password

下一个主题Kafka-hadoop-loader

← 上一个下一个 →

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 Broker？

Kafka 架构中的 Kafka Broker

Kafka Broker 的作用

设置 Kafka 和 ZooKeeper

生产者程序

消费者程序

Kafka Broker 如何处理通信

Kafka Broker 的关键职责

1. 消息存储和检索

2. 分区和复制管理

3. 领导者选举

4. 容错和高可用性

5. 处理消费者偏移量

6. 负载均衡和可伸缩性

7. 监控和指标

8. 安全和访问控制

Broker 配置和参数

Kafka Broker 操作

1. 主题的创建和删除

2. 分区重新分配

3. 领导者选举

4. 数据复制

5. 日志清理

6. 监控和管理

Kafka Broker 示例程序

设置多 Broker 的 Kafka 集群

简单的生产者-消费者示例

生产者程序

消费者程序

Broker 故障转移和恢复示例

高级 Kafka Broker 配置

Broker 机架感知

日志保留策略

监控和管理 Kafka Broker

Kafka Broker 监控工具

性能调优和优化

处理 Broker 故障

示例程序脚本

身份验证和授权

加密和 SSL 配置

相关帖子

构建从 Kafka 到 Cosmos DB 的复制管道

事件流架构的用例

Cosmos DB 作为 Apache Kafka 主题的接收器

Kafka 连接器

使用 Kafka 在零售业中实现超个性化

Kafka Streams 的时间旅行查询

Kafka 重新平衡

Kafka 事件流