SpringBoot集成Kafka企业级实战教程

本教程从零开始，基于 IntelliJ IDEA 2023 Ultimate 搭建可直接用于生产环境的 SpringBoot-Kafka 项目，覆盖生产者/消费者封装、SASL 认证、异常处理、死信队列、性能调优等企业级核心特性，代码规范且具备高可用性。

---

1. 环境准备

1.1 开发环境要求

确保本地开发环境符合以下版本要求（SpringBoot 3.x 对 JDK 版本有强制要求）：

组件	版本/要求
IDE	IntelliJ IDEA 2023 Ultimate 版
JDK	17（SpringBoot 3.x 最低要求）
Maven	3.9.4
SpringBoot	3.5.10
Kafka 服务端	3.9.1
操作系统	Windows/macOS/Ubuntu（无特殊限制）

1.2 Kafka 服务端核心信息

本次集成使用的 Kafka 服务配置：

• 连接地址：www.huerpu.cc
• 端口：9092
• 认证方式：SASL_PLAINTEXT
• 认证账号：kelsen / k1029384756
• 服务端配置详情：见「9. Kafka服务器端配置」章节

---

2. 项目初始化

创建 SpringBoot 项目

企业级项目结构 初始化后整理项目结构如下（后续代码严格按此路径编写）：

SpringBoot-Kafka/
├── src/
│   ├── main/
│   │   ├── java/
│   │   │   └── cc/
│   │   │       └── hep/
│   │   │           ├── SpringBootKafkaApplication.java  // 启动类
│   │   │           ├── config/                         // 配置类
│   │   │           │   └── KafkaConfig.java
│   │   │           ├── constant/                       // 常量类
│   │   │           │   └── KafkaConstant.java
│   │   │           ├── entity/                         // 消息实体
│   │   │           │   └── TestMessage.java
│   │   │           ├── service/                        // 核心服务
│   │   │           │   ├── KafkaProducerService.java
│   │   │           │   └── KafkaConsumerService.java
│   │   │           ├── controller/                     // 测试控制器
│   │   │           │   └── KafkaTestController.java
│   │   └── resources/
│   │       ├── application.yml                         // 核心配置
│   │       └── logback-spring.xml                      // 日志配置（可选）
│   └── test/
│       └── java/
│           └── cc/
│               └── hep/
│                   └── KafkaIntegrationTest.java
└── pom.xml

---

3. 核心依赖配置

替换 IDEA 自动生成的 pom.xml，确保依赖版本匹配且包含企业级必备组件：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.5.10</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <groupId>cc.hep</groupId>
    <artifactId>SpringBoot-Kafka</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>SpringBoot-Kafka</name>
    <description>SpringBoot 3.5.10 integrate Kafka 3.9.1 enterprise tutorial</description>

    <properties>
        <java.version>17</java.version>
    </properties>

    <dependencies>
        <!-- SpringBoot Web 依赖（测试用） -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <!-- Spring Kafka 核心依赖（使用 Spring Boot 管理的版本） -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

        <!-- JSON 序列化（企业级常用） -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>

        <!-- Lombok（简化代码） -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>

        <!-- 测试依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>

            <!-- 配置跳过测试 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>3.2.5</version>
                <configuration>
                    <skipTests>true</skipTests>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

---

4. Kafka 核心配置

在 resources/application.yml 中配置 Kafka 连接、认证、生产者/消费者参数（企业级调优版）：

spring:
  kafka:
    # 连接地址（根据实际地址修改）
    bootstrap-servers: www.huerpu.cc:9092
    # ====== 消费者配置（核心）======
    consumer:
      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      group-id: hep-kafka-group
      # 批量消费配置
      fetch-max-wait: 5000
      max-poll-records: 50
      properties:
        # 委托真正的反序列化器
        spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
        # 关键：设置默认反序列化类型
        spring.json.value.default.type: java.lang.String
        spring.json.trusted.packages: "*"
    # ====== 安全与公共属性 ======
    properties:
      security.protocol: SASL_PLAINTEXT
      sasl.mechanism: PLAIN
      # SASL 认证配置（核心）
      sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="jason" password="jason1029384756";
    # ====== 生产者配置（测试用）======
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

