Kafka 黑科技大揭秘：主题创建、分区修改、生产者和消费者全解析

2023-09-10 11:02:22

掌控Kafka主题、分区与数据流：解锁数据处理的无限潜力

踏入Kafka的奇幻世界，你将开启一场数据处理的冒险之旅。在这个奇幻王国中，主题是数据的专属居所，分区是数据的存储空间，生产者和消费者是数据流动的使者。掌握这些元素，你将成为Kafka黑科技专家，在数据处理的战场上所向披靡。

主题：让数据找到专属的居所

主题是Kafka中数据的组织单位，就好比一座座独立的公寓楼，每一座公寓楼都容纳着特定的数据类型。你可以根据不同的数据特征创建不同的主题，让数据井然有序。

创建主题：

使用命令行：bin/kafka-topics.sh --create --topic [主题名] --partitions [分区数] --replication-factor [副本数]
使用Java API：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
AdminClient adminClient = AdminClient.create(props);
NewTopic newTopic = new NewTopic("my-topic", 3, (short) 2);
adminClient.createTopics(Arrays.asList(newTopic));

分区：满足数据扩容与优化需求

分区是主题的子集，就像公寓楼中的不同房间。它负责存储数据，并通过复制机制保证数据的安全性和可用性。你可以根据数据量和访问需求来调整分区数量。

修改分区数量：

使用命令行：bin/kafka-topics.sh --alter --topic [主题名] --partitions [分区数]
使用Java API：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
AdminClient adminClient = AdminClient.create(props);
NewPartitions newPartitions = NewPartitions.increaseTo(5);
adminClient.createPartitions(TopicPartitionInfo.of("my-topic", newPartitions));

生产者：将数据源源不断地注入Kafka

生产者是数据的搬运工，它们将数据源源不断地注入Kafka。你可以将生产者想象成管道工人，将数据从不同的来源输送到Kafka这个巨大的水库中。

发送数据：

使用命令行：bin/kafka-console-producer.sh --topic [主题名] --message "数据内容"
使用Java API：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<String, String>("my-topic", "数据内容"));

消费者：从Kafka中提取数据并处理

消费者是数据的接收器，它们从Kafka中提取数据并进行处理。你可以将消费者想象成管道工人的助手，将数据从Kafka的水库中取出，并送到不同的目的地。

接收数据：

使用命令行：bin/kafka-console-consumer.sh --topic [主题名] --from-beginning
使用Java API：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));
while (true) {
  ConsumerRecords<String, String> records = consumer.poll(100);
  for (ConsumerRecord<String, String> record : records) {
    System.out.println(record.value());
  }
}