springboot+kafka

一、Spring Boot 2.7.18 + JDK 8 集成 Kafka Demo

1.1 Maven 依赖

Spring Boot 2.7.18 的父工程已管理 spring-kafka 版本（2.8.11），底层 Kafka 客户端为 3.1.2，兼容 JDK 8。

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.7.18</version>
</parent>

<dependencies>
    <!-- Web 模块（用于提供 REST 接口发送消息） -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

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

    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
</dependencies>

1.2 application.yml 配置

spring:
  kafka:
    bootstrap-servers: localhost:9092

    # ========== 生产者配置 ==========
    producer:
      acks: all                       # 所有 ISR 副本确认
      batch-size: 16384
      buffer-memory: 33554432
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      retries: 3

    # ========== 消费者配置 ==========
    consumer:
      group-id: demo-group
      auto-offset-reset: earliest     # 无偏移量时从最早开始
      enable-auto-commit: false       # 关闭自动提交，配合手动 ack
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      max-poll-records: 50            # 一次 poll 最多拉取条数

    # ========== 监听器配置 ==========
    listener:
      ack-mode: manual_immediate      # 手动立即提交
      concurrency: 1                  # 消费者并发数
      type: single                    # single / batch

# 自定义 Topic 名称
kafka:
  topic:
    demo: spring-kafka-demo-topic

1.3 自动创建 Topic（可选）

Spring Boot 启动时会自动创建 NewTopic Bean，若 Topic 已存在则忽略。

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfig {

    @Bean
    public NewTopic demoTopic() {
        return TopicBuilder.name("spring-kafka-demo-topic")
                .partitions(3)
                .replicas(1)
                .build();
    }
}

1.4 消息发送端（生产者）

KafkaTemplate 由 Spring Boot 自动配置，直接注入使用。

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;

@Slf4j
@RestController
public class KafkaProducerController {

    private static final String TOPIC = "spring-kafka-demo-topic";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 同步发送，等待结果
     */
    @PostMapping("/kafka/send")
    public String send(@RequestParam String message) {
        try {
            SendResult<String, String> result =
                    kafkaTemplate.send(TOPIC, message).get(3, TimeUnit.SECONDS);
            log.info("发送成功 topic={} partition={} offset={} value={}",
                    result.getRecordMetadata().topic(),
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset(),
                    message);
            return "发送成功，offset: " + result.getRecordMetadata().offset();
        } catch (Exception e) {
            log.error("发送失败", e);
            return "发送失败: " + e.getMessage();
        }
    }

    /**
     * 异步发送，带回调
     */
    @PostMapping("/kafka/sendAsync")
    public String sendAsync(@RequestParam String message) {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(TOPIC, message);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                log.info("异步发送成功，offset: {}", result.getRecordMetadata().offset());
            }
            @Override
            public void onFailure(Throwable ex) {
                log.error("异步发送失败", ex);
            }
        });
        return "消息已异步提交";
    }

    /**
     * 带 key 发送，相同 key 进入同一分区
     */
    @PostMapping("/kafka/sendWithKey")
    public String sendWithKey(@RequestParam String key, @RequestParam String message) {
        kafkaTemplate.send(TOPIC, key, message);
        return "已发送，key=" + key;
    }
}

1.5 消息监听端（消费者）

逐条消费（默认）

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.Component;

@Slf4j
@Component
public class KafkaConsumerListener {

    @KafkaListener(topics = "${kafka.topic.demo}", groupId = "demo-group")
    public void listen(ConsumerRecord<String, String> record, Acknowledgment ack) {
        log.info("收到消息 | topic: {} | partition: {} | offset: {} | key: {} | value: {}",
                record.topic(), record.partition(), record.offset(),
                record.key(), record.value());
        try {
            // 业务处理
            processMessage(record.value());
            ack.acknowledge();          // 手动提交
        } catch (Exception e) {
            log.error("处理失败，不提交偏移量", e);
            // 不提交，等待重试
        }
    }

    private void processMessage(String message) {
        log.info("业务处理中：{}", message);
    }
}

简洁写法（自动提交）

@KafkaListener(topics = "${kafka.topic.demo}", groupId = "demo-group")
public void listenSimple(String message) {
    log.info("收到消息: {}", message);
}

1.6 自定义监听器容器工厂（高级）

若需细粒度控制（并发数、错误处理器、死信队列），可自定义 ConcurrentKafkaListenerContainerFactory。注意：自定义工厂会覆盖自动配置，yml 中的 listener 配置将失效。

import org.apache.kafka.common.TopicPartition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaListenerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String>
            kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory,
                                          KafkaTemplate<String, String> kafkaTemplate) {

        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(3);                       // 并发消费者数

        // 错误处理：重试 3 次，间隔 2 秒，超过后发送到 .DLT 死信 Topic
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
                kafkaTemplate,
                (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition())
        );
        factory.setCommonErrorHandler(new DefaultErrorHandler(recoverer,
                new FixedBackOff(2000L, 3)));

        return factory;
    }
}

