一、引言

在微服务架构和分布式系统日益普及的今天，服务之间的通信方式成为架构设计中的关键决策点。传统的同步RESTful调用虽然直观，但会带来服务间的强耦合——上游服务必须等待下游服务响应，一旦下游服务出现故障或响应缓慢，整个调用链都会受到影响。

Apache Kafka 作为一种高性能的分布式消息队列系统，为这一问题提供了优雅的解决方案。它基于发布-订阅模式，具备高吞吐量、持久化存储和良好的可扩展性等特性。而 Spring Boot 则通过 spring-kafka 项目提供了与 Kafka 集成的自动配置支持，让开发者能够以极少的样板代码实现消息的发送与接收。

本文将系统性地介绍如何在 Spring Boot 项目中集成 Kafka，从基础概念到代码实现，从配置详解到最佳实践，帮助读者快速上手并应用于实际项目。

二、环境准备

在开始集成之前，请确保本地已安装并启动 Kafka。

三、Spring Boot 集成 Kafka

本次采用 spring-boot-3.3.5 + java21 为例进行集成演示：

1. 添加依赖

在 Spring Boot 项目的 pom.xml 中添加 spring-kafka 依赖：

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

2. 配置文件YML

spring:
# Kafka 消息队列配置
  kafka:
    # Kafka 服务端地址（IP:端口）
    bootstrap-servers: ip:9092
    # 生产者配置：负责发送消息到 Kafka
    producer:
      # Key 序列化方式：字符串序列化
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # Value 序列化方式：字符串序列化
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 消息确认机制：1 = 仅Leader确认、all = 全部确认
      acks: all
      # 消息发送失败重试次数
      retries: 3
      # 批量发送的消息字节大小
      batch-size: 16384
      # 消息延迟发送时间（毫秒），提高吞吐量
      linger-ms: 1
      # 生产者可用的缓存总量
      buffer-memory: 33554432
      # 压缩类型（none / gzip / snappy / lz4 / zstd）
      compression-type: gzip
      # 扩展属性（Spring Boot 未直接映射的 Kafka 原生配置）
      properties:
        # 幂等生产者（保证 exactly-once，要求 acks=all）
        enable.idempotence: true
        # 单连接最大未确认请求数（幂等模式下安全上限为 5）
        max.in.flight.requests.per.connection: 5
        # 单次请求超时（毫秒）
        request.timeout.ms: 30000
        # 端到端投递超时（含重试，毫秒）
        delivery.timeout.ms: 60000
#        # 事务ID前缀（设置后自动开启事务支持、创建 KafkaTransactionManager）
#        transaction.id.prefix: tx-template-
#        # 事务超时时间（毫秒）
#        transaction.timeout.ms: 60000
    # 消费者配置：负责从 Kafka 拉取消息
    consumer:
      # Key 反序列化方式
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # Value 反序列化方式
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 消费者组ID，同一组内的消费者共同消费消息
      group-id: template-group
      # 无初始偏移量时，从头开始消费 earliest / 从最新开始消费 latest
      auto-offset-reset: earliest
      # 是否开启自动提交偏移量
      enable-auto-commit: true
      # 自动提交偏移量的时间间隔（毫秒）
      auto-commit-interval: 1000
      # 每次拉取的最大消息条数
      max-poll-records: 500
      # 扩展属性
      properties:
        # 会话超时（毫秒），超时触发 rebalance
        session.timeout.ms: 30000
        # 心跳间隔（毫秒），建议为 session.timeout 的 1/10
        heartbeat.interval.ms: 3000
        # poll 间隔上限（毫秒），超过触发 rebalance
        max.poll.interval.ms: 300000
    # 监听器配置：Spring Kafka 消费监听规则
    listener:
      # 批量消费确认模式
      ack-mode: batch
      # 消费者并发线程数 = 消费者数量
      concurrency: 2
    # 自定义 Topic 配置
    topics:
      # 测试调试专用
      test: test-topic
      # 模板业务
      template: template-topic

四、Kafka 配置类

1.KafkaConfig

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;

/**
 * <h3>Kafka 集中配置</h3>
 * <pre>
 * 职责分层：
 *   1. AdminClient  — 运维管理（健康检查、Topic 列表）
 *   2. Producer     — 消息生产（幂等 + Snappy 压缩 + 超时控制）
 *   3. Consumer     — 消息消费（手动创建 Factory，注入 ErrorHandler）
 *   4. ErrorHandler — 消费异常重试（3 次 + 1s 间隔，耗尽记日志）
 *   5. HealthCheck  — Broker 连通性探测
 * </pre>
 *
 */
@Slf4j
@Configuration
@RequiredArgsConstructor
public class KafkaConfig {

    private final KafkaProperties kafkaProperties;

    // ═══════════════════════════════════════════════════════════════════
    // 1. AdminClient — 运维管理
    // ═══════════════════════════════════════════════════════════════════

