springboot kafka怎样实现消息消费监控

在Spring Boot中，我们可以使用Kafka的消费者监听器来监控消息消费情况。以下是实现消息消费监控的步骤：

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


    org.springframework.kafka
    spring-kafka

创建一个Kafka消费者配置类，例如KafkaConsumerConfig.java，并配置消费者属性，如Bootstrap服务器、消费者组ID等：

@Configuration
public class KafkaConsumerConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    @Bean
    public Map consumerConfigs() {
        Map props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        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);
        return props;
    }
}

创建一个Kafka消费者监听器接口，例如KafkaMessageListener.java：

public interface KafkaMessageListener {
    void onMessage(String message);
}

实现KafkaMessageListener接口，例如MyKafkaMessageListener.java：

@Service
public class MyKafkaMessageListener implements KafkaMessageListener {
    @Override
    public void onMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

创建一个Kafka消费者配置类，例如KafkaConsumerConfig.java，并配置消费者属性，如Bootstrap服务器、消费者组ID等：

@Configuration
public class KafkaConsumerConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    @Bean
    public Map consumerConfigs() {
        Map props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        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);
        return props;
    }
    @Bean
    public ConsumerFactory consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }
    @Bean
    public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

创建一个Kafka消费者监听器，例如KafkaConsumerListener.java：

@Service
public class KafkaConsumerListener {
    @Autowired
    private MyKafkaMessageListener myKafkaMessageListener;
    @KafkaListener(topics = "${spring.kafka.consumer.topic}", groupId = "${spring.kafka.consumer.group}")
    public void listen(ConsumerRecord record) {
        myKafkaMessageListener.onMessage(record.value());
    }
}

在application.properties或application.yml文件中配置Kafka消费者属性，如主题、组ID等：

spring.kafka.consumer.topic=my-topic
spring.kafka.consumer.group=my-consumer-group
spring.kafka.bootstrap-servers=localhost:9092

现在，当你的应用程序接收到Kafka消息时，MyKafkaMessageListener中的onMessage方法将被调用，你可以在该方法中实现消息消费监控的逻辑。