Kafka消费者JavaAPI Kafka生产者JavaAPI
2017-09-04 20:28
344 查看
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;
public class ConsumerDemo {
private static final String topic = “test888”;
private static final Integer threads = 2;
}
————————————–生产者——————————————-
package cn.edu360.kafka;
import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
public class ProducerDemo {
}
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;
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<String, Integer> topicCountMap = new HashMap<String, Integer>(); topicCountMap.put(topic, threads); Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap); List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic); for(final KafkaStream<byte[], byte[]> kafkaStream : streams){ new Thread(new Runnable() { public void run() { for(MessageAndMetadata<byte[], byte[]> 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<String, String> producer = new Producer<String, String>(config); for (int i = 1001; i <= 1100; i++) producer.send(new KeyedMessage<String, String>("test888", "xiaoniu" + i)); }
}
相关文章推荐
- Kafka java api-消费者代码与消费分析、生产者消费者配置文件详解
- kafka生产者和消费者的javaAPI的示例代码
- Kafka 生产者消费者 Java API 编程
- kafka集群搭建和使用Java写kafka生产者消费者
- kafka生产者、消费者java示例
- kafka集群搭建和使用Java写kafka生产者消费者
- 使用java创建kafka的生产者和消费者
- kafka生产者、消费者java示例
- kafka集群搭建和使用Java写kafka生产者消费者
- kafka生产者和消费者的javaAPI demo
- KAFKA生产者消费者API学习
- kafka生产者与消费者java代码示例
- java 实现kafka消息生产者和消费者
- kafka集群搭建和使用Java写kafka生产者消费者
- Kafka java api-生产者代码、高性能吞吐
- kafka集群搭建和使用Java写kafka生产者消费者
- afka集群搭建和使用Java写kafka生产者消费者
- 使用java创建kafka的生产者和消费者
- kafka集群搭建和使用Java写kafka生产者消费者
- kafka集群搭建和使用Java写kafka生产者消费者