跳到主要内容
版本：4.1.1

配置主题

Doubao 中英对照 Configuring Topics

如果你在应用程序上下文中定义了一个 KafkaAdmin bean，它可以自动向代理添加主题。为此，你可以为每个主题向应用程序上下文添加一个 NewTopic @Bean。2.3 版本引入了一个新类 TopicBuilder，使此类 bean 的创建更加方便。以下示例展示了如何做到这一点：

@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();
}

从 2.6 版本开始，你可以省略 partitions() 和/或 replicas()，这些属性将应用代理默认值。代理版本必须至少为 2.4.0 才支持此特性——请参阅 KIP-464。

@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();
}

从 2.7 版本开始，你可以在单个 KafkaAdmin.NewTopics bean 定义中声明多个 NewTopic：

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

使用 Spring Boot 时，会自动注册一个 KafkaAdmin bean，因此你只需要 NewTopic（和/或 NewTopics）@Bean。

默认情况下，如果代理不可用，会记录一条消息，但上下文会继续加载。你可以以编程方式调用 admin 的 initialize() 方法，以便稍后重试。如果你希望将此条件视为致命错误，请将 admin 的 fatalIfBrokerNotAvailable 属性设置为 true。随后上下文将初始化失败。

备注

如果代理支持（1.0.0 或更高），当发现某个已存在主题的分区数少于 NewTopic.numPartitions 时，admin 会增加分区数。

从 2.7 版本开始，KafkaAdmin 提供了在运行时创建和查看主题的方法。从 4.0 版本开始，它还提供了删除主题的方法。

  • createOrModifyTopics
  • describeTopics
  • deleteTopics（自 4.0 起）

对于更高级的管理功能，你可以直接使用 AdminClient。以下示例展示了如何做到这一点：

@Autowired
private KafkaAdmin admin;

...

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

从 2.9.10、3.0.9 版本开始，你可以提供一个 Predicate<NewTopic>，用于判断某个特定的 NewTopic bean 是否应被纳入创建或修改。例如，如果你有多个指向不同集群的 KafkaAdmin 实例，并希望选择应由每个 admin 创建或修改的那些主题，这会很有用。

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