springboot 整合 kafka 如何實現消息順序消費

在 Spring Boot 中整合 Kafka 并實現消息順序消費，可以通過以下幾種方法來實現：

1. 使用同一個消費者組：確保所有的消費者實例使用相同的消費者組 ID。這樣，Kafka 會自動將消息分發給消費者組中的一個消費者實例，從而實現負載均衡。在消費者實例中，可以使用 ConsumerGroupConfig 配置消費者組 ID。

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

2. 使用單線程消費者：在消費者實例中，使用單線程來處理消息。這樣可以確保消息按順序處理，因為每個消息只會被一個線程處理。在 Spring Boot 中，可以使用 @KafkaListener 注解來實現單線程消費者。

@KafkaListener(topics = "${kafka.topic}", groupId = "${kafka.groupId}")
public void listen(ConsumerRecord<String, MyMessage> record) {
    // 處理消息
}

3. 使用順序消費插件：Kafka 提供了一個順序消費插件，可以實現全局的順序消費。要使用這個插件，需要在 Kafka 配置文件中啟用它，并設置一個唯一的消費者組 ID。然后，在 Spring Boot 中，使用相同的消費者組 ID 來創建消費者實例。

首先，在 application.yml 或 application.properties 文件中啟用順序消費插件：

spring:
  kafka:
    consumer:
      group-id: my-group
      auto-offset-reset: earliest
      enable-auto-commit: false
      properties:
        isolation-level: read_committed

然后，在消費者實例中，使用相同的消費者組 ID 來創建消費者實例：

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

通過以上方法，可以在 Spring Boot 中整合 Kafka 并實現消息順序消費。