1 package storm.test.kafka;
 2
 3         import java.util.Properties;
 4
 5         import kafka.javaapi.producer.Producer;
 6         import kafka.producer.KeyedMessage;
 7         import kafka.producer.ProducerConfig;
 8         import kafka.serializer.StringEncoder;
 9
10         public class TestProducer {
11
12             public static void main(String[] args) throws Exception {
13                 Properties prop = new Properties();
14                 prop.put("zookeeper.connect", "h5:2181,h6:2181,h7:2181");
15                 prop.put("metadata.broker.list", "h5:9092,h6:9092,h7:9092");
16                 prop.put("serializer.class", StringEncoder.class.getName());
17                 Producer<String, String> producer = new Producer<String, String>(new ProducerConfig(prop));
18                 int i = 0;
19                 while(true){
20                     producer.send(new KeyedMessage<String, String>("test111", "msg:"+i++));
21                     Thread.sleep(1000);
22                 }
23             }
24
25         }

 1 package storm.test.kafka;
 2
 3         import java.util.HashMap;
 4         import java.util.List;
 5         import java.util.Map;
 6         import java.util.Properties;
 7
 8         import kafka.consumer.Consumer;
 9         import kafka.consumer.ConsumerConfig;
10         import kafka.consumer.ConsumerIterator;
11         import kafka.consumer.KafkaStream;
12         import kafka.javaapi.consumer.ConsumerConnector;
13         import kafka.serializer.StringEncoder;
14
15         public class TestConsumer {
16
17             static final String topic = "test111";
18
19             public static void main(String[] args) {
20                 Properties prop = new Properties();
21                 prop.put("zookeeper.connect", "h5:2181,h6:2181,h7:2181");
22                 prop.put("serializer.class", StringEncoder.class.getName());
23                 prop.put("metadata.broker.list", "h5:9092,h6:9092,h7:9092");
24                 prop.put("group.id", "group1");
25                 ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(prop));
26                 Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
27                 topicCountMap.put(topic, 1);
28                 Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = consumer.createMessageStreams(topicCountMap);
29                 final KafkaStream<byte[], byte[]> kafkaStream = messageStreams.get(topic).get(0);
30                 ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
31                 while (iterator.hasNext()) {
32                     String msg = new String(iterator.next().message());
33                     System.out.println("收到消息："+msg);
34                 }
35             }
36
37         }