在 Java 中使用 Apache Kafka：开发者完整指南

在 Java 中使用 Apache Kafka：从基础到高级应用

Apache Kafka 是一个分布式流数据处理平台，已成为构建事件驱动架构的事实标准。在本指南中，我们将探讨如何高效地在 Java 中使用 Kafka：从安装必要的库到编写带有错误处理的生产者和消费者。

为什么 Java 和 Kafka 是完美组合？

Java 是与 Kafka 配合使用的主要语言之一，原因如下：

• Kafka 官方客户端使用 Java 编写，提供最佳性能
• 丰富的库生态系统（Spring Kafka, Kafka Streams, Akka Streams）
• 强类型化，可通过配置对象进行精细调优

安装和添加依赖

要在 Java 中使用 Kafka，您需要在 pom.xml（Maven）或 build.gradle（Gradle）中添加依赖。

Maven

<dependency>    <groupId>org.apache.kafka</groupId>    <artifactId>kafka-clients</artifactId>    <version>3.6.0</version></dependency>

Gradle

implementation 'org.apache.kafka:kafka-clients:3.6.0'

创建 Kafka Producer（生产者）

生产者负责将消息发送到 Kafka 主题。以下是一个基本配置示例：

import org.apache.kafka.clients.producer.*;import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;

public class SimpleProducer {    public static void main(String[] args) {        // 1. 配置生产者属性        Properties props = new Properties();        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());                // 2. 创建生产者        KafkaProducer<String, String> producer = new KafkaProducer<>(props);                // 3. 发送消息        ProducerRecord<String, String> record = new ProducerRecord<>(            "my-topic",             "key-1",             "Hello, Kafka from Java!"        );                // 4. 异步发送并带回调        producer.send(record, (metadata, exception) -> {            if (exception == null) {                System.out.println("消息已发送: " +                     "topic=" + metadata.topic() +                     ", partition=" + metadata.partition() +                     ", offset=" + metadata.offset());            } else {                System.err.println("发送错误: " + exception.getMessage());            }        });                // 5. 关闭生产者        producer.close();    }}

创建 Kafka Consumer（消费者）

消费者从主题中读取消息。正确配置消费者组对于负载均衡非常重要。

import org.apache.kafka.clients.consumer.*;import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;import java.util.Collections;import java.util.Properties;

public class SimpleConsumer {    public static void main(String[] args) {        // 1. 配置消费者属性        Properties props = new Properties();        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());        props.put(Consu