package com.muyu.common.kafka.config;

import com.muyu.common.core.constant.KafkaConstants;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;

/**
 * kafka消费者配置信息
 * @Author：杨鹏
 * @Package：com.muyu.common.kafka.config
 * @Project：car-cloud-server
 * @name：KafkaConsumerConfig
 * @Date：2024/9/28 17:26
 */
@Configuration
public class KafkaConsumerConfig {

    /**
     * 服务端id+端口号
     */
    @Value("@{spring.kafka.consumer.bootstrap-servers}")
    private String bootstrapServers;

    /**
     * 偏移量
     */
//    @Value("{spring.kafka.consumer.enable-auto-commit}")
//    public boolean enableAutoCommit;

    /**
     * 自动提交时间间隔
     */
//    @Value("{spring.kafka.consumer.auto-commit-interval}")
//    private Integer autoCommitInterval;

    /**
     * 偏移量配置
     */
    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;

    /**
     * 堵塞最大时间
     */
//    @Value("{spring.kafka.consumer.fetch-max-wait}")
//    private Integer fetchMaxWait;

    /**
     * 请求最小字节
     */
//    @Value("${spring.kafka.consumer.fetch-min-size}")
//    private Integer fetchMinSize;

    /**
     * 心跳间隔时间
     */
//    @Value("${spring.kafka.consumer.heartbeat-interval}")
//    private Integer heartbeatInterval;

    /**
     * 最大记录条数
     */
//    @Value("${spring.kafka.consumer.max-poll-records")
//    private Integer maxPollRecords;

    /**
     * 消费组
     */
    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Bean
    public KafkaConsumer kafkaConsumer() {
        HashMap<String, Object> configs = new HashMap<>();
        configs.put(KafkaConstants.BOOTSTRAP_SERVERS, "111.231.50.146:9092");
        configs.put(KafkaConstants.ENABLE_AUTO_COMMIT, false);
        configs.put(KafkaConstants.AUTO_COMMIT_INTERVAL, 5000);
        configs.put(KafkaConstants.AUTO_OFFSET_RESET, autoOffsetReset);
        configs.put(KafkaConstants.FETCH_MAX_WAIT, 500);
        configs.put(KafkaConstants.FETCH_MIN_SIZE, 1);
        configs.put(KafkaConstants.HEARTBEAT_INTERVAL, 3000);
        configs.put(KafkaConstants.MAX_POLL_RECORDS, 500);
        configs.put(KafkaConstants.GROUP_ID, groupId);
        StringDeserializer keyDeserializer = new StringDeserializer();
        StringDeserializer valueDeserializer = new StringDeserializer();
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(configs, keyDeserializer, valueDeserializer);
        return kafkaConsumer;
    }
}