🔴 核心配置说明：

1. SASL 认证：sasl.jaas.config 是连接 Kafka 的核心，需与服务端账号一致；
2. 消费者容错：ErrorHandlingDeserializer 包装反序列化器，避免单条消息反序列化失败导致消费中断；
3. 手动提交：结合配置类的 MANUAL_IMMEDIATE，确保消息消费成功后再提交 offset，避免丢失；
4. JSON 反序列化：trusted.packages: "*" 避免安全限制，生产环境建议指定具体包名。

---

5. 企业级核心代码实现

5.1 常量类

封装 Topic 常量，避免硬编码（企业级规范）：

package cc.hep.constant;

/**
 * Kafka 常量类（统一管理 Topic、配置键等）
 * @author hep
 */
public class KafkaConstant {
    /**
     * 测试 Topic（命名规范：业务_模块_功能）
     */
    public static final String TEST_TOPIC = "hep-test-topic";

    /**
     * 死信 Topic（消费失败后转发）
     */
    public static final String DEAD_LETTER_TOPIC = "hep-dead-letter-topic";

    /**
     * 消费异常重试间隔（ms）
     */
    public static final long CONSUME_RETRY_INTERVAL = 2000L;
}

5.2 消息实体类

定义业务消息结构（JSON 序列化必备）：

package cc.hep.entity;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.io.Serializable;
import java.time.LocalDateTime;

/**
 * Kafka 测试消息实体（实现 Serializable 保证序列化兼容性）
 * @author hep
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class TestMessage implements Serializable {
    /**
     * 消息ID
     */
    private String msgId;

    /**
     * 消息内容
     */
    private String content;

    /**
     * 发送时间
     */
    private LocalDateTime sendTime;
}

📌 实体类注意点：

• 必须添加 @NoArgsConstructor（JSON 反序列化需要无参构造）；
• 实现 Serializable 确保跨版本/跨系统序列化兼容；
• 字段命名规范，避免反序列化字段不匹配。

5.3 Kafka 配置类

自定义 KafkaTemplate 和消费者容器工厂，扩展企业级特性：

package cc.hep.config;

import cc.hep.constant.KafkaConstant;
import cc.hep.entity.TestMessage;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.util.backoff.FixedBackOff;

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

/**
 * Kafka 自定义配置类（扩展默认配置，添加死信队列、异常处理）
 * @author hep
 */