    /**
     * 原生 AdminClient，用于 Topic 管理、集群元数据查询。
     * 每次调用都创建新实例（轻量），避免长连接空转。
     */
    @Bean
    public AdminClient kafkaAdminClient() {
        Map<String, Object> config = kafkaProperties.buildAdminProperties(null);
        return AdminClient.create(config);
    }

    /**
     * Spring KafkaAdmin Bean，配合 @Bean newTopic() 自动创建 Topic。
     * 调试期间 Broker 不可用时不阻塞启动。
     */
    @Bean
    public KafkaAdmin kafkaAdmin() {
        KafkaAdmin admin = new KafkaAdmin(kafkaProperties.buildProducerProperties(null));
        admin.setFatalIfBrokerNotAvailable(false);
        return admin;
    }

    /**
     * KafkaTemplate — 线程安全，单例复用。
     * String 泛型，适合 JSON 或纯文本消息。
     */
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    // ═══════════════════════════════════════════════════════════════════
    // 2. Producer — 消息生产
    // ═══════════════════════════════════════════════════════════════════

    /**
     * Producer 工厂。
     * <p>
     * 所有参数通过 {@code spring.kafka.producer.*} 统一配置在 application.yml，<br>
     * 包括 compression-type、acks、retries、以及扩展 properties（enable.idempotence、<br>
     * max.in.flight.requests、request.timeout.ms、delivery.timeout.ms 等）。
     * <p>
     * 这里不做任何硬编码覆盖，保持 Java 代码与 yml 配置单一数据源。
     */
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> props = kafkaProperties.buildProducerProperties(null);
        return new DefaultKafkaProducerFactory<>(props);
    }


    // ═══════════════════════════════════════════════════════════════════
    // 3. Consumer — 消息消费
    // ═══════════════════════════════════════════════════════════════════

    /**
     * Consumer 工厂。
     * <p>
     * 所有参数通过 {@code spring.kafka.consumer.*} 统一配置在 application.yml，<br>
     * 包括 session.timeout.ms、heartbeat.interval.ms、max.poll.interval.ms 等扩展属性。
     * <p>
     * 不做硬编码覆盖，与 yml 保持单一数据源。
     */
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = kafkaProperties.buildConsumerProperties(null);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    /**
     * Listener 容器工厂，生产环境定制入口。
     * <p>
     * concurrency 从 {@code spring.kafka.listener.concurrency} 读取（yml 单一数据源），
     * CommonErrorHandler 注入统一异常重试策略。
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        // 从 yml 读取并发数，不硬编码
        Integer concurrency = kafkaProperties.getListener().getConcurrency();
        factory.setConcurrency(concurrency != null ? concurrency : 1);
        factory.setBatchListener(false);
        factory.setCommonErrorHandler(kafkaErrorHandler());
        return factory;
    }

    // ═══════════════════════════════════════════════════════════════════
    // 4. ErrorHandler — 消费异常处理
    // ═══════════════════════════════════════════════════════════════════

    /**
     * 消费者异常处理策略。
     * <p>
     * 重试 3 次，每次间隔 1 秒。全部失败后记录日志并提交 offset（跳过该消息）。
     * <p>
     * 线上加强方案：
     * <ol>
     *   <li>升级为指数退避：{@code new ExponentialBackOff(1000L, 2.0)}</li>
     *   <li>接入 DLT（死信队列）：{@code new DeadLetterPublishingRecoverer(template)}</li>
     * </ol>
     */
    private DefaultErrorHandler kafkaErrorHandler() {
        return new DefaultErrorHandler(
                (record, exception) -> log.error(
                        "Kafka 消费失败（重试耗尽）→ topic={}, partition={}, offset={}, key={}, value={}",
                        record.topic(), record.partition(), record.offset(), record.key(), record.value(), exception),
                new FixedBackOff(1000L, 3L)
        );
    }

    // ═══════════════════════════════════════════════════════════════════
    // 5. 运维工具 — 健康检查 & Topic 管理
    // ═══════════════════════════════════════════════════════════════════

    /**
     * 探测 Broker 是否可达。
     * 调用 listTopics() 做轻量 RTT，超时由 AdminClient 默认值兜底。
     *
     * @return true = Broker 连通
     */
    public boolean isBrokerAlive() {
        try (AdminClient client = AdminClient.create(kafkaProperties.buildAdminProperties(null))) {
            client.listTopics().names().get();
            return true;
        } catch (Exception e) {
            log.warn("Kafka Broker 连接失败: {}", e.getMessage());
            return false;
        }
    }

    /**
     * 列出 Broker 上所有 Topic。
     *
     * @return Topic 名称集合
     */
    public Set<String> listTopics() throws ExecutionException, InterruptedException {
        try (AdminClient client = AdminClient.create(kafkaProperties.buildAdminProperties(null))) {
            return client.listTopics().names().get();
        }
    }

