springboot kafka如何进行消息监控

在Spring Boot应用中，可以使用Kafka的消费者监听器来监控消息。以下是一个简单的示例，展示了如何在Spring Boot应用中监控Kafka消息：

1. 首先，确保你的项目中已经添加了Kafka依赖。在pom.xml文件中添加以下依赖：

    org.springframework.kafka
    spring-kafka

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 record);
}

4. 实现Kafka消费者监听器接口，并在处理消息的方法中添加监控逻辑：

@Service
public class MyKafkaMessageListener implements KafkaMessageListener {

    @Override
    public void onMessage(ConsumerRecord record) {
        // 在这里添加监控逻辑，例如记录日志、发送通知等
        System.out.printf("接收到消息: key = %s, value = https://www.yisu.com/ask/%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 consumerConfigs() {
        Map 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 kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerConfigs(consumerConfigs());
        factory.setConcurrency(3); // 设置并发消费者数量
        factory.getContainerProperties().setAckMode(AckMode.MANUAL); // 设置手动提交偏移量
        return factory;
    }

    @Bean
    public ConsumerFactory 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 record) {
        myKafkaMessageListener.onMessage(record);
    }
}

现在，当有新消息进入my-topic主题时，MyKafkaConsumer类中的listen方法将被调用，MyKafkaMessageListener接口的onMessage方法将处理接收到的消息，并执行监控逻辑。