对于最新的稳定版本，请使用 Spring for Apache Kafka 3.3.0！spring-doc.cn

配置主题

如果您在应用程序上下文中定义了一个 Bean，它可以自动向代理添加主题。为此，您可以向应用程序上下文添加 for each 主题。版本 2.3 引入了一个新类，使创建此类 bean 更加方便。以下示例显示了如何执行此操作：KafkaAdminNewTopic@BeanTopicBuilderspring-doc.cn

Javaspring-doc.cn
Kotlinspring-doc.cn

@Bean
public KafkaAdmin admin() {
    Map<String, Object> configs = new HashMap<>();
    configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    return new KafkaAdmin(configs);
}

@Bean
public NewTopic topic1() {
    return TopicBuilder.name("thing1")
            .partitions(10)
            .replicas(3)
            .compact()
            .build();
}

@Bean
public NewTopic topic2() {
    return TopicBuilder.name("thing2")
            .partitions(10)
            .replicas(3)
            .config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
            .build();
}

@Bean
public NewTopic topic3() {
    return TopicBuilder.name("thing3")
            .assignReplicas(0, List.of(0, 1))
            .assignReplicas(1, List.of(1, 2))
            .assignReplicas(2, List.of(2, 0))
            .config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
            .build();
}

@Bean
fun admin() = KafkaAdmin(mapOf(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG to "localhost:9092"))

@Bean
fun topic1() =
    TopicBuilder.name("thing1")
        .partitions(10)
        .replicas(3)
        .compact()
        .build()

@Bean
fun topic2() =
    TopicBuilder.name("thing2")
        .partitions(10)
        .replicas(3)
        .config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
        .build()

@Bean
fun topic3() =
    TopicBuilder.name("thing3")
        .assignReplicas(0, Arrays.asList(0, 1))
        .assignReplicas(1, Arrays.asList(1, 2))
        .assignReplicas(2, Arrays.asList(2, 0))
        .config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
        .build()

从版本 2.6 开始，您可以省略 and/or，并且代理默认值将应用于这些属性。代理版本必须至少为 2.4.0 才能支持此功能 - 请参阅 KIP-464。partitions()replicas()spring-doc.cn

Javaspring-doc.cn
Kotlinspring-doc.cn

@Bean
public NewTopic topic4() {
    return TopicBuilder.name("defaultBoth")
            .build();
}

@Bean
public NewTopic topic5() {
    return TopicBuilder.name("defaultPart")
            .replicas(1)
            .build();
}

@Bean
public NewTopic topic6() {
    return TopicBuilder.name("defaultRepl")
            .partitions(3)
            .build();
}

@Bean
fun topic4() = TopicBuilder.name("defaultBoth").build()

@Bean
fun topic5() = TopicBuilder.name("defaultPart").replicas(1).build()

@Bean
fun topic6() = TopicBuilder.name("defaultRepl").partitions(3).build()

从版本 2.7 开始，你可以在单个 bean 定义中声明多个 s：NewTopicKafkaAdmin.NewTopicsspring-doc.cn

Javaspring-doc.cn
Kotlinspring-doc.cn

@Bean
public KafkaAdmin.NewTopics topics456() {
    return new NewTopics(
            TopicBuilder.name("defaultBoth")
                .build(),
            TopicBuilder.name("defaultPart")
                .replicas(1)
                .build(),
            TopicBuilder.name("defaultRepl")
                .partitions(3)
                .build());
}

@Bean
fun topics456() = KafkaAdmin.NewTopics(
    TopicBuilder.name("defaultBoth")
        .build(),
    TopicBuilder.name("defaultPart")
        .replicas(1)
        .build(),
    TopicBuilder.name("defaultRepl")
        .partitions(3)
        .build()
)

使用 Spring Boot 时，会自动注册一个 bean，因此您只需要（和/或）s。KafkaAdminNewTopicNewTopics@Bean

默认情况下，如果代理不可用，则会记录一条消息，但会继续加载上下文。您可以以编程方式调用 admin 的方法，以便稍后重试。如果您希望将此情况视为致命情况，请将 admin 的属性设置为。然后，上下文无法初始化。initialize()fatalIfBrokerNotAvailabletruespring-doc.cn

如果 broker 支持它（1.0.0 或更高版本），则如果发现现有主题的分区数少于 .NewTopic.numPartitions

从版本 2.7 开始，提供了在运行时创建和检查主题的方法。KafkaAdminspring-doc.cn

createOrModifyTopicsspring-doc.cn
describeTopicsspring-doc.cn

有关更高级的功能，您可以直接使用。以下示例显示了如何执行此操作：AdminClientspring-doc.cn

@Autowired
private KafkaAdmin admin;

...

    AdminClient client = AdminClient.create(admin.getConfigurationProperties());
    ...
    client.close();

从版本 2.9.10 和 3.0.9 开始，您可以提供一个，该版本可用于确定是否应考虑创建或修改特定 bean。例如，如果您有多个实例指向不同的集群，并且您希望选择应由每个管理员创建或修改的主题，则这非常有用。Predicate<NewTopic>NewTopicKafkaAdminspring-doc.cn

admin.setCreateOrModifyTopic(nt -> !nt.name().equals("dontCreateThisOne"));