Kafka生产者无法发布消息且应用卡住、出现HTTP请求头解析错误如何解决

• 核心配置错误：你在KafkaConstants.HOST中配置的Kafka服务地址错误，从日志可以看到生产者尝试连接localhost:8084，该端口是你本地SpringBoot服务的Tomcat端口，不是Kafka服务的监听端口。
• 报错原因：Kafka客户端使用原生TCP协议向Tomcat端口发送请求，Tomcat将非HTTP格式的请求当成HTTP报文解析，因此抛出「HTTP方法名包含非法字符」的异常。
• 应用卡住原因：Kafka生产者默认开启无限重试机制，且你没有配置请求超时时间，发送消息的异步异常不会被主线程的catch捕获，后台线程会一直尝试连接错误的地址，导致整体服务阻塞。

1. 修正Kafka服务地址配置 #

找到常量类KafkaConstants中HOST的定义，修改为正确的Kafka服务地址，Kafka默认监听端口为9092，本地部署的话配置示例如下：

// 示例：本地Kafka服务地址
public static final String HOST = "localhost:9092";

2. 优化生产者配置避免服务阻塞 #

在KafkaProducerConfig的producerFactory方法中添加超时、重试限制配置，防止异常场景下服务长时间卡住：

@Bean
public ProducerFactory<String, Book> producerFactory(){
    Map<String,Object> configProps = new HashMap<>();
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstants.HOST);
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    // 新增配置：请求超时时间3秒
    configProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 3000);
    // 新增配置：最大重试次数3次
    configProps.put(ProducerConfig.RETRIES_CONFIG, 3);
    // 新增配置：元数据获取超时时间2秒
    configProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 2000);
    return new DefaultKafkaProducerFactory<>(configProps);
}

3. 优化消息发送逻辑捕获异常 #

默认kafkaTemplate.send是异步发送，异常仅在后台线程抛出，主线程无法捕获，可改为带超时的同步等待，确保异常可以被正常捕获：

@Override
public String sendBooksDetails(List<Book> booksList) {
    String msg = "";
    try {
        if (!booksList.isEmpty()){
            for (Book book:booksList) {
                // 增加超时等待，2秒未返回则抛出超时异常
                kafkaTemplate.send(KafkaConstants.TOPIC,book).get(2, TimeUnit.SECONDS);
            }
        }
        msg = "Books are added to kafka topic successfully";
    }catch (Exception e){
        e.printStackTrace();
        msg = "Unable to publish message";
    }
    return msg;
}

内容的提问来源于stack exchange，提问作者Ram Rikhi