kafka 简单 java 生产消费API

kafka 简单 java 生产消费API

1. KafkaProperties

package com.immooc.spark.kafka;public class KafkaProperties { public static final String ZK = "localhost:2181"; public static final String TOPIC = "test"; public static final String BROKER_LIST = "localhost:9092"; public static final String GROUP_ID = "test_group1";}

2. KafkaProducer

package com.immooc.spark.kafka;import kafka.javaapi.producer.Producer;import kafka.producer.KeyedMessage;import kafka.producer.ProducerConfig;import java.util.Properties;public class KafkaProducer extends Thread{ private String topic; private Producer producer; public KafkaProducer(String topic){ this.topic = topic; Properties properties = new Properties(); properties.put("metadata.broker.list", KafkaProperties.BROKER_LIST); properties.put("serializer.class", "kafka.serializer.StringEncoder"); properties.put("request.required.acks", "1"); producer = new Producer(new ProducerConfig(properties)); } @Override public void run(){ int messageNo = 1; while (true){ String message = "message_" + messageNo; producer.send(new KeyedMessage(topic, message)); System.out.println("Send: " + message); messageNo ++; try{ Thread.sleep(2000); }catch (Exception e){ e.printStackTrace(); } } }}

3. KafkaConsumer

package com.immooc.spark.kafka;import kafka.consumer.Consumer;import kafka.consumer.ConsumerConfig;import kafka.consumer.ConsumerIterator;import kafka.consumer.KafkaStream;import kafka.javaapi.consumer.ConsumerConnector;import java.util.HashMap;import java.util.List;import java.util.Map;import java.util.Properties;public class KafkaConsumer extends Thread{ private String topic; public KafkaConsumer(String topic){ this.topic = topic; } private ConsumerConnector createConnector(){ Properties properties = new Properties(); properties.put("zookeeper.connect", KafkaProperties.ZK); properties.put("group.id", KafkaProperties.GROUP_ID); return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties)); } @Override public void run(){ ConsumerConnector consumer = createConnector(); Map topicCountMap = new HashMap(); topicCountMap.put(topic, 1); Map>> messageStream = consumer.createMessageStreams(topicCountMap); KafkaStream stream = messageStream.get(topic).get(0); ConsumerIterator iterator = stream.iterator(); while (iterator.hasNext()){ String message = new String(iterator.next().message()); System.out.println("rec:" + message); } }}

kafka架构参考这篇文章

​​1.1的看这篇文章​​《kafka 简单 java 生产消费API 1.1》​​

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。