Kafka Client Java：方法、示例、安装

Apache Kafka Client for Java 简介

\n\n

Apache Kafka 是一个分布式事件流平台，能够实时发布、存储和处理数据流。Kafka Client for Java 库是官方客户端，提供了与 Kafka 代理交互的 API。它允许 Java 应用程序充当生产者（发送消息）和消费者（读取消息），并管理元数据和管理任务。

\n\n

该库在构建微服务架构、日志处理系统、ETL 管道、实时分析系统以及任何需要在组件之间进行可靠、可扩展且容错的数据传输的场景中不可或缺。

\n\n

安装 Kafka Client for Java

\n\n

Kafka Client for Java 通过 Maven Central Repository 分发。要将该库添加到项目中，需要在 pom.xml（用于 Maven）或 build.gradle（用于 Gradle）文件中指定依赖项。

\n\n

Maven

<dependency>\n    <groupId>org.apache.kafka</groupId>\n    <artifactId>kafka-clients</artifactId>\n    <version>3.6.0</version> <!-- 请使用最新版本 -->\n</dependency>

\n\n

Gradle

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

\n\n

添加依赖项并更新项目后，该库的所有类即可导入使用。

\n\n

快速入门：Producer 和 Consumer

\n\n

我们来看一个发送和接收消息的最小示例。

\n\n

Producer（发送者）

import org.apache.kafka.clients.producer.*;\nimport java.util.Properties;\n\npublic class SimpleProducer {\n    public static void main(String[] args) {\n        // 1. 配置属性\n        Properties props = new Properties();\n        props.put("bootstrap.servers", "localhost:9092");\n        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");\n        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");\n\n        // 2. 创建生产者\n        Producer<String, String> producer = new KafkaProducer<>(props);\n\n        // 3. 发送消息\n        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "Hello, Kafka!");\n        producer.send(record);\n\n        // 4. 关闭生产者\n        producer.close();\n    }\n}

\n\n

Consumer（接收者）

import org.apache.kafka.clients.consumer.*;\nimport java.time.Duration;\nimport java.util.Collections;\nimport java.util.Properties;\n\npublic class SimpleConsumer {\n    public static void main(String[] args) {\n        // 1. 配置属性\n        Properties props = new Properties();\n        props.put("bootstrap.servers", "localhost:9092");\n        props.put("group.id", "my-group");\n        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");\n        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");\n        props.put("auto.offset.reset", "earliest");\n\n        // 2. 创建消费者\n        Consumer<String, String> consumer = new KafkaConsumer<>(props);\n\n        // 3. 订阅主题\n        consumer.subscribe(Collections.singletonList("my-topic"));\n\n        // 4. 读取消息\n        while (true) {\n            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));\n            for (ConsumerRecord<String, String> record : records) {\n                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());\n            }\n        }\n    }\n}

\n\n