springboot kafka如何進行消息監控

在Spring Boot應用中，可以使用Kafka的消費者監聽器來監控消息。以下是一個簡單的示例，展示了如何在Spring Boot應用中監控Kafka消息：

1. 首先，確保你的項目中已經添加了Kafka依賴。在pom.xml文件中添加以下依賴：

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

2. 在application.properties或application.yml文件中配置Kafka消費者屬性。例如：

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=my-consumer-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

3. 創建一個Kafka消費者監聽器接口，用于處理接收到的消息：

public interface KafkaMessageListener {
    void onMessage(ConsumerRecord<String, String> record);
}

4. 實現Kafka消費者監聽器接口，并在處理消息的方法中添加監控邏輯：

@Service
public class MyKafkaMessageListener implements KafkaMessageListener {

    @Override
    public void onMessage(ConsumerRecord<String, String> record) {
        // 在這里添加監控邏輯，例如記錄日志、發送通知等
        System.out.printf("接收到消息: key = %s, value = %s, partition = %d, offset = %d%n",
                record.key(), record.value(), record.partition(), record.offset());
    }
}

5. 創建一個Kafka消費者配置類，用于配置消費者監聽器：

@Configuration
public class KafkaConsumerConfig {

    @Autowired
    private MyKafkaMessageListener myKafkaMessageListener;

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        return props;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerConfigs(consumerConfigs());
        factory.setConcurrency(3); // 設置并發消費者數量
        factory.getContainerProperties().setAckMode(AckMode.MANUAL); // 設置手動提交偏移量
        return factory;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public KafkaListenerEndpointRegistrar endpointRegistrar(KafkaListenerEndpointRegistry registry) {
        KafkaListenerEndpointRegistrar registrar = new KafkaListenerEndpointRegistrar(registry);
        registrar.setBeanFactory(applicationContext);
        return registrar;
    }
}

6. 在需要監聽Kafka消息的類中，使用@KafkaListener注解來定義一個監聽器方法：

@Service
public class MyKafkaConsumer {

    @Autowired
    private MyKafkaMessageListener myKafkaMessageListener;

    @KafkaListener(topics = "my-topic", groupId = "my-consumer-group")
    public void listen(ConsumerRecord<String, String> record) {
        myKafkaMessageListener.onMessage(record);
    }
}

現在，當有新消息進入my-topic主題時，MyKafkaConsumer類中的listen方法將被調用，MyKafkaMessageListener接口的onMessage方法將處理接收到的消息，并執行監控邏輯。