2.KafkaTopicProperties

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

/**
 * Kafka 自定义 Topic 配置（映射 spring.kafka.topics.*）
 * <p>
 * 标准属性（bootstrap-servers、producer、consumer、listener）由 Spring Boot 自动装配，<br>
 * 可直接注入 {@link org.springframework.boot.autoconfigure.kafka.KafkaProperties} 获取。<br>
 * 本类仅定义自定义 topic 名称，避免与 Spring Boot 的 KafkaProperties Bean 冲突。
 *
 */
@Data
@Component
@ConfigurationProperties(prefix = "spring.kafka.topics")
public class KafkaTopicProperties {

    /** 测试调试专用 */
    private String test = "test-topic";

    /** 模板业务 */
    private String template = "template-topic";
}

3.KafkaProducerService

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;

/**
 * Kafka 生产者服务（生产级）
 * <p>
 * 功能：异步/同步发送、消息头注入、超时兜底、链路追踪 ID
 *
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class KafkaProducerService {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final KafkaTopicProperties kafkaTopicProperties;

    private static final DateTimeFormatter TIME_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");

    // ==================== 异步发送 ===========================================

    /**
     * 发送消息（异步），返回 CompletableFuture 供调用方自行处理
     */
    public CompletableFuture<SendResult<String, String>> sendAsync(String topic, String message) {
        return sendAsync(topic, null, message);
    }

    /**
     * 发送消息（异步，带 key）
     */
    public CompletableFuture<SendResult<String, String>> sendAsync(String topic, String key, String message) {
        // 自动注入消息头 ProducerRecord
        ProducerRecord<String, String> record = buildRecord(topic, key, message);
        long start = System.currentTimeMillis();
        return kafkaTemplate.send(record)
                .whenComplete((result, ex) -> {
                    long elapsed = System.currentTimeMillis() - start;
                    if (ex == null) {
                        log.info("Kafka 发送成功 → topic={}, partition={}, offset={}, key={}, size={}B, cost={}ms",
                                result.getRecordMetadata().topic(),
                                result.getRecordMetadata().partition(),
                                result.getRecordMetadata().offset(),
                                key,
                                message.length(),
                                elapsed);
                    } else {
                        log.error("Kafka 发送失败 → topic={}, key={}, cost={}ms, error={}",
                                topic, key, elapsed, ex.getMessage(), ex);
                    }
                });
    }

    // ==================== 同步发送（有超时） ====================================

    /**
     * 同步发送，5 秒超时，返回结果或抛异常
     */
    public SendResult<String, String> sendSync(String topic, String message) throws Exception {
        return sendSync(topic, null, message);
    }

    /**
     * 同步发送（带 key），5 秒超时
     */
    public SendResult<String, String> sendSync(String topic, String key, String message) throws Exception {
        try {
            SendResult<String, String> result = sendAsync(topic, key, message).get();
            return result;
        } catch (Exception e) {
            log.error("Kafka 同步发送异常 → topic={}, key={}", topic, key, e);
            throw e;
        }
    }

    // ==================== 便捷方法（兼容旧接口） ====================

    /**
     * 发送消息（异步，不返回值）— 兼容旧调用
     */
    public void send(String topic, String message) {
        sendAsync(topic, message);
    }

    /**
     * 发送消息（异步，带 key）— 兼容旧调用
     */
    public void send(String topic, String key, String message) {
        sendAsync(topic, key, message);
    }

    // ==================== 内部构建 ===============================================

    /**
     * 构建 ProducerRecord，自动注入消息头
     */
    private ProducerRecord<String, String> buildRecord(String topic, String key, String message) {
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, null, System.currentTimeMillis(), key, message);

        // 注入消息头（线上标配：链路追踪 ID + 来源 + 时间戳）
        record.headers().add(new RecordHeader("X-Message-Id", UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8)));
        record.headers().add(new RecordHeader("X-Source", "template".getBytes(StandardCharsets.UTF_8)));
        record.headers().add(new RecordHeader("X-Timestamp", LocalDateTime.now().format(TIME_FMT).getBytes(StandardCharsets.UTF_8)));

        return record;
    }

    /**
     * 检查 broker 是否可用（简单 probe）
     */
    public boolean isReady() {
        try {
            kafkaTemplate.partitionsFor(kafkaTopicProperties.getTest());
            return true;
        } catch (Exception e) {
            return false;
        }
    }
}

4.KafkaController

import com.github.xiaoymin.knife4j.annotations.ApiOperationSupport;
import com.www.template.exception.R;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.*;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * Kafka 调试控制器
 *
 */
@Slf4j
@AllArgsConstructor
@RestController
@RequestMapping("/kafka")
@Tag(name = "Kafka调试")
public class KafkaController {

