Kafka制作人Timeout异常：即将到期的1条记录

Question

我在用Kafka搭配Spring靴：

Kafka制作人班：

@Service
public class MyKafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private static Logger LOGGER = LoggerFactory.getLogger(NotificationDispatcherSender.class);

    // Send Message
    public void sendMessage(String topicName, String message) throws Exception {
        LOGGER.debug("========topic Name===== " + topicName + "=========message=======" + message);
        ListenableFuture<SendResult<String, String>> result = kafkaTemplate.send(topicName, message);
        result.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                LOGGER.debug("sent message='{}' with offset={}", message, result.getRecordMetadata().offset());
            }

            @Override
            public void onFailure(Throwable ex) {
                LOGGER.error(Constants.PRODUCER_MESSAGE_EXCEPTION.getValue() + " : " + ex.getMessage());
            }
        });
    }
}

Kafka配置：

spring.kafka.producer.retries=0
spring.kafka.producer.batch-size=100000
spring.kafka.producer.request.timeout.ms=30000
spring.kafka.producer.linger.ms=10
spring.kafka.producer.acks=0
spring.kafka.producer.buffer-memory=33554432
spring.kafka.producer.max.block.ms=5000
spring.kafka.bootstrap-servers=192.168.1.161:9092,192.168.1.162:9093

假设我在主题我的测试主题中发送了10次1000条消息。

10次中有8次我成功地在我的消费者中获取所有消息，但有时我会收到以下错误：

2017-10-05 07:24:11，[ERROR][my service-LoggingProducerListener-onError:76]发送key='null'和payload='{“deviceType”：“X”，“deviceKeys”：[{“apiKey”：“X-X-o”}]，“devices…”主题为我的测试主题时引发异常

和org。阿帕奇。Kafka。常见的错误。TimeoutException:自批创建加上延迟时间后，my-test-topic-4的1条记录因30024毫秒而过期




Answer 1

我通过正确处理主机spring解决了这个问题。Kafka。引导服务器及其DNS。即使网络解析IP地址，它似乎也需要DNS。

Answer 2

>

错误的第一条线索是30024 ms已通过-配置spring。Kafka。制作人要求暂停。ms=30000相关。这30秒的等待是为了填满生产者端的缓冲区。

当消息发布时，它会在生产者端得到缓冲，并等待30秒（见上面的配置）填满<代码>Spring。Kafka。制作人batch size=100000表示100KB，因此，如果消息接收负载较低，并且缓冲区在30秒内没有填充到100KB的更多消息，您可能会看到此消息。

spring.kafka.producer.linger.ms=10用于摄入负载较高且生产者希望限制对Kafka代理的发送（）调用。这是批处理准备就绪后（即缓冲区填充到100KB的批处理大小后）生产者在向代理发送消息之前等待的时间。

解决方案：

• 增加linger.ms以便在批处理就绪后保留消息更长时间。如果需要更多时间来填充批处理，请增加request.timeout.ms。
• 另一种方法：减少批处理大小，或增加request.timeout.ms，或两者兼而有之。

Answer 3

有三种可能性：

1. 增加request.timeout.ms-这是Kafka在缓冲区中等待整个批次准备就绪的时间。因此，在您的情况下，如果缓冲区中的消息少于100,000条，将发生超时。更多信息请参见：https://stackoverflow.com/a/34794261/2707179
2. 减少批处理大小-与上一点相关，它将更频繁地发送批处理，但它们将包含更少的消息。
3. 根据消息大小，您的网络可能无法赶上高负载？检查您的吞吐量是否不是瓶颈。