package com.muyu.data.config;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.transaction.KafkaTransactionManager;

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

/**
 * 主题生产者的配置类
 */
@Configuration
public class KafkaProviderConfig {

    /**
     * kafka 的主机地址
     */
    @Value("${spring.kafka.producer.bootstrap-servers}")
    private String bootstrapServers;
    /**
     * 配置 Kafka的事务
     */
    @Value("${spring.kafka.producer.transaction-id-prefix}")
    private String transactionIdPrefix;
    /**
     * 发送确认机制
     */
    @Value("${spring.kafka.producer.acks}")
    private String acks;
    /**
     * 发送重试
     */
    @Value("${spring.kafka.producer.retries}")
    private String retries;
    /**
     * 发送消息的批次大小
     */
    @Value("${spring.kafka.producer.batch-size}")
    private String batchSize;
    /**
     * 消息的缓冲区内存大小
     */
    @Value("${spring.kafka.producer.buffer-memory}")
    private String bufferMemory;

    /**
     * 设置 健的序列化方式
     */
    @Value("${spring.kafka.producer.key-serializer}")
    private String keySerializer;

    /**
     * 设置 值的序列化方式
     */
    @Value("${spring.kafka.producer.value-serializer}")
    private String valueSerializer;

    /**
     * 构建 map 配置消息生产者对象的配置
     * @return
     */
    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>(16);
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        //acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
        //acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
        //acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
        //开启事务必须设为all
        props.put(ProducerConfig.ACKS_CONFIG, acks);
        //发生错误后，消息重发的次数，开启事务必须大于0
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        //当多个消息发送到相同分区时,生产者会将消息打包到一起,以减少请求交互. 而不是一条条发送
        //批次的大小可以通过batch.size 参数设置.默认是16KB
        //较小的批次大小有可能降低吞吐量（批次大小为0则完全禁用批处理）。
        //比如说，kafka里的消息5秒钟Batch才凑满了16KB，才能发送出去。那这些消息的延迟就是5秒钟
        //实测batchSize这个参数没有用
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        //有的时刻消息比较少,过了很久,比如5min也没有凑够16KB,这样延时就很大,所以需要一个参数. 再设置一个时间,到了这个时间,
        //即使数据没达到16KB,也将这个批次发送出去
        props.put(ProducerConfig.LINGER_MS_CONFIG, "5000");
        //生产者内存缓冲区的大小
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        //反序列化，和生产者的序列化方式对应
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
        return props;
    }

    /**
     * 构建 主题生产者工厂
     * @return
     */
    @Bean
    public ProducerFactory<Object, Object> producerFactory() {
        DefaultKafkaProducerFactory<Object, Object> factory = new DefaultKafkaProducerFactory<>(producerConfigs());
        //开启事务，会导致 LINGER_MS_CONFIG 配置失效
        factory.setTransactionIdPrefix(transactionIdPrefix);
        return factory;
    }

    /**
     * 配置 Kafka的事务管理器
     * @param producerFactory
     * @return
     */
    @Bean
    public KafkaTransactionManager<Object, Object> kafkaTransactionManager(ProducerFactory<Object, Object> producerFactory) {
        return new KafkaTransactionManager<>(producerFactory);
    }

    /**
     * 构建 KafkaTemplate
     * @return
     */
    @Bean
    public KafkaTemplate<Object, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

}
