kafka|Kafka消费者JavaAPI Kafka生产者JavaAPI SparkStreaming|and

package cn.edu360.kafka;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
【kafka|Kafka消费者JavaAPI Kafka生产者JavaAPI】public class ConsumerDemo {
private static final String topic = “test888”;
private static final Integer threads = 2;

public static void main(String[] args) {Properties props = new Properties(); props.put("zookeeper.connect", "node-1.xiaoniu.com:2181,node-2.xiaoniu.com:2181,node-3.xiaoniu.com:2181"); props.put("group.id", "vvvvv"); //smallest重最开始消费,largest代表重消费者启动后产生的数据才消费 props.put("auto.offset.reset", "smallest"); ConsumerConfig config = new ConsumerConfig(props); ConsumerConnector consumer =Consumer.createJavaConsumerConnector(config); Map topicCountMap = new HashMap(); topicCountMap.put(topic, threads); Map>> consumerMap = consumer.createMessageStreams(topicCountMap); List> streams = consumerMap.get(topic); for(final KafkaStream kafkaStream : streams){ new Thread(new Runnable() {public void run() { for(MessageAndMetadata mm : kafkaStream){ String msg = new String(mm.message()); System.out.println(msg); } } }).start(); } }

}
————————————–生产者——————————————-
package cn.edu360.kafka;
import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
public class ProducerDemo {

public static void main(String[] args) { Properties props = new Properties(); props.put("metadata.broker.list", "node-1.xiaoniu.com:9092,node-2.xiaoniu.com:9092,node-3.xiaoniu.com:9092"); props.put("serializer.class", "kafka.serializer.StringEncoder"); ProducerConfig config = new ProducerConfig(props); Producer producer = new Producer(config); for (int i = 1001; i <= 1100; i++) producer.send(new KeyedMessage("test888", "xiaoniu" + i)); }

}