@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 使JsonDeserializer信任所有包
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        
        return new DefaultKafkaConsumerFactory<>(props);
    }

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

@Service
public class KafkaConsumerService {

    // 基本用法：監(jiān)聽單個(gè)主題
    @KafkaListener(topics = "test-topic", groupId = "test-group")
    public void listen(String message) {
        System.out.println("接收到消息：" + message);
    }
    
    // 監(jiān)聽多個(gè)主題
    @KafkaListener(topics = {"topic1", "topic2"}, groupId = "multi-topic-group")
    public void listenMultipleTopics(String message) {
        System.out.println("從多個(gè)主題接收到消息：" + message);
    }
    
    // 指定分區(qū)監(jiān)聽
    @KafkaListener(topicPartitions = {
        @TopicPartition(topic = "partitioned-topic", partitions = {"0", "1"})
    }, groupId = "partitioned-group")
    public void listenPartitions(String message) {
        System.out.println("從特定分區(qū)接收到消息：" + message);
    }
    
    // 使用ConsumerRecord獲取消息元數(shù)據(jù)
    @KafkaListener(topics = "metadata-topic", groupId = "metadata-group")
    public void listenWithMetadata(ConsumerRecord<String, String> record) {
        System.out.println("主題：" + record.topic() + 
                          "，分區(qū)：" + record.partition() +
                          "，偏移量：" + record.offset() +
                          "，鍵：" + record.key() +
                          "，值：" + record.value());
    }
    
    // 批量消費(fèi)
    @KafkaListener(topics = "batch-topic", groupId = "batch-group", 
                  containerFactory = "batchListenerFactory")
    public void listenBatch(List<String> messages) {
        System.out.println("接收到批量消息，數(shù)量：" + messages.size());
        messages.forEach(message -> System.out.println("批量消息：" + message));
    }
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> batchListenerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = 
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);  // 啟用批量監(jiān)聽
    factory.getContainerProperties().setPollTimeout(3000);  // 輪詢超時(shí)時(shí)間
    return factory;
}

// 配置消費(fèi)組屬性
@Bean
public ConsumerFactory<String, Object> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    // 基本配置
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    
    // 消費(fèi)組配置
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-application-group");
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);  // 禁用自動(dòng)提交
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);  // 單次輪詢最大記錄數(shù)
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);  // 會(huì)話超時(shí)時(shí)間
    props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 10000);  // 心跳間隔
    
    return new DefaultKafkaConsumerFactory<>(props);
}

// 消費(fèi)者1
@KafkaListener(topics = "shared-topic", groupId = "shared-group")
public void consumer1(String message) {
    System.out.println("消費(fèi)者1接收到消息：" + message);
}

// 消費(fèi)者2
@KafkaListener(topics = "shared-topic", groupId = "shared-group")
public void consumer2(String message) {
    System.out.println("消費(fèi)者2接收到消息：" + message);
}

@Configuration
public class ManualCommitConfig {
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> manualCommitFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }
}

@Service
public class ManualCommitService {
    
    @KafkaListener(topics = "manual-commit-topic", 
                  groupId = "manual-group",
                  containerFactory = "manualCommitFactory")
    public void listenWithManualCommit(String message, Acknowledgment ack) {
        try {
            System.out.println("處理消息：" + message);
            // 處理消息的業(yè)務(wù)邏輯
            // ...
            // 成功處理后確認(rèn)消息
            ack.acknowledge();
        } catch (Exception e) {
            // 異常處理，可以選擇不確認(rèn)
            System.err.println("消息處理失?。? + e.getMessage());
        }
    }
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> retryListenerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = 
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    
    // 配置重試
    factory.setRetryTemplate(retryTemplate());
    
    // 配置恢復(fù)回調(diào)
    factory.setRecoveryCallback(context -> {
        ConsumerRecord<String, String> record = 
            (ConsumerRecord<String, String>) context.getAttribute("record");
        System.err.println("重試失敗，發(fā)送到死信隊(duì)列：" + record.value());
        // 可以將消息發(fā)送到死信主題
        // kafkaTemplate.send("dead-letter-topic", record.value());
        return null;
    });
    
    return factory;
}

private RetryTemplate retryTemplate() {
    RetryTemplate template = new RetryTemplate();
    
    // 固定間隔重試策略
    FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
    backOffPolicy.setBackOffPeriod(1000);  // 1秒重試間隔
    template.setBackOffPolicy(backOffPolicy);
    
    // 簡(jiǎn)單重試策略
    SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
    retryPolicy.setMaxAttempts(3);  // 最大重試次數(shù)
    template.setRetryPolicy(retryPolicy);
    
    return template;
}

@KafkaListener(topics = "retry-topic", groupId = "retry-group", 
               containerFactory = "retryListenerFactory")
public void listenWithRetry(String message) {
    System.out.println("接收到需要重試處理的消息：" + message);
    // 模擬處理失敗
    if (message.contains("error")) {
        throw new RuntimeException("處理失敗，將重試");
    }
    System.out.println("消息處理成功");
}