kafka 0.10.2 消息生产者(producer)

package cn.xiaojf.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Utils; import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger; /**
* 消息生产者
* @author xiaojf 2017/3/22 14:27
*/
public class MsgProducer extends Thread { private final KafkaProducer<String, String> producer;
private final String topic;
private final boolean isAsync; public MsgProducer(String topic, boolean isAsync) {
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.59.130:9092");//broker 集群地址
properties.put(ProducerConfig.CLIENT_ID_CONFIG, "MsgProducer");//自定义客户端id
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");//key 序列号方式
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");//value 序列号方式
// properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,CustomPartitioner.class.getCanonicalName());//自定义分区函数 // properties.load("properties配置文件"); this.producer = new KafkaProducer<String, String>(properties);
this.topic = topic;
this.isAsync = isAsync;
} @Override
public void run() {
int msgNo = 0; while (true) {
String msg = "Msg: " + msgNo;
String key = msgNo + "";
if (isAsync) {//异步
producer.send(new ProducerRecord<String, String>(this.topic,msg));
// producer.send(new ProducerRecord<String, String>(this.topic, key, msg));
} else {//同步
producer.send(new ProducerRecord<String, String>(this.topic, key, msg),
new MsgProducerCallback(System.currentTimeMillis(), key, msg));
}
}
} /**
* 消息发送后的回调函数
*/
class MsgProducerCallback implements Callback { private final long startTime;
private final String key;
private final String msg; public MsgProducerCallback(long startTime, String key, String msg) {
this.startTime = startTime;
this.key = key;
this.msg = msg;
} public void onCompletion(RecordMetadata recordMetadata, Exception e) {
long elapsedTime = System.currentTimeMillis() - startTime;
if (recordMetadata != null) {
System.out.println(msg + " be sended to partition no : " + recordMetadata.partition());
}
}
} public static void main(String args[]) {
new MsgProducer("my-replicated-topic",true).start();//开始发送消息
}
}
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.10.2.0</version>
</dependency>
上一篇:OpenStack Newton版本Ceph集成部署记录


下一篇:python中一切皆是对象,对象都是在堆上存放的,一切都是指针