springboot怎么配置集群的kafka

Spring Boot 配置集群 Kafka 主要有以下几个步骤：

1. 添加依赖：首先，你需要在项目的 `pom.xml` 或者 `build.gradle` 文件中添加 Spring Boot 对 Kafka 的支持，例如：

<!-- Maven -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-kafka</artifactId>
</dependency>

<!-- Gradle -->
implementation 'org.springframework.boot:spring-boot-starter-kafka'

2. 配置 Kafka 客户端：在 `application.properties` 或者 `application.yml` 中设置 Kafka 的连接信息，包括集群地址（bootstrap servers）：

spring.kafka.bootstrap-servers=localhost:9092,localhost:9093,localhost:9094

这里的 `localhost:9092,localhost:9093,localhost:9094` 是集群中的节点地址。

3. 创建消费者或生产者：使用 `@KafkaListener` 注解创建消费者，或者使用 `KafkaTemplate` 或 `ProducerFactory` 创建生产者。确保你指定了集群的配置：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;

public class MyConsumer {

    @KafkaListener(topics = "my-topic")
    public void consume(String message) {
        // ...
    }
    
}

// 生产者示例
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
    return new KafkaTemplate<>(kafkaSessionFactory());
}

4. 定义消费者组：Kafka 使用消费者组（consumer group）来划分消息消费任务。每个消费者组内的消费者会共享消息，所以需要在配置中指定消费者组名：

spring.kafka.consumer.group.id=my-consumer-group

5. 高可用性配置：如果需要更高级的配置，如自动发现集群节点、动态失败重试等，可以使用第三方库如 `spring-cloud-stream-kafka` 或 `spring-kafka-connector`。