非测试场景下内存/嵌入式Kafka实现及Spring Boot适配方案咨询

我来给你拆解这两个适配你场景的方案，结合你提到的多应用架构、仅作为跨应用消息中转的特点，具体实现如下：

这里的核心是把内嵌Kafka配置成外部可访问、稳定运行的实例，而非仅用于单元测试的临时组件。

• 依赖引入
  不需要额外引入测试依赖，直接用Spring Kafka自带的EmbeddedKafka组件，确保项目中已有spring-kafka依赖（以Maven为例）：

  <dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
  </dependency>
  

• 配置类实现
  创建配置类初始化内嵌Kafka Broker，设置监听地址为0.0.0.0让同环境的其他应用可以连接，同时调整分区数等参数适配业务需求：

  @Configuration
  public class EmbeddedKafkaConfig {
  
      @Value("${embedded.kafka.port:9092}")
      private int kafkaPort;
  
      @Bean
      public EmbeddedKafkaBroker embeddedKafkaBroker() {
          return new EmbeddedKafkaBroker(1)
                  .brokerListProperty("spring.kafka.bootstrap-servers")
                  .listeners("PLAINTEXT://0.0.0.0:" + kafkaPort)
                  .port(kafkaPort)
                  .partitions(3); // 根据业务吞吐量调整分区数
      }
  
      // 复用Spring Kafka自动配置的ProducerFactory和ConsumerFactory即可，它们会自动读取bootstrap-servers配置
  }
  

• 关键注意事项

  • 仅适合单实例部署你的Spring Boot应用：如果部署多实例，每个实例会启动独立的内嵌Kafka，导致跨应用消息无法互通。
  • 调整JVM内存：内嵌Kafka会占用一定内存，建议设置-Xmx2G以上的堆内存，避免OOM。
  • 可选临时持久化：如果需要重启后保留消息，可以通过brokerProperties配置log.dir指向本地磁盘目录。

针对你提到的「生产者相关方案较少」的痛点，重点解决生产者启动时的强制校验问题，同时兼顾消费者的容错：

生产者容错配置 #

默认情况下，Spring Kafka的生产者在启动时会尝试连接Kafka集群获取元数据，失败则直接导致应用启动失败。我们可以通过以下方式规避：

• 延迟生产者Bean初始化
  用@Lazy注解标记生产者相关Bean，让它们直到第一次发送消息时才初始化，避免启动时的连接校验：

  @Configuration
  public class KafkaProducerConfig {
  
      @Lazy
      @Bean
      public ProducerFactory<String, Object> producerFactory(KafkaProperties properties) {
          return new DefaultKafkaProducerFactory<>(properties.buildProducerProperties());
      }
  
      @Lazy
      @Bean
      public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> producerFactory) {
          return new KafkaTemplate<>(producerFactory);
      }
  }
  

• 添加发送重试与本地缓存
  为了避免Kafka不可用时丢失消息，用RetryTemplate包装发送逻辑，并将发送失败的消息暂存到本地数据库或文件，待Kafka恢复后重试：

  @Service
  public class KafkaMessageSender {
  
      private final KafkaTemplate<String, Object> kafkaTemplate;
      private final RetryTemplate retryTemplate;
      private final LocalMessageRepository localMessageRepository; // 自定义本地消息存储组件
  
      // 构造函数注入省略
  
      public void sendMessage(String topic, Object payload) {
          try {
              retryTemplate.execute(context -> {
                  kafkaTemplate.send(topic, payload).get(5, TimeUnit.SECONDS);
                  return null;
              });
          } catch (Exception e) {
              // 发送失败，保存到本地待重试
              localMessageRepository.save(new LocalMessage(topic, payload, LocalDateTime.now()));
              log.error("Failed to send message to topic {}: {}", topic, e.getMessage());
          }
      }
  
      // 定时任务：每分钟扫描本地消息，尝试重新发送
      @Scheduled(fixedRate = 60000)
      public void retryLocalMessages() {
          // 实现逻辑：读取未发送的消息，调用sendMessage重试，成功后删除记录
      }
  }
  

消费者容错配置 #

这部分你提到方案较多，这里给出适配你场景的最简配置：

• 在application.yml中添加：

  spring:
    kafka:
      consumer:
        auto-offset-reset: latest # 主题不存在或无偏移量时，从最新位置开始消费
      listener:
        missing-topics-fatal: false # 订阅的主题不存在时，不阻止应用启动
        auto-startup: false # 手动控制消费者容器启动，避免启动时连接Kafka
  

• 可以在应用启动后，通过KafkaListenerEndpointRegistry手动启动消费者容器，或者在Kafka连接恢复后再触发启动。

健康状态提示 #

添加自定义HealthIndicator，让运维能直观看到Kafka的连接状态：

@Component
public class KafkaHealthIndicator implements HealthIndicator {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    // 构造函数注入省略

    @Override
    public Health health() {
        try {
            // 发送一条测试消息到临时主题（或检查集群元数据）
            kafkaTemplate.send("health-check-topic", "test").get(3, TimeUnit.SECONDS);
            return Health.up().withDetail("kafka-status", "connected").build();
        } catch (Exception e) {
            return Health.down().withDetail("kafka-status", "disconnected").withDetail("error", e.getMessage()).build();
        }
    }
}

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