@Configuration
@EnableKafka // 启用 Kafka 监听
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    /**
     * 自定义 KafkaTemplate（生产者核心）
     */
    @Bean
    public KafkaTemplate<String, TestMessage> kafkaTemplate(ProducerFactory<String, TestMessage> producerFactory) {
        KafkaTemplate<String, TestMessage> template = new KafkaTemplate<>(producerFactory);
        template.setDefaultTopic(KafkaConstant.TEST_TOPIC);
        return template;
    }

    /**
     * 生产者工厂（自定义序列化配置）
     */
    @Bean
    public ProducerFactory<String, TestMessage> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        // Kafka 服务器地址
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        // SASL 认证配置
        configProps.put("security.protocol", "SASL_PLAINTEXT");
        configProps.put("sasl.mechanism", "PLAIN");
        configProps.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"kelsen\" password=\"k1029384756\";");

        // 序列化配置
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        configProps.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);

        // 其他生产者配置
        configProps.put(ProducerConfig.ACKS_CONFIG, "1");
        configProps.put(ProducerConfig.RETRIES_CONFIG, 3);

        return new DefaultKafkaProducerFactory<>(configProps);
    }

    /**
     * 消费者工厂（自定义反序列化配置）
     */
    @Bean
    public ConsumerFactory<String, TestMessage> consumerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        // 1. 基础连接
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        configProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        configProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 批量消费配置
        configProps.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 5000);
        configProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);

        // 2. 核心：ErrorHandlingDeserializer 包装反序列化器
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
        // 指定委托类
        configProps.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, StringDeserializer.class);
        configProps.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);

        // 3. JsonDeserializer 专属配置
        configProps.put(JsonDeserializer.VALUE_DEFAULT_TYPE, TestMessage.class.getName());
        configProps.put(JsonDeserializer.TRUSTED_PACKAGES, "cc.hep");
        configProps.put(JsonDeserializer.REMOVE_TYPE_INFO_HEADERS, false);

        // 4. SASL 认证配置
        configProps.put("security.protocol", "SASL_PLAINTEXT");
        configProps.put("sasl.mechanism", "PLAIN");
        configProps.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"kelsen\" password=\"k1029384756\";");

        return new DefaultKafkaConsumerFactory<>(configProps);
    }

    /**
     * 消费者监听容器工厂（核心：批量消费、手动提交、死信队列）
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, TestMessage> kafkaListenerContainerFactory(
            ConsumerFactory<String, TestMessage> consumerFactory,
            KafkaTemplate<String, TestMessage> kafkaTemplate) {

        ConcurrentKafkaListenerContainerFactory<String, TestMessage> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);

        // 配置批量消费
        factory.setBatchListener(true);
        // 手动提交 offset
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        // 并发消费线程数
        factory.setConcurrency(3);

        // 异常处理：重试3次后转发到死信队列
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
                (record, exception) -> new org.apache.kafka.common.TopicPartition(KafkaConstant.DEAD_LETTER_TOPIC, record.partition()));

        // 重试间隔配置
        FixedBackOff backOff = new FixedBackOff(KafkaConstant.CONSUME_RETRY_INTERVAL, 3);
        DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoverer, backOff);
        factory.setCommonErrorHandler(errorHandler);

        return factory;
    }
}

5.4 生产者服务封装

企业级生产者封装，支持同步/异步发送、回调处理、异常重试：

package cc.hep.service;

import cc.hep.constant.KafkaConstant;
import cc.hep.entity.TestMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import jakarta.annotation.Resource;
import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;

/**
 * Kafka 生产者服务（企业级封装：同步/异步发送、异常处理、回调）
 * @author hep
 */
@Slf4j
@Service
public class KafkaProducerService {

    @Resource
    private KafkaTemplate<String, TestMessage> kafkaTemplate;

    /**
     * 异步发送消息（无回调，简单场景）
     * @param content 消息内容
     */
    public void sendAsyncMessage(String content) {
        try {
            TestMessage message = new TestMessage(UUID.randomUUID().toString(), content, LocalDateTime.now());
            kafkaTemplate.send(KafkaConstant.TEST_TOPIC, message.getMsgId(), message);
            log.info("异步发送消息成功，msgId: {}", message.getMsgId());
        } catch (Exception e) {
            log.error("异步发送消息失败", e);
            throw new RuntimeException("Kafka 异步发送失败", e);
        }
    }