1.7 启动类

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

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

1.8 验证步骤

启动 ZooKeeper 与 Kafka（默认端口 9092）。
启动 Spring Boot 应用。
发送消息：

curl -X POST "http://localhost:8080/kafka/send?message=HelloKafka"

二、Kafka 核心概念详解

2.1 Partition（分区）

partitions(3) 表示 Topic 划分为 3 个分区，编号 0、1、2。

物理存储：每个分区对应 Broker 上的一个日志目录。
副本：每个分区可有多个副本，一个 Leader 多个 Follower。
有序性：仅保证同一分区内消息有序，不保证 Topic 全局有序。
并行度上限：同一消费者组内，一个分区只能被一个消费者线程消费，因此最大有效并行数 = 分区数。
吞吐量：分区越多，理论吞吐越高，但元数据开销、文件句柄、Rebalance 时间也增加。
修改：只能增加分区，不能减少。增加后 hash(key) % partitions 可能改变，破坏相同 key 的顺序性。

分区数建议

场景	建议 partitions
本地开发测试	1~3
小规模生产	6~12
高吞吐生产	按吞吐量与消费者数计算，预留 2~3 倍
要求全局顺序	1
要求相同 key 顺序	按 key 基数与吞吐评估

2.2 消费者组与消费模式

核心规则：同一个消费者组内，一条消息只会被一个消费者实例消费；不同消费者组各自独立消费全量消息。

A/B 的 group-id	同一条消息
相同	只被 A 或 B 其中一个消费（竞争消费）
不同	A 和 B 都会消费一次（广播）

示例：

A、B 均配置 group-id: demo-group → 一条消息只被其中一个处理。
A 配置 group-a，B 配置 group-b → 两个服务都收到同一条消息。

与分区的关系：同组内，分区会按 Rebalance 分配给各消费者。例如 3 个分区、2 个消费者，可能 A 拿 2 个分区，B 拿 1 个。

2.3 消息 Key 与分区策略

Key 是消息的可选字段，与 value 一起构成 ProducerRecord(topic, partition, key, value)。

发送时传入 Key

java

// 无 key：粘性分区（Kafka 2.4+ 默认），尽量均衡
kafkaTemplate.send(topic, value);

// 有 key：hash(key) % partitions
kafkaTemplate.send(topic, key, value);

// 指定分区：绕过 key 计算
kafkaTemplate.send(topic, partition, key, value);

Key 的选择原则

业务场景	推荐 key
订单状态变更	`orderId`
用户行为日志	`userId`
设备上报	`deviceId`
银行交易	`accountNo`
无需保序	不传 key

需要保序的维度做 key，相同 key 进同一分区，保证分区内有序。
key 基数要足够大，避免数据倾斜（如用 province 做 key 只有 30 多个值，会导致热点）。
不要用随机值做 key，否则失去保序意义且效率低。

消费端获取 Key

java

@KafkaListener(topics = "wss-kafka-test-topic", groupId = "wss-group")
public void listen(ConsumerRecord<String, String> record, Acknowledgment ack) {
    String key = record.key();
    String value = record.value();
    int partition = record.partition();
    long offset = record.offset();
    // ...
    ack.acknowledge();
}

自定义分区器

实现 org.apache.kafka.clients.producer.Partitioner，并在配置中指定：

yaml

spring:
  kafka:
    producer:
      properties:
        partitioner.class: com.example.MyPartitioner

三、批量消费（Batch Listener）

3.1 开启方式

方式一：application.yml（自动配置工厂）

yaml

spring:
  kafka:
    consumer:
      max-poll-records: 50          # 一次 poll 最多拉取 50 条
    listener:
      type: batch                   # 开启批量模式
      ack-mode: manual_immediate
      concurrency: 1

方式二：自定义容器工厂（必须显式设置）

java

factory.setBatchListener(true);
factory.getContainerProperties().setAckMode(
        ContainerProperties.AckMode.MANUAL_IMMEDIATE);

⚠️ 若自定义了 ConcurrentKafkaListenerContainerFactory，yml 中的 listener.type: batch 不生效，必须 setBatchListener(true)。

3.2 监听方法签名

java

@KafkaListener(topics = "wss-kafka-test-topic", groupId = "wss-group")
public void listen(List<ConsumerRecord<String, String>> records,
                   Acknowledgment ack) {
    log.info("本批收到 {} 条消息", records.size());
    for (ConsumerRecord<String, String> record : records) {
        log.info("partition={}, offset={}, key={}, value={}",
                record.partition(), record.offset(), record.key(), record.value());
    }
    // 业务处理
    ack.acknowledge();   // 提交本批所有分区的最高 offset
}

也可直接接收 List<String> messages。

3.3 注意事项

max-poll-records: 50 表示最多 50 条，实际条数取决于积压量、分区数、fetch.min.bytes 等。
批量模式下 acknowledge() 提交整批，若某条失败且仍提交，则整批视为已消费，无法重试。建议处理失败时抛异常，让错误处理器重试或转死信。
单批处理时间不宜过长，避免超过 max.poll.interval.ms（默认 5 分钟）触发 Rebalance。
concurrency × max-poll-records 控制峰值内存。