    private final KafkaProducerService producerService;
    private final KafkaConfig kafkaConfig;

    // ==================== 发送 ====================

    @PostMapping("/send")
    @Operation(summary = "1.发送消息（异步）")
    @ApiOperationSupport(order = 1)
    public R<?> send(@RequestParam(defaultValue = "test-topic") String topic,
                     @RequestParam String message) {
        producerService.sendAsync(topic, message);
        return R.ok("消息已发送 → topic=" + topic + ", message=" + message);
    }

    @PostMapping("/send-with-key")
    @Operation(summary = "2.发送带Key的消息（异步）")
    @ApiOperationSupport(order = 2)
    public R<?> sendWithKey(@RequestParam(defaultValue = "test-topic") String topic,
                            @RequestParam String key,
                            @RequestParam String message) {
        producerService.sendAsync(topic, key, message);
        return R.ok("消息已发送 → topic=" + topic + ", key=" + key + ", message=" + message);
    }

    @PostMapping("/send-batch")
    @Operation(summary = "3.批量发送（N条消息）")
    @ApiOperationSupport(order = 3)
    public R<?> sendBatch(@RequestParam(defaultValue = "test-topic") String topic,
                          @RequestParam(defaultValue = "hello kafka") String prefix,
                          @RequestParam(defaultValue = "10") int count) {
        for (int i = 0; i < count; i++) {
            producerService.sendAsync(topic, prefix + " [" + i + "]");
        }
        return R.ok("已发送 " + count + " 条消息 → topic=" + topic);
    }

    @PostMapping("/send-sync")
    @Operation(summary = "4.发送消息（同步，等待 Broker 确认）")
    @ApiOperationSupport(order = 4)
    public R<?> sendSync(@RequestParam(defaultValue = "test-topic") String topic,
                         @RequestParam String message) {
        try {
            var result = producerService.sendSync(topic, message);
            Map<String, Object> data = new HashMap<>();
            data.put("topic", result.getRecordMetadata().topic());
            data.put("partition", result.getRecordMetadata().partition());
            data.put("offset", result.getRecordMetadata().offset());
            data.put("message", message);
            return R.ok(data);
        } catch (Exception e) {
            return R.error("发送失败: " + e.getMessage());
        }
    }

    // ==================== 运维 ====================

    @GetMapping("/health")
    @Operation(summary = "5.Broker 连通性检查")
    @ApiOperationSupport(order = 5)
    public R<?> health() {
        Map<String, Object> info = new HashMap<>();
        info.put("brokerAlive", kafkaConfig.isBrokerAlive());
        info.put("producerReady", producerService.isReady());
        return R.ok(info);
    }

    @GetMapping("/topics")
    @Operation(summary = "6.Topic 列表")
    @ApiOperationSupport(order = 6)
    public R<?> topics() {
        try {
            Set<String> topics = kafkaConfig.listTopics();
            return R.ok(topics);
        } catch (Exception e) {
            return R.error("获取 Topic 列表失败: " + e.getMessage());
        }
    }
}

5.

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * Kafka 消费者服务
 * <p>
 * 启动即监听，收到消息后处理业务逻辑
 */
@Slf4j
@Component
public class KafkaConsumerService {

    /**
     * 监听 test-topic
     */
    @KafkaListener(topics = "${spring.kafka.topics.test}", groupId = "${spring.kafka.consumer.group-id}")
    public void listenTestTopic(ConsumerRecord<String, String> record) {
        handle(record);
    }

    /**
     * 监听 template-topic
     */
    @KafkaListener(topics = "${spring.kafka.topics.template}", groupId = "${spring.kafka.consumer.group-id}")
    public void listenTemplateTopic(ConsumerRecord<String, String> record) {
        handle(record);
    }

    /**
     * 业务处理入口（线上在这里接具体业务逻辑）
     */
    private void handle(ConsumerRecord<String, String> record) {
        log.info("Kafka 消费 → topic={}, partition={}, offset={}, key={}, value={}",
                record.topic(), record.partition(), record.offset(), record.key(), record.value());
    }
}

五、总结

本文从 Kafka 的核心概念出发，系统介绍了在 Spring Boot 项目中集成 Kafka 的完整流程：

1. 添加依赖 —— 引入 spring-kafka

2. 配置参数 —— 在 application.yml 中配置 Broker 地址、生产者和消费者参数

3. 实现生产者 —— 使用自动配置的 KafkaTemplate 发送消息

4. 实现消费者 —— 使用 @KafkaListener 注解轻松消费消息

5. 进阶实践 —— 自定义配置、错误处理、幂等设计和性能调优

Spring Boot 与 Kafka 的结合，让开发者能够以极低的门槛构建高吞吐、低延迟的消息驱动应用。希望本文能帮助读者快速上手，并在实际项目中灵活运用。