    /**
     * 异步发送消息（带回调，企业级常用）
     * @param content 消息内容
     */
    public void sendAsyncMessageWithCallback(String content) {
        TestMessage message = new TestMessage(UUID.randomUUID().toString(), content, LocalDateTime.now());

        CompletableFuture<SendResult<String, TestMessage>> future =
                kafkaTemplate.send(KafkaConstant.TEST_TOPIC, message.getMsgId(), message);

        // 回调处理
        future.whenComplete((result, ex) -> {
            if (ex == null) {
                log.info("回调发送消息成功，msgId: {}, partition: {}, offset: {}",
                        message.getMsgId(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
            } else {
                log.error("回调发送消息失败，msgId: {}", message.getMsgId(), ex);
                retrySendMessage(message);
            }
        });
    }

    /**
     * 同步发送消息（阻塞等待，关键业务用）
     * @param content 消息内容
     * @return 发送结果
     */
    public SendResult<String, TestMessage> sendSyncMessage(String content) {
        TestMessage message = new TestMessage(UUID.randomUUID().toString(), content, LocalDateTime.now());
        try {
            SendResult<String, TestMessage> result = kafkaTemplate.send(KafkaConstant.TEST_TOPIC, message.getMsgId(), message)
                    .join();

            log.info("同步发送消息成功，msgId: {}, partition: {}, offset: {}",
                    message.getMsgId(),
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset());
            return result;
        } catch (Exception e) {
            log.error("同步发送消息失败，msgId: {}", message.getMsgId(), e);
            throw new RuntimeException("Kafka 同步发送失败", e);
        }
    }

    /**
     * 异步发送消息（链式调用，现代写法）
     */
    public void sendAsyncMessageModern(String content) {
        TestMessage message = new TestMessage(UUID.randomUUID().toString(), content, LocalDateTime.now());

        kafkaTemplate.send(KafkaConstant.TEST_TOPIC, message.getMsgId(), message)
                .thenAccept(result -> {
                    log.info("消息发送成功，msgId: {}, partition: {}, offset: {}",
                            message.getMsgId(),
                            result.getRecordMetadata().partition(),
                            result.getRecordMetadata().offset());
                })
                .exceptionally(ex -> {
                    log.error("消息发送失败，msgId: {}", message.getMsgId(), ex);
                    retrySendMessage(message);
                    return null;
                });
    }

    /**
     * 重试发送消息（示例）
     */
    private void retrySendMessage(TestMessage message) {
        log.warn("准备重试发送消息，msgId: {}", message.getMsgId());

        try {
            kafkaTemplate.send(KafkaConstant.TEST_TOPIC, message.getMsgId(), message)
                    .whenComplete((result, ex) -> {
                        if (ex == null) {
                            log.info("重试发送成功，msgId: {}", message.getMsgId());
                        } else {
                            log.error("重试发送失败，msgId: {}", message.getMsgId(), ex);
                        }
                    });
        } catch (Exception e) {
            log.error("重试发送异常，msgId: {}", message.getMsgId(), e);
        }
    }
}

5.5 消费者服务封装

企业级消费者封装，支持批量消费、手动提交、死信队列处理：

package cc.hep.service;

import cc.hep.constant.KafkaConstant;
import cc.hep.entity.TestMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * Kafka 消费者服务（批量消费、手动提交、异常处理）
 * @author hep
 */
@Slf4j
@Service
public class KafkaConsumerService {

    /**
     * 批量消费消息（企业级常用）
     * @param records 消息列表
     * @param ack 手动提交 offset 工具
     */
    @KafkaListener(topics = KafkaConstant.TEST_TOPIC, containerFactory = "kafkaListenerContainerFactory")
    public void consumeBatchMessage(List<ConsumerRecord<String, TestMessage>> records, Acknowledgment ack) {
        log.info("收到批量消息，条数：{}", records.size());

        try {
            // 遍历处理每条消息
            for (ConsumerRecord<String, TestMessage> record : records) {
                String msgId = record.key();
                TestMessage message = record.value();
                log.info("处理消息，msgId: {}, content: {}, sendTime: {}",
                        msgId, message.getContent(), message.getSendTime());

                // 业务逻辑处理（替换为实际业务）
                processBusinessLogic(message);
            }

            // 手动提交 offset（确保所有消息处理成功）
            ack.acknowledge();
            log.info("批量消息处理完成，已提交 offset");
        } catch (Exception e) {
            log.error("批量消费消息失败", e);
            // 异常时不提交 offset，Kafka 会重新推送
        }
    }

    /**
     * 死信队列消费（处理消费失败的消息）
     */
    @KafkaListener(topics = KafkaConstant.DEAD_LETTER_TOPIC, containerFactory = "kafkaListenerContainerFactory")
    public void consumeDeadLetterMessage(List<ConsumerRecord<String, TestMessage>> records, Acknowledgment ack) {
        log.warn("收到死信队列消息，条数：{}", records.size());
        try {
            // 死信消息处理：记录日志、告警、人工审核
            for (ConsumerRecord<String, TestMessage> record : records) {
                log.warn("死信消息详情：msgId: {}, content: {}", record.key(), record.value().getContent());
            }
            ack.acknowledge();
        } catch (Exception e) {
            log.error("死信队列消费失败", e);
        }
    }

    /**
     * 模拟业务逻辑处理
     * @param message 消息实体
     */
    private void processBusinessLogic(TestMessage message) {
        log.info("执行业务逻辑：{}", message.getContent());
    }
}

5.6 测试控制器

提供 HTTP 接口快速测试生产者功能：

package cc.hep.controller;

import cc.hep.service.KafkaProducerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import jakarta.annotation.Resource;

/**
 * Kafka 测试控制器（HTTP 接口测试生产者）
 * @author hep
 */
@Slf4j
@RestController
@RequestMapping("/kafka")
public class KafkaTestController {

    @Resource
    private KafkaProducerService kafkaProducerService;

    /**
     * 异步发送消息（无回调）
     * @param content 消息内容
     * @return 响应结果
     */
    @GetMapping("/send/async")
    public ResponseEntity<String> sendAsyncMessage(@RequestParam String content) {
        try {
            kafkaProducerService.sendAsyncMessage(content);
            return ResponseEntity.ok("异步消息发送成功：" + content);
        } catch (Exception e) {
            return ResponseEntity.internalServerError().body("异步消息发送失败：" + e.getMessage());
        }
    }

    /**
     * 异步发送消息（带回调）
     * @param content 消息内容
     * @return 响应结果
     */
    @GetMapping("/send/async/callback")
    public ResponseEntity<String> sendAsyncMessageWithCallback(@RequestParam String content) {
        try {
            kafkaProducerService.sendAsyncMessageWithCallback(content);
            return ResponseEntity.ok("异步消息（带回调）发送成功：" + content);
        } catch (Exception e) {
            return ResponseEntity.internalServerError().body("异步消息（带回调）发送失败：" + e.getMessage());
        }
    }

    /**
     * 同步发送消息
     * @param content 消息内容
     * @return 响应结果
     */
    @GetMapping("/send/sync")
    public ResponseEntity<String> sendSyncMessage(@RequestParam String content) {
        try {
            kafkaProducerService.sendSyncMessage(content);
            return ResponseEntity.ok("同步消息发送成功：" + content);
        } catch (Exception e) {
            return ResponseEntity.internalServerError().body("同步消息发送失败：" + e.getMessage());
        }
    }
}

5.7 启动类

package cc.hep;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class SpringBootKafkaApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringBootKafkaApplication.class, args);
    }

}

