package com.muyu.kafkaconfig;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;

/**
 * @Author：张腾
 * @Package：com.muyu.carData.config.kafkaconfig
 * @Project：cloud-server-8
 * @name：KafkaConfig
 * @Date：2024/9/28 12:19
 */
@Configuration
public class
KafkaConfig {

    @Bean
    public KafkaProducer kafkaProducer(){
        HashMap<String, Object> configs = new HashMap<>();
        configs.put("bootstrap.servers","60.204.221.52:9092");
        configs.put("retries",2);
        configs.put("batch.size",16384);
        //生产者可用于缓冲等待发送到服务器的记录的总内存字节数
        configs.put("buffer-memory",3554432);
        /**
         * ,用于控制发送记录在服务端的持久化
         *acks=0,设置为0,则生产者producer将不会等待来自服务器的任何确认.该记录将立即添加到套接字(socket)缓冲区并视为已发送
         * .在这种情况下,无法保证服务器已收到记录,并且重试配置(retries)将不会生效(因为客户端通常不会知道任何故障),每条记录返回的偏移量始终设置为-1.
         *acks=1,设置为1,leader节点会把记录写入本地日志,不需要等待所有follower节点完全确认就会立即应答producer.在这种情况下,
         * 在follower节点复制前,leader节点确认记录后立即失败的话,记录将会丢失.
         */
        configs.put("acks","-1");
        //指定key使用的序列化类
        StringSerializer keySerializer = new StringSerializer();
        //指定value使用的序列化类
        StringSerializer valueSerializer = new StringSerializer();
        //创建kafka生产者
        return new KafkaProducer(configs, keySerializer, valueSerializer);
    }

    @Bean
    public KafkaConsumer kafkaConsumer(){
        HashMap<String, Object> configs = new HashMap<>();
        configs.put("bootstrap.servers","60.204.221.52:9092");
        //开启consumer的偏移量（offset）自动提交到kafka
        configs.put("enable.auto.commit",true);
        //偏移量自动提交的时间间隔，单位毫秒
        configs.put("auto.commit.interval",5000);
        //在Kafka中没有初始化偏移量或者当前偏移量不存在情况
        //earliest, 在偏移量无效的情况下, 自动重置为最早的偏移量
        //latest, 在偏移量无效的情况下, 自动重置为最新的偏移量
        //none, 在偏移量无效的情况下, 抛出异常.
        configs.put("auto.offset.reset","latest");
        //请求阻塞的最大时间（毫秒）
        configs.put("fetch.max.wait",500);
        //请求应答的最小字节数
        configs.put("fetch.min.size",1);
        //心跳间隔时间（毫秒）
        configs.put("heartbeat-interval",3000);
        //一次调用poll返回的最大记录条数
        configs.put("max.poll.records",500);
        //指定消费组
        configs.put("group.id","firstGroup");
        //指定key使用的反序列化类
        StringDeserializer keyDeserializer = new StringDeserializer();
        //指定value使用的反序列化类
        StringDeserializer valueDeserializer = new StringDeserializer();
        //创建kafka消费者
        return new KafkaConsumer(configs,keyDeserializer,valueDeserializer);
    }
}
