// Path: cango-common/src/main/java/com/bella/cango/exception/CangoException.java
// public class CangoException extends RuntimeException {
//     public CangoException() {
//         super();
//     }
// 
//     public CangoException(String message) {
//         super(message);
//     }
// 
//     public CangoException(String message, Throwable cause) {
//         super(message, cause);
//     }
// 
//     public CangoException(Throwable cause) {
//         super(cause);
//     }
// 
//     protected CangoException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
//         super(message, cause, enableSuppression, writableStackTrace);
//     }
// }
// 
// Path: cango-common/src/main/java/com/bella/cango/utils/PropertiesFileLoader.java
// public class PropertiesFileLoader {
//     public static final String PROJECT_DIR;
// 
//     public static final String CLASSPATH_DIR;
// 
//     public static final String DEFAULT_CONF_DIR;
// 
//     public static final String DEFAULT_KAFKA_PATH;
// 
//     public static final String DEFAULT_LOG_HOME;
// 
//     static {
//         File classFile = new File(Thread.currentThread().getContextClassLoader().getResource(".").getFile());
//         CLASSPATH_DIR = classFile.getAbsolutePath();
//         PROJECT_DIR = classFile.getParentFile().getParentFile().getAbsolutePath();
//         File confFile = new File(PROJECT_DIR, "src/conf");
//         DEFAULT_CONF_DIR = confFile.getAbsolutePath();
//         DEFAULT_KAFKA_PATH = DEFAULT_CONF_DIR + File.separator + "kafkaProduce.properties";
//         DEFAULT_LOG_HOME = DEFAULT_CONF_DIR + File.separator + "log";
//     }
// 
//     public static Properties loadKafkaProperties() throws IOException {
//         String kafkaPath = System.getProperty("kafkaProduce");
//         if (StringUtils.isEmpty(kafkaPath)) {
//             kafkaPath = DEFAULT_KAFKA_PATH;
//         }
// 
//         return PropertiesLoaderUtils.loadProperties(new FileSystemResource(kafkaPath));
//     }
// 
//     public static Properties load(String path) throws IOException {
//         if (StringUtils.isEmpty(path)) {
//             return new Properties();
//         }
// 
//         return PropertiesLoaderUtils.loadProperties(new FileSystemResource(path));
//     }
// }


import com.alibaba.fastjson.JSON;
import com.bella.cango.exception.CangoException;
import com.bella.cango.utils.PropertiesFileLoader;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
            }
        });
    }

    public void send(String message) {
        messageQueue.offer(message);
    }

    public void send(Object object) {
        messageQueue.offer(JSON.toJSONString(object));
    }

    public void start() {
        if (!start) {
            worker.start();
        }
    }

    private void init() {
        try {
            Properties props = PropertiesFileLoader.loadKafkaProperties();
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("serializer.class", "kafka.serializer.StringEncoder");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("request.required.acks", "0");
            producer = new Producer<>(new ProducerConfig(props));
            topic = props.getProperty("topic", "cango");
        } catch (FileNotFoundException e) {
            e.printStackTrace();
            LOGGER.error("not found kafkaProduce.properties, please check it");
// Path: cango-common/src/main/java/com/bella/cango/exception/CangoException.java
// public class CangoException extends RuntimeException {
//     public CangoException() {
//         super();
//     }
// 
//     public CangoException(String message) {
//         super(message);
//     }
// 
//     public CangoException(String message, Throwable cause) {
//         super(message, cause);
//     }
// 
//     public CangoException(Throwable cause) {
//         super(cause);
//     }
// 
//     protected CangoException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
//         super(message, cause, enableSuppression, writableStackTrace);
//     }
// }
// 
// Path: cango-common/src/main/java/com/bella/cango/utils/PropertiesFileLoader.java
// public class PropertiesFileLoader {
//     public static final String PROJECT_DIR;
// 
//     public static final String CLASSPATH_DIR;
// 
//     public static final String DEFAULT_CONF_DIR;
// 
//     public static final String DEFAULT_KAFKA_PATH;
// 
//     public static final String DEFAULT_LOG_HOME;
// 
//     static {
//         File classFile = new File(Thread.currentThread().getContextClassLoader().getResource(".").getFile());
//         CLASSPATH_DIR = classFile.getAbsolutePath();
//         PROJECT_DIR = classFile.getParentFile().getParentFile().getAbsolutePath();
//         File confFile = new File(PROJECT_DIR, "src/conf");
//         DEFAULT_CONF_DIR = confFile.getAbsolutePath();
//         DEFAULT_KAFKA_PATH = DEFAULT_CONF_DIR + File.separator + "kafkaProduce.properties";
//         DEFAULT_LOG_HOME = DEFAULT_CONF_DIR + File.separator + "log";
//     }
// 
//     public static Properties loadKafkaProperties() throws IOException {
//         String kafkaPath = System.getProperty("kafkaProduce");
//         if (StringUtils.isEmpty(kafkaPath)) {
//             kafkaPath = DEFAULT_KAFKA_PATH;
//         }
// 
//         return PropertiesLoaderUtils.loadProperties(new FileSystemResource(kafkaPath));
//     }
// 
//     public static Properties load(String path) throws IOException {
//         if (StringUtils.isEmpty(path)) {
//             return new Properties();
//         }
// 
//         return PropertiesLoaderUtils.loadProperties(new FileSystemResource(path));
//     }
// }


// Path: cango-message/src/main/java/com/bella/cango/message/producer/KafkaProducer.java
import com.alibaba.fastjson.JSON;
import com.bella.cango.exception.CangoException;
import com.bella.cango.utils.PropertiesFileLoader;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

            }
        });
    }

    public void send(String message) {
        messageQueue.offer(message);
    }

    public void send(Object object) {
        messageQueue.offer(JSON.toJSONString(object));
    }

    public void start() {
        if (!start) {
            worker.start();
        }
    }

    private void init() {
        try {
            Properties props = PropertiesFileLoader.loadKafkaProperties();
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("serializer.class", "kafka.serializer.StringEncoder");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("request.required.acks", "0");
            producer = new Producer<>(new ProducerConfig(props));
            topic = props.getProperty("topic", "cango");
        } catch (FileNotFoundException e) {
            e.printStackTrace();
            LOGGER.error("not found kafkaProduce.properties, please check it");
            throw new CangoException(e);