本地Kafka控制台正常，Java SDK发送消息超时问题求助

核心问题 #

你的Java生产者代码存在两个关键问题，导致消息发送超时或无法被接收：

1. 异步发送未等待完成
   Kafka生产者的send()方法是异步执行的，代码中调用producer.send(producerRecord)后没有等待发送结果，程序可能直接终止，生产者还没来得及完成消息发送就被销毁了。

2. 冗余的消费者配置
   生产者配置中添加了key.deserializer和value.deserializer属性，这是消费者专属配置，属于无效冗余项，虽然不会直接引发错误，但应该移除。

修复后的代码 #

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
properties.setProperty("key.serializer", StringSerializer.class.getName());
properties.setProperty("value.serializer", StringSerializer.class.getName());
// 移除不必要的消费者反序列化配置

KafkaProducer<String,String> producer = new KafkaProducer<>(properties);
ProducerRecord<String,String> producerRecord = new ProducerRecord<>("one","hello abhishek");

// 方式1：同步等待发送结果，捕获异常
try {
    producer.send(producerRecord).get(); // 调用get()阻塞等待发送完成
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
} finally {
    producer.close(); // 关闭生产者，确保所有缓存消息发送完毕
}

// 或者方式2：异步回调+关闭前刷新缓存
/*
producer.send(producerRecord, (metadata, exception) -> {
    if (exception != null) {
        exception.printStackTrace();
    } else {
        System.out.println("消息发送成功，偏移量：" + metadata.offset());
    }
});
producer.flush(); // 强制刷新缓存，确保消息发送到Broker
producer.close();
*/

额外配置检查（可选） #

如果修复后仍有超时问题，可以检查以下Kafka配置：

• acks：对于单分区无副本的Topic，acks=1（默认值）是合理的；如果设置为all会超时，因为没有副本节点可以确认。
• request.timeout.ms：如果本地环境网络较慢，可以适当调大这个值（默认30000ms）。

内容的提问来源于stack exchange，提问作者abhishek vashistha