Java -jar运行Kafka生产者无法发消息，mvn/spring-boot:run正常

数据老陈

2026-7-1

问题描述

提供的Kafka生产者代码如下：

@Bean
public KafkaTemplate<String, byte[]> kafkaTemplate() {
    return new KafkaTemplate<>(producerFactory());
}

@Bean
public ProducerFactory<String, byte[]> producerFactory() {
   
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    configProps.put("spring.kafka.bootstrap-servers", "127.0.0.1:9092");
    configProps.put("spring.kafka.consumer.auto-offset-reset","earliest");
    configProps.put("spring.kafka.consumer.group-id","data-group");
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
   
    configProps.put(ConsumerConfig.CLIENT_ID_CONFIG,"ClientIDdata-group");
    return new DefaultKafkaProducerFactory<>(configProps);
}

ListenableFuture<SendResult<String, byte[] >> futureResult = kafkaTemplate.send(topic, binarydata);
futureResult.addCallback(PublishCallback());

使用mvn spring-boot:run运行时生产者可正常发送消息，但打包成Jar后用java -jar运行无法发送，且无报错日志。已确认kafka-client 2.7.1依赖已包含在Jar和pom.xml中。

可能原因

配置混乱干扰初始化：代码同时混用原生Kafka ProducerConfig 和Spring Kafka前缀（spring.kafka.*）的配置，打包运行时Spring Boot的配置加载逻辑与直接运行存在差异，导致生产者配置未正确生效；另外生产者配置中混入了消费者专属参数（如spring.kafka.consumer.*、ConsumerConfig.CLIENT_ID_CONFIG），会干扰生产者初始化流程。
回调异常被吞噬：PublishCallback()的实现未正确处理发送失败的异常，导致错误信息无法输出到日志，掩盖了真实问题。
打包依赖隐藏问题：虽然确认kafka-client依赖存在，但Spring Boot打包时可能存在间接依赖缺失、版本冲突，导致生产者初始化静默失败。
日志级别过低：默认日志级别未开启Kafka相关调试日志，无法捕捉到初始化、连接环节的细节问题。

解决方法

1. 标准化生产者配置

移除消费者相关参数，统一使用原生Kafka ProducerConfig 常量，或改用Spring Boot自动配置减少手动出错：

@Bean
public ProducerFactory<String, byte[]> producerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    // 仅保留生产者必要配置
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
    configProps.put(ProducerConfig.CLIENT_ID_CONFIG,"ClientIDdata-group"); // 使用Producer专属的CLIENT_ID配置
    return new DefaultKafkaProducerFactory<>(configProps);
}

更推荐通过application.yml配置让Spring Boot自动创建KafkaTemplate：

spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.ByteArraySerializer
      client-id: ClientIDdata-group

之后直接注入KafkaTemplate即可，无需手动定义Bean。

2. 完善回调的异常日志

确保回调逻辑完整记录失败信息：

futureResult.addCallback(new ListenableFutureCallback<SendResult<String, byte[]>>() {
    @Override
    public void onSuccess(SendResult<String, byte[]> result) {
        log.info("消息发送成功，topic: {}, offset: {}", topic, result.getRecordMetadata().offset());
    }

    @Override
    public void onFailure(Throwable ex) {
        // 强制记录错误堆栈，避免异常被吞噬
        log.error("消息发送失败，topic: {}", topic, ex);
    }
});

3. 检查打包配置

确认Spring Boot打包插件配置正确，避免依赖缺失：

<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
            <version>匹配你的Spring Boot版本</version>
            <executions>
                <execution>
                    <goals>
                        <goal>repackage</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

可通过jar tf your-app.jar命令查看Jar内的kafka-client相关类是否存在，验证版本一致性。

4. 开启Kafka调试日志

在application.properties中提高日志级别，捕捉初始化细节：

logging.level.org.apache.kafka=DEBUG
logging.level.org.springframework.kafka=DEBUG

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