springboot集成kafka使用

要在Spring Boot中集成Kafka，可以按照以下步骤进行操作：

1. 添加Kafka依赖：在`pom.xml`文件中添加以下依赖项：

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

2. 配置Kafka连接信息：在`application.properties`文件中添加以下配置：

spring.kafka.bootstrap-servers=localhost:9092     # Kafka服务器地址

3. 创建生产者：使用`KafkaTemplate`类创建一个生产者，并使用`send()`方法将消息发送到指定的主题。

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

public void sendMessage(String topic, String message) {
    kafkaTemplate.send(topic, message);
}

4. 创建消费者：使用`@KafkaListener`注解来创建一个消费者，指定要监听的主题以及处理接收到消息的方法。

@KafkaListener(topics = "myTopic")
public void receiveMessage(String message) {
    // 处理接收到的消息
    System.out.println("Received message: " + message);
}

5. 配置Kafka消费者：可以在Spring Boot应用程序的配置类中添加以下配置，以更改默认的消费者配置。

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "myGroup");
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}