如何让Spring Boot项目自动创建所需的Kafka Topic

有两种常用的实现方式，具体操作如下：

方法1：注册NewTopic Bean实现（推荐，可控性高） #

该方案可以精准控制每个Topic的分区数、副本数、消息保留策略等配置，不会覆盖已存在的同名Topic配置，安全性高。

• 第一步：确保项目已引入Spring Kafka依赖，Maven配置示例：

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

• 第二步：在配置文件中添加Kafka服务地址，application.yml示例：

spring:
  kafka:
    bootstrap-servers: localhost:9092

• 第三步：编写Topic配置类，将需要创建的Topic注册为NewTopic类型的Bean，项目启动时Spring会自动检测目标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 {
    // 示例1：创建订单通知Topic，2个分区，1个副本，消息保留7天
    @Bean
    public NewTopic orderNotifyTopic() {
        return TopicBuilder.name("order_notify")
                .partitions(2)
                .replicas(1)
                .config("retention.ms", "604800000")
                .build();
    }

    // 示例2：创建支付通知Topic，1个分区，1个副本
    @Bean
    public NewTopic payNotifyTopic() {
        return TopicBuilder.name("pay_notify")
                .partitions(1)
                .replicas(1)
                .build();
    }
}

如果是多模块项目，可将该配置类放在公共依赖模块中，所有用到Kafka的业务模块引入公共模块即可，避免重复配置

方法2：开启默认自动创建配置（简单但可控性低） #

如果不需要自定义Topic参数，可以用该方案快速实现：

• Kafka服务端默认已开启auto.create.topics.enable=true配置，当生产者发送消息、消费者监听某个不存在的Topic时，Kafka会自动用默认参数创建Topic（默认1分区、1副本）
• 可在Spring配置中添加如下配置，避免项目启动时因Topic不存在报错：

spring:
  kafka:
    listener:
      missing-topics-fatal: false

注意事项 #

• 副本数配置不要超过Kafka集群的Broker节点数，本地单节点Kafka副本数设置为1即可，否则会创建失败
• 已存在的同名Topic不会被自动修改配置，不会影响已有数据

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