package com.tuoheng.common.config.kafka;

import com.tuoheng.common.utils.JacksonUtil;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * kafka配置类
 *
 * @author 拓恒
 * @since 2022-03-07
 */
@Configuration
@EnableKafka
@AutoConfigureAfter(KafkaPropeties.class)
@ConditionalOnProperty(name = KafkaPropeties.SPRING_KAFKA_ENABLE, havingValue = KafkaPropeties.TRUE)
public class KafkaConfiguration {
    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConfiguration.class);
    //kafkaTemplate实现了Kafka发送接收等功能
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {
        KafkaTemplate template = new KafkaTemplate<>(producerFactory);
        return template;
    }

//    @Bean
//    public KafkaTransactionManager transactionManager(ProducerFactory<String, String> producerFactory) {
//        KafkaTransactionManager manager = new KafkaTransactionManager(producerFactory);
//        return manager;
//    }

    /**
     * 根据senderProps填写的参数创建生产者工厂
     */
    @Bean
    public ProducerFactory<String, String> producerFactory(KafkaPropeties kafkaPropeties) {
        DefaultKafkaProducerFactory defaultKafkaProducerFactory = new DefaultKafkaProducerFactory<>(senderProps(kafkaPropeties));
        return defaultKafkaProducerFactory;
    }

    // 生产者配置
    private Map<String, Object> senderProps (KafkaPropeties kafkaPropeties){
        Map<String, Object> props = new HashMap<>();
        // 连接地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaPropeties.getBootstrapServersConfig());
        // ack配置
        props.put(ProducerConfig.ACKS_CONFIG, kafkaPropeties.getProducerAckConfig());
        // 重试次数
        props.put(ProducerConfig.RETRIES_CONFIG, kafkaPropeties.getProducerRetries());
        // 批量发送，延迟为1毫秒，启用该功能能有效减少生产者发送消息次数，从而提高并发量
        props.put(ProducerConfig.LINGER_MS_CONFIG, kafkaPropeties.getProducerLingerMs());
        // 启用幂等性，保证数据失败重试不重复，启用幂等性ack必须设置为-1
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, kafkaPropeties.getProducerEnableIdempotence());
        // 键的序列化方式
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 值的序列化方式
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 设置事务前缀
//        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, kafkaPropeties.getProducerTransactionIdPrefix());
        return props;
    }

    // 批量监听消费
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // 设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        // factory.setBatchListener(true);
        // 设置手动提交ackMode
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }

    // 根据consumerProps填写的参数创建消费者工厂
    @Bean
    public ConsumerFactory<String, String> consumerFactory(KafkaPropeties kafkaPropeties) {
        return new DefaultKafkaConsumerFactory<>(consumerProps(kafkaPropeties));
    }

    //消费者配置参数
    private Map<String, Object> consumerProps(KafkaPropeties kafkaPropeties) {
        Map<String, Object> props = new HashMap<>();
        // 连接地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaPropeties.getBootstrapServersConfig());
        // 是否自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaPropeties.getConsumerEnableAutoCommit());
        // offset策略
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaPropeties.getConsumerAutoOffsetReset());
        // 最大拉取条数
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaPropeties.getConsumerMaxPollRecords());
        // 设置事务读取隔离级别为已提交读
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, kafkaPropeties.getIsolationLevel());
        // 是否可以自动创建topic
        props.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, kafkaPropeties.getAutoCreateTopics());
        //键的反序列化方式
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        //值的反序列化方式
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
        return new ConsumerAwareListenerErrorHandler() {
            @Override
            public Object handleError(Message<?> message, ListenerExecutionFailedException e, Consumer<?, ?> consumer) {
                MessageHeaders headers = message.getHeaders();
                List<String> topics = headers.get(KafkaHeaders.RECEIVED_TOPIC, List.class);
                List<Integer> partitions = headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, List.class);
                List<Long> offsets = headers.get(KafkaHeaders.OFFSET, List.class);
                LOGGER.error("消息监听失败：message:{} | topics:{} | partition:{} | offsets:{}", JacksonUtil.obj2json(message.getPayload()),
                        JacksonUtil.obj2json(topics), JacksonUtil.obj2json(partitions), JacksonUtil.obj2json(offsets), e);
                return null;
            }
        };
    }
}
