java怎么连接Kafka发送和消费消息教你在Java中实现Kafka消息通信-java教程-PHP中文网

java怎么连接Kafka发送和消费消息教你在Java中实现Kafka消息通信

畫卷琴夢

发布： 2025-11-01 13:00:02

原创

338人浏览过

使用Java连接Kafka需引入kafka-clients依赖，配置生产者序列化参数并发送消息，消费者设置组ID和反序列化器后订阅Topic拉取消息，通过poll循环处理，确保网络通畅与序列化匹配即可实现通信。

java怎么连接kafka发送和消费消息教你在java中实现kafka消息通信

用Java连接Kafka发送和消费消息，核心是使用Kafka客户端库，通过配置生产者和消费者来完成通信。整个过程不复杂，只要把依赖、配置和代码逻辑写对，很快就能跑通。

添加Maven依赖

项目基于Maven构建的话，先在pom.xml中引入Kafka客户端依赖。推荐使用官方提供的kafka-clients，通用性强，不绑定特定框架。

以下是常用依赖配置：

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

如果你用的是Spring Boot，也可以直接引入spring-kafka，它封装了底层操作，使用更方便。

立即学习“Java免费学习笔记（深入）”；

编写Kafka生产者发送消息

生产者负责把数据发到指定的Topic。创建KafkaProducer对象时，需要设置必要的参数。

基础步骤如下：

微信 WeLM

WeLM不是一个直接的对话机器人，而是一个补全用户输入信息的生成模型。

查看详情

定义Properties配置，包括Kafka服务器地址（bootstrap.servers）、key和value的序列化器（如StringSerializer）。
创建KafkaProducer<String, String>实例。
构造ProducerRecord，指定Topic名称、可选的Key和消息内容。
调用producer.send()发送消息，可以加回调处理发送结果。
程序结束前调用producer.close()释放资源。

示例代码片段：

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");

try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
    ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "key1", "Hello Kafka");
    producer.send(record, (metadata, exception) -> {
        if (exception == null) {
            System.out.printf("发送成功，分区：%d，偏移量：%d%n", metadata.partition(), metadata.offset());
        } else {
            exception.printStackTrace();
        }
    });
}

登录后复制

编写Kafka消费者接收消息

消费者从Topic订阅并拉取消息，需要配置组ID（group.id），以便Kafka管理消费进度（offset）。

基本流程是：

配置Properties，除了服务器地址和反序列化器，还要设置group.id和auto.offset.reset（如earliest从头开始，latest从最新开始）。
创建KafkaConsumer实例。
调用consumer.subscribe()订阅一个或多个Topic。
在循环中调用consumer.poll()获取消息，然后遍历处理。
处理完后提交offset（自动或手动），最后关闭消费者。

简单消费示例：

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");
props.put("auto.offset.reset", "earliest");

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(Arrays.asList("test-topic"));
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("收到消息：topic=%s, partition=%d, offset=%d, key=%s, value=%s%n",
                    record.topic(), record.partition(), record.offset(), record.key(), record.value());
        }
    }
}

登录后复制

基本上就这些。本地启动Kafka服务，创建好Topic，再运行生产和消费代码，就能看到消息通了。调试时注意看日志，网络和序列化问题最常见。

以上就是java怎么连接Kafka发送和消费消息教你在Java中实现Kafka消息通信的详细内容，更多请关注php中文网其它相关文章！