SEEC-LogTracking/seecoder-logTracking-common/src/main/java/com/nju/edu/logTracking/kafka/KafkaUtils.java

package com.nju.edu.logTracking.kafka;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;

import java.util.List;
import java.util.Properties;


public class KafkaUtils {
    private volatile static KafkaProducer<String, String> kafkaProducer;

    public static KafkaConsumer<String, String> createConsumer(List<String> topic) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", Config.KAFKA_SERVERS);
        properties.put("group.id", "group-1");
        properties.put("enable.auto.commit", "false");
        properties.put("auto.commit.interval.ms", "1000");
        properties.put("auto.offset.reset", "earliest");
        properties.put("session.timeout.ms", "30000");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        kafkaConsumer.subscribe(topic);
        return kafkaConsumer;
    }

    public static KafkaProducer<String, String> getProducer() {
        if(kafkaProducer == null){
            synchronized (KafkaUtils.class){
                if(kafkaProducer==null){
                    Properties properties = new Properties();
                    properties.put("bootstrap.servers", Config.KAFKA_SERVERS);
                    properties.put("acks", "all");
                    properties.put("linger.ms", 5);
                    properties.put("retries", 3);
                    properties.put("request.timeout.ms", 5000);
                    properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
                    properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
                    kafkaProducer = new KafkaProducer<>(properties);
                }
            }
        }
        return kafkaProducer;
    }
}