首页 > 代码库 > kafka的生产者类
kafka的生产者类
kafka的生产者类
1 import com.*.message.Configuration; 2 import org.apache.kafka.clients.producer.Callback; 3 import org.apache.kafka.clients.producer.ProducerRecord; 4 import org.apache.kafka.clients.producer.RecordMetadata; 5 import org.apache.kafka.common.serialization.ByteArraySerializer; 6 import org.slf4j.Logger; 7 import org.slf4j.LoggerFactory; 8 9 import java.io.IOException; 10 import java.util.List; 11 import java.util.concurrent.Future; 12 import java.util.concurrent.atomic.AtomicInteger; 13 14 /** 15 * kafka消息生产者 16 * 17 * @author 18 * @version V1.0 19 * @param <T> 20 * @modify by user: {修改人} 2015-4-14 21 * @modify by reason:{方法名}:{原因} 22 */ 23 public abstract class KafkaProducer<T> implements IProducer<T>{ 24 25 private final Logger LOGGER = LoggerFactory.getLogger(this.getClass()); 26 public AtomicInteger sendNum = new AtomicInteger(); 27 28 private String topic; 29 private Integer partition = null; 30 private org.apache.kafka.clients.producer.KafkaProducer<byte[], byte[]> producer; 31 32 /** 33 * 创建一个新的实例KafkaProducer. 34 * 35 * @param configuration 36 * 配置 37 * @param topic 38 * 消息主题 39 */ 40 public KafkaProducer(Configuration configuration, String topic) { 41 this.topic = topic; 42 producer = new org.apache.kafka.clients.producer.KafkaProducer<byte[], byte[]>(configuration.getConfig(), 43 new ByteArraySerializer(), new ByteArraySerializer()); 44 } 45 46 /** 47 * 创建一个新的实例KafkaProducer. 48 * 49 * @param configuration 50 * 配置 51 * @param topic 52 * 消息主题 53 * @param partition 54 * 分区数 55 */ 56 public KafkaProducer(Configuration configuration, String topic, Integer partition) { 57 this(configuration, topic); 58 this.partition = partition; 59 } 60 61 /** 62 * 消息序列化方法 63 * 64 * @author 65 * @param msgInfo 66 * @return 67 */ 68 protected abstract byte[] serialize(T msgInfo); 69 70 /* 71 * (non-Javadoc) 72 * 73 * @see com.hikvision.bsp.message.producer.IProducer#send(T) 74 */ 75 @Override 76 public boolean send(final T msg) { 77 return send(msg, new Callback() { 78 @Override 79 public void onCompletion(RecordMetadata metadata, Exception exception) { 80 if (exception != null) { 81 LOGGER.error("send message ,error info:{}", exception.toString()); 82 } 83 } 84 }); 85 } 86 87 /* 88 * (non-Javadoc) 89 * 90 * @see com.hikvision.bsp.message.producer.IProducer#send(T, 91 * org.apache.kafka.clients.producer.Callback) 92 */ 93 @Override 94 public boolean send(final T msg, final Callback callback) { 95 if (msg == null) { 96 LOGGER.error("send msg is null. "); 97 return false; 98 } 99 100 // serialize list to protostuff, and send msg 101 byte[] data =http://www.mamicode.com/ serialize(msg); 102 try { 103 producer.send(new ProducerRecord<byte[], byte[]>(topic, partition, null, data), callback); 104 } catch (Throwable e) { 105 LOGGER.error("send msg faild.", e); 106 return false; 107 } 108 return true; 109 } 110 111 @Override 112 public boolean synSend(final T msg) { 113 if (msg == null) { 114 LOGGER.error("send msg is null. "); 115 return false; 116 } 117 118 // serialize list to protostuff, and send msg 119 byte[] data =http://www.mamicode.com/ serialize(msg); 120 Future<?> future = null; 121 try { 122 future = producer.send(new ProducerRecord<byte[], byte[]>(topic, partition, null, data)); 123 } catch (Throwable e) { 124 LOGGER.error("send msg faild.", e); 125 return false; 126 } finally { 127 try { 128 future.get(); 129 } catch (Exception e) { 130 LOGGER.error("send msg faild.", e); 131 return false; 132 } 133 } 134 return true; 135 } 136 137 /* 138 * (non-Javadoc) 139 * 140 * @see com.hikvision.bsp.message.producer.IProducer#send(java.util.List) 141 */ 142 @Override 143 public boolean send(List<T> msgList) { 144 // check arguments validation 145 if (msgList.size() == 0) { 146 LOGGER.error("send msg is null. "); 147 return false; 148 } 149 long curTime = System.currentTimeMillis(); 150 151 for (final T msg : msgList) { 152 send(msg); 153 } 154 LOGGER.info("Send {} messages to kafka take: {}ms", msgList.size(), (System.currentTimeMillis() - curTime)); 155 return true; 156 } 157 158 @Override 159 public void close() throws IOException { 160 if (producer != null) { 161 producer.close(); 162 } 163 } 164 }
kafka的生产者类
声明:以上内容来自用户投稿及互联网公开渠道收集整理发布,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任,若内容有误或涉及侵权可进行投诉: 投诉/举报 工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。