Java连接Kafka发送消费消息指南

本文详细介绍了如何使用Java连接Kafka进行消息的发送和消费。通过引入`kafka-clients`依赖，并配置生产者端的序列化参数，开发者可以轻松地将消息发送到指定的Kafka Topic。同时，消费者端通过设置组ID和反序列化器，能够订阅Topic并拉取消息，利用`poll`方法循环处理接收到的消息。文章提供了详细的Maven依赖配置、生产者和消费者的代码示例，并强调了网络连接和序列化配置的重要性。掌握这些要点，即可快速实现Java与Kafka的通信，为构建高效的消息队列系统奠定基础。

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

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

添加Maven依赖

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

以下是常用依赖配置：

•   org.apache.kafka
•   kafka-clients
•   3.4.0

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

编写Kafka生产者发送消息

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

基础步骤如下：

• 定义Properties配置，包括Kafka服务器地址（bootstrap.servers）、key和value的序列化器（如StringSerializer）。
• 创建KafkaProducer实例。
• 构造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();
        }
    });
}</string></string>

编写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());
        }
    }
}</string></string></string>

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

今天关于《Java连接Kafka发送消费消息指南》的内容就介绍到这里了，是不是学起来一目了然！想要了解更多关于的内容请关注golang学习网公众号！