四、Kafka 与 RabbitMQ 模型对比

RabbitMQ	Kafka	说明
Exchange	❌ 无	Kafka 没有交换机
Binding	❌ 无	无绑定规则
Routing Key	⚠️ 部分对应消息 key	key 只用于选分区，不做路由匹配
Queue	⚠️ 部分对应 Topic + Partition	Topic 是逻辑队列，Partition 是物理存储
Consumer	Consumer	消费者
—	Consumer Group	RabbitMQ 无直接对应，靠多 Queue 模拟

Kafka 替代 Exchange 功能的机制

分发：靠 Consumer Group。相同 group-id 竞争消费，不同 group-id 广播。
路由：靠多个 Topic + 业务约定，生产端决定发往哪个 Topic；消费端无法按规则过滤，需业务层自行判断。

Kafka 的“路由”动作只有：生产者决定消息发到哪个 Topic、哪个 Partition。

五、Kafka 管理工具

5.1 开源 Web UI

工具	特点	协议
UI for Apache Kafka (Provectus)	功能全面，多集群、Topic 管理、消息浏览、消费者组 Lag、RBAC	免费开源
Kafbat UI	轻量快速，Provectus UI 社区分支	Apache 2.0
KnowStreaming	滴滴开源，运维管控、监控告警、资源治理	AGPL-3.0
kouncil	现代化界面，实时查看主题与消费者延迟	GPL-3.0
Kafka Manager	经典老牌，云服务商常集成	开源

5.2 桌面客户端

工具	特点
Offset Explorer（原 Kafka Tool）	功能强大，支持多种格式，浏览/消费/监控
Kafka-King	跨平台 GUI，现代化界面
Kafka Assistant	实时监控、拓扑图、指标监控

5.3 商业/云平台

Confluent Control Center
阿里云、华为云等 Kafka 控制台

5.4 命令行工具

bash

# 查看所有 Topic
kafka-topics.sh --bootstrap-server localhost:9092 --list

# 查看消费者组及 Lag
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group wss-group

# 查看 Topic 详情
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic wss-kafka-test-topic

# 消费消息（调试）
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic wss-kafka-test-topic --from-beginning

选型建议：

本地开发：Offset Explorer / Kafka-King
团队共享：UI for Apache Kafka / Kafbat UI
云 Kafka：云厂商控制台
临时排查：命令行

六、常见问题与总结

6.1 关键配置速查

配置项	作用	建议值
`spring.kafka.producer.acks`	消息确认级别	`all`
`spring.kafka.consumer.auto-offset-reset`	无偏移量策略	`earliest`
`spring.kafka.consumer.enable-auto-commit`	自动提交	`false`
`spring.kafka.listener.ack-mode`	偏移量提交模式	`manual_immediate`
`spring.kafka.listener.type`	监听类型	`single` / `batch`
`factory.setConcurrency(n)`	消费者并发数	≤ 分区数

6.2 常见坑

自定义容器工厂后 yml 配置失效：必须显式设置 setBatchListener(true)、setAckMode 等。
key 分布不均导致热点：低基数字段做 key 会使分区倾斜。
误以为相同 key 全局有序：仅同一分区内有序，增加分区后顺序可能被破坏。
无 key 时以为轮询：Kafka 2.4+ 默认粘性分区，非严格轮询。
批量消费异常粒度粗：默认整批重试，需业务层细化处理。
消费者处理超时：超过 max.poll.interval.ms 会触发 Rebalance。

6.3 一句话总结

Kafka 用 Topic + Partition + Consumer Group + 消息 key 构建了简洁而强大的消息模型，没有 RabbitMQ 的 Exchange/Binding/RoutingKey，但通过分区、消费者组和 key 设计，同样能实现竞争消费、广播、顺序保证和高吞吐。Spring Boot 通过自动配置和 KafkaTemplate、@KafkaListener 极大简化了集成，生产环境需关注分区规划、消费者组设计、手动提交与错误处理。

分享

springboot+kafka

一、Spring Boot 2.7.18 + JDK 8 集成 Kafka Demo

1.1 Maven 依赖

1.2 application.yml 配置

1.3 自动创建 Topic（可选）

1.4 消息发送端（生产者）

1.5 消息监听端（消费者）

1.6 自定义监听器容器工厂（高级）

1.7 启动类

1.8 验证步骤

二、Kafka 核心概念详解

2.1 Partition（分区）

2.2 消费者组与消费模式

2.3 消息 Key 与分区策略

三、批量消费（Batch Listener）

3.1 开启方式

3.2 监听方法签名

3.3 注意事项

四、Kafka 与 RabbitMQ 模型对比

五、Kafka 管理工具

5.1 开源 Web UI

5.2 桌面客户端

5.3 商业/云平台

5.4 命令行工具

六、常见问题与总结

6.1 关键配置速查

6.2 常见坑

6.3 一句话总结