时间戳

首页 > 解决方案 > 无法从 KafkaProducer 发布消息

java - 无法从 KafkaProducer 发布消息

问题描述

我无法从 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

标签： javaapache-kafkakafka-producer-api

解决方案

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

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

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

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

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

推荐阅读