kafka的生产者类

kafka的生产者类

import com.*.message.Configuration;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * kafka消息生产者
 * 
 * @author 
 * @version V1.0
 * @param <T>
 * @modify by user: {修改人} 2015-4-14
 * @modify by reason:{方法名}:{原因}
 */
public abstract class KafkaProducer<T> implements IProducer<T>{

    private final Logger LOGGER = LoggerFactory.getLogger(this.getClass());
    public AtomicInteger sendNum = new AtomicInteger();

    private String topic;
    private Integer partition = null;
    private org.apache.kafka.clients.producer.KafkaProducer<byte[], byte[]> producer;

    /**
     * 创建一个新的实例KafkaProducer.
     * 
     * @param configuration
     *            配置
     * @param topic
     *            消息主题
     */
    public KafkaProducer(Configuration configuration, String topic) {
        this.topic = topic;
        producer = new org.apache.kafka.clients.producer.KafkaProducer<byte[], byte[]>(configuration.getConfig(),
                new ByteArraySerializer(), new ByteArraySerializer());
    }

    /**
     * 创建一个新的实例KafkaProducer.
     * 
     * @param configuration
     *            配置
     * @param topic
     *            消息主题
     * @param partition
     *            分区数
     */
    public KafkaProducer(Configuration configuration, String topic, Integer partition) {
        this(configuration, topic);
        this.partition = partition;
    }

    /**
     * 消息序列化方法
     * 
     * @author 
     * @param msgInfo
     * @return
     */
    protected abstract byte[] serialize(T msgInfo);

    /*
     * (non-Javadoc)
     * 
     * @see com.hikvision.bsp.message.producer.IProducer#send(T)
     */
    @Override
    public boolean send(final T msg) {
        return send(msg, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    LOGGER.error("send message ,error info:{}", exception.toString());
                }
            }
        });
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.hikvision.bsp.message.producer.IProducer#send(T,
     * org.apache.kafka.clients.producer.Callback)
     */
    @Override
    public boolean send(final T msg, final Callback callback) {
        if (msg == null) {
            LOGGER.error("send msg is null. ");
            return false;
        }

        // serialize list to protostuff, and send msg
        byte[] data = serialize(msg);
        try {
            producer.send(new ProducerRecord<byte[], byte[]>(topic, partition, null, data), callback);
        } catch (Throwable e) {
            LOGGER.error("send msg faild.", e);
            return false;
        }
        return true;
    }

    @Override
    public boolean synSend(final T msg) {
        if (msg == null) {
            LOGGER.error("send msg is null. ");
            return false;
        }

        // serialize list to protostuff, and send msg
        byte[] data = serialize(msg);
        Future<?> future = null;
        try {
            future = producer.send(new ProducerRecord<byte[], byte[]>(topic, partition, null, data));
        } catch (Throwable e) {
            LOGGER.error("send msg faild.", e);
            return false;
        } finally {
            try {
                future.get();
            } catch (Exception e) {
                LOGGER.error("send msg faild.", e);
                return false;
            }
        }
        return true;
    }
    
    /*
     * (non-Javadoc)
     * 
     * @see com.hikvision.bsp.message.producer.IProducer#send(java.util.List)
     */
    @Override
    public boolean send(List<T> msgList) {
        // check arguments validation
        if (msgList.size() == 0) {
            LOGGER.error("send msg is null. ");
            return false;
        }
        long curTime = System.currentTimeMillis();

        for (final T msg : msgList) {
            send(msg);
        }
        LOGGER.info("Send {} messages to kafka take: {}ms", msgList.size(), (System.currentTimeMillis() - curTime));
        return true;
    }

    @Override
    public void close() throws IOException {
        if (producer != null) {
            producer.close();
        }
    }
}