---

6. 测试验证

6.1 启动项目

直接运行 SpringBootKafkaApplication.java 的 main 方法，控制台无报错且输出「SpringBoot-Kafka 项目启动成功」即表示启动正常。

6.2 发送消息测试

通过浏览器/Postman 访问以下接口发送消息：

1. 异步发送：http://localhost:8080/kafka/send/async?content=测试异步消息
2. 异步带回调：http://localhost:8080/kafka/send/async/callback?content=测试回调消息
3. 同步发送：http://localhost:8080/kafka/send/sync?content=测试同步消息

6.3 消费结果验证

控制台输出如下日志，说明消费成功：

收到批量消息，条数：1
处理消息，msgId: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx, content: 测试异步消息, sendTime: 2026-01-25T10:00:00
执行业务逻辑：测试异步消息
批量消息处理完成，已提交 offset

---

7. 企业级最佳实践

7.1 消息幂等性

• 生产端：开启 Kafka 幂等性（enable.idempotence: true），避免重复发送；
• 消费端：基于 msgId 做幂等处理（如数据库唯一索引、Redis 缓存标记）。

7.2 全链路消息追踪

• 在消息头中添加 traceId，贯穿生产/消费全链路：

  kafkaTemplate.send(Topic, MessageBuilder.withPayload(msg).setHeader("traceId", traceId).build())
  

• 结合日志框架（如 Logback）输出 traceId，便于问题排查。

7.3 监控与告警

• 集成 Prometheus + Grafana 监控：生产/消费速率、消息堆积量、消费延迟；
• 消费失败/消息堆积时，触发钉钉/邮件/短信告警。

7.4 性能调优

维度	调优建议
生产者	调整 batch-size（批量大小）、compression-type: snappy（压缩）；
消费者	调整 concurrency（并发数）、max-poll-records（批量大小）；
服务端	分区数建议等于消费者并发数，副本数≥2（生产环境）；

---

8. 常见问题及解决方案

