java利用kafka生产消费消息

2019独角兽企业重金招聘Python工程师标准>>>

1.producer程序

package com.test.frame.kafka.controller;import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig;import java.util.Properties;public class KafkaProducer {private final Producer<String, String> producer;public final static String TOPIC = "my-multi-topic";//构造方法private KafkaProducer() {Properties props = new Properties();props.put("metadata.broker.list", "localhost:9092");props.put("serializer.class", "kafka.serializer.StringEncoder");props.put("key.serializer.class", "kafka.serializer.StringEncoder");props.put("request.required.acks", "-1");producer = new Producer<String, String>(new ProducerConfig(props));}void produce() {int messageNo = 90;final int COUNT = 100;while (messageNo < COUNT) {String key = String.valueOf(messageNo);String data = "hello kafka message" + key;producer.send(new KeyedMessage<String, String>(TOPIC, key ,data));System.out.println(data);messageNo++;}}public static void main(String[] args) throws Exception {new KafkaProducer().produce();}}

运行结果：

消费方接收到的消息如下：

2.consumer端程序：

package com.test.frame.kafka.controller;import kafka.consumer.ConsumerConfig; import kafka.consumer.ConsumerIterator; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector; import kafka.serializer.StringDecoder; import kafka.utils.VerifiableProperties;import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties;public class KafkaConsumer {private final ConsumerConnector consumer;private KafkaConsumer() {Properties props = new Properties();//zookeeper 配置props.put("zookeeper.connect", "localhost:2181");//group 代表一个消费组props.put("group.id", "jd-group");//zk连接超时props.put("zookeeper.session.timeout.ms", "4000");props.put("zookeeper.sync.time.ms", "200");props.put("auto.commit.interval.ms", "1000");props.put("auto.offset.reset", "smallest");//序列化类props.put("serializer.class", "kafka.serializer.StringEncoder");ConsumerConfig config = new ConsumerConfig(props);consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config); }void consume() {Map<String, Integer> topicCountMap = new HashMap<String, Integer>();topicCountMap.put(KafkaProducer.TOPIC, new Integer(1));StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());Map<String, List<KafkaStream<String, String>>> consumerMap =consumer.createMessageStreams(topicCountMap,keyDecoder,valueDecoder);KafkaStream<String, String> stream = consumerMap.get(KafkaProducer.TOPIC).get(0);ConsumerIterator<String, String> it = stream.iterator();while (it.hasNext())System.out.println(it.next().message());}public static void main(String[] args) {new KafkaConsumer().consume();}}

运行结果如下：

此时已经联通成功。

 

 

 

 

 

转载于:https://my.oschina.net/u/2263272/blog/1527979

总结

以上是生活随笔为你收集整理的java利用kafka生产消费消息的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。