java - 无法从 KafkaProducer 发布消息

Question

我无法从 Kafka Producer 发送消息。我的配置不起作用，如下所示:

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());
properties.setProperty("acks", "1");
properties.setProperty("retries", "3");
properties.setProperty("linger.ms", "1");

Producer<String, String> producer =
    new org.apache.kafka.clients.producer.KafkaProducer<String, String>(properties);
ProducerRecord<String, String> producerRecord =
    new ProducerRecord<String, String>("second_topic", "3", "messagtest");
Future<RecordMetadata> s = producer.send(producerRecord);

producer.flush();
producer.close();

我执行 s.get() 后出现错误；

 java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for second_topic-0: 30021 ms has passed since batch creation plus linger time
at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.valueOrError(FutureRecordMetadata.java:94)
at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:64)
at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:29)
at ai.sys.producer.Test.main(Test.java:33)
Caused by: org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for second_topic-0: 30021 ms has passed since batch creation plus linger time

Answer 1

Kafka Producer 中默认启用批处理，大小为 16K 字节。但是，在您的代码中，您仅发送一条记录，这可能不满足批量大小。

因此，为了让您的代码正常工作，请尝试将以下“batch.size”属性添加到 Kafka Producer 属性中，并将其值设置为“0”。

properties.setProperty("batch.size", "0");

这将禁用批处理机制并允许您的生产者将记录写入 Kafka Broker。

注意:实时情况下，禁用批处理会增加对代理的写入请求数量，并降低生产者和服务器的 I/O 吞吐量和性能。