Kafka 主题、分区、副本入门：通往可靠消息传递的路径

2023-11-25 11:04:22

Kafka 主题：消息传递的骨干

简介

在现代数据处理领域，可靠且高效的消息传递系统至关重要。Kafka 是一个强大的分布式流媒体平台，为消息传递提供了坚实的基础。其核心概念之一是主题，它为消息流提供了组织和结构。

主题

一个 Kafka 主题可以被视为一个虚拟的信箱，其中存储着相关的消息。生产者向主题发送消息，消费者从主题中接收消息。这种架构类似于电子邮件系统，其中发送方将邮件发送到收件箱，而接收方从中检索邮件。

分区

为了提高吞吐量和可扩展性，Kafka 主题可以划分为多个分区。每个分区都是一个独立的存储单元，可以容纳一定数量的消息。将主题划分为多个分区可以将吞吐量提升到原来的好几倍。例如，如果一个主题每天产生数百万条消息，将其划分为 10 个分区可以将吞吐量提高 10 倍。

副本

为了确保数据的安全和高可用性，Kafka 主题的每个分区都可以在不同的服务器上存储多个副本。如果某个分区出现故障，副本可以保证数据的安全和可用性。用户可以为每个主题配置副本的数量，副本越多，数据就越安全，但系统开销也就越大。

可靠、可扩展且高可用

Kafka 主题、分区和副本是构建可靠、可扩展且高可用消息传递系统的三大支柱。通过理解这些概念并正确使用它们，您可以构建出满足您需求的强大消息传递系统。

技术实践：构建您的第一个 Kafka 主题

创建主题

使用 Kafka 命令行工具或 API 创建一个新的主题。例如，以下命令创建一个名为 "my-topic" 的主题，具有 3 个分区和 2 个副本：

bin/kafka-topics.sh --create --topic my-topic --partitions 3 --replication-factor 2

发送消息

使用生产者 API 将消息发送到主题。以下示例代码展示了如何使用 Java 发送消息：

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 KafkaProducerExample {

  public static void main(String[] args) {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

    KafkaProducer<String, String> producer = new KafkaProducer<>(props);

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

    producer.send(record);

    producer.close();
  }
}

消费消息

使用消费者 API 从主题中消费消息。以下示例代码展示了如何使用 Java 消费消息：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

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

public class KafkaConsumerExample {

  public static void main(String[] args) {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

    KafkaConsumer<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.println(record.key() + ": " + record.value());
      }
    }

    consumer.close();
  }
}