问题现象	排查方向
连接 Kafka 失败	1. 检查 bootstrap-servers 地址/端口； 2. 确认 SASL 账号密码； 3. 测试网络连通性（telnet www.huerpu.cc 9092）；
JSON 反序列化失败	1. 实体类添加无参构造； 2. trusted.packages 包含实体包名； 3. 检查消息格式；
消费后消息重复	1. 确保手动提交仅在业务成功后执行； 2. 消费端实现幂等逻辑；
消费阻塞	1. 检查死信队列配置； 2. 优化业务逻辑执行时间； 3. 调整批量消费大小；

---

9. Kafka服务器端配置

9.1 核心前提：SASL 认证配置

Kafka 服务端需开启 SASL_PLAINTEXT 认证，否则客户端无法正常连接。

9.1.1 服务端 server.properties 配置

# 启用SASL认证
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://www.huerpu.cc:9092

# SASL机制
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN

# JAAS 认证配置
listener.name.sasl_plaintext.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
    username="kelsen" \
    password="k1029384756" \
    user_kelsen="k1029384756";

9.1.2 验证 SASL 配置生效

重启 Kafka 后，检查日志确认监听正常：

# 查看启动日志
grep "SASL_PLAINTEXT" $KAFKA_HOME/logs/server.log

日志中出现「Started SASL_PLAINTEXT listener on port 9092」即生效。

9.2 创建核心 Topic

9.2.1 准备客户端 JAAS 配置文件

创建 kafka_client_jaas.conf：

KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kelsen"
    password="k1029384756";
};

9.2.2 创建业务 Topic

# 加载JAAS配置（Linux/macOS）
export KAFKA_OPTS="-Djava.security.auth.login.config=/path/to/kafka_client_jaas.conf"

# 创建业务Topic
$KAFKA_HOME/bin/kafka-topics.sh \
    --create \
    --topic hep-test-topic \
    --bootstrap-server www.huerpu.cc:9092 \
    --partitions 3 \
    --replication-factor 2 \
    --config cleanup.policy=delete \
    --config retention.ms=86400000 \
    --config max.message.bytes=1048576

# 创建死信Topic
$KAFKA_HOME/bin/kafka-topics.sh \
    --create \
    --topic hep-dead-letter-topic \
    --bootstrap-server www.huerpu.cc:9092 \
    --partitions 1 \
    --replication-factor 2 \
    --config retention.ms=604800000

9.3 权限配置

授予 kelsen 用户 Topic 读写权限：

# 业务Topic权限
$KAFKA_HOME/bin/kafka-acls.sh \
    --bootstrap-server www.huerpu.cc:9092 \
    --command-config $KAFKA_HOME/config/client.properties \
    --add \
    --allow-principal User:kelsen \
    --operation Read \
    --operation Write \
    --topic hep-test-topic

# 死信Topic权限
$KAFKA_HOME/bin/kafka-acls.sh \
    --bootstrap-server www.huerpu.cc:9092 \
    --command-config $KAFKA_HOME/config/client.properties \
    --add \
    --allow-principal User:kelsen \
    --operation Read \
    --operation Write \
    --topic hep-dead-letter-topic

# 消费者组权限
$KAFKA_HOME/bin/kafka-acls.sh \
    --bootstrap-server www.huerpu.cc:9092 \
    --command-config $KAFKA_HOME/config/client.properties \
    --add \
    --allow-principal User:kelsen \
    --operation Read \
    --group hep-kafka-group

9.4 验证 Topic 配置

# 列出所有Topic
$KAFKA_HOME/bin/kafka-topics.sh \
    --list \
    --bootstrap-server www.huerpu.cc:9092 \
    --command-config $KAFKA_HOME/config/client.properties

# 查看Topic详情
$KAFKA_HOME/bin/kafka-topics.sh \
    --describe \
    --topic hep-test-topic \
    --bootstrap-server www.huerpu.cc:9092 \
    --command-config $KAFKA_HOME/config/client.properties

---

总结

1. 核心配置：SASL 认证是连接 Kafka 的关键，需确保客户端/服务端账号、认证方式一致；
2. 企业级特性：死信队列、手动提交 offset、批量消费、异常重试是生产环境必备；
3. 关键规范：消息实体序列化、幂等性处理、权限配置可避免大部分生产问题；
4. 测试验证：通过 HTTP 接口测试生产者，控制台日志验证消费者，确保端到端流程正常。