本例以kafka2.10_0.10.0.0为例，不同版本的kafka Java api有些区别!

 

增加maven依赖

           
      
       org.apache.kafka
            
      
       kafka_2.10
            
      
       0.10.0.0
        
     

生产者

package com.zns.kafka;import java.util.Properties;import org.apache.kafka.clients.producer.KafkaProducer;import org.apache.kafka.clients.producer.ProducerConfig;import org.apache.kafka.clients.producer.ProducerRecord;import org.apache.kafka.common.serialization.StringSerializer;public class KafkaProducerTest {    public static String topicName = "test";    public static void main(String[] args) {        Properties props = new Properties();        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");// 9092是kafka默认端口        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());        KafkaProducer
     
       producer = new KafkaProducer
      
       (props);        for (int i = 1; i <= 10; i++) {            ProducerRecord
       
         message = new ProducerRecord
        
         (topicName, "hello world " + i);            producer.send(message);            producer.flush();        }    }}
        
       
      
     

 

 

消费者

package com.zns.kafka;import java.util.HashMap;import java.util.List;import java.util.Map;import java.util.Properties;import org.apache.kafka.clients.producer.ProducerConfig;import org.apache.kafka.common.serialization.StringSerializer;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;public class KafkaConsumerTest {    public static String topicName = "test";    public static void main(String[] args) {        Properties props = new Properties();        props.put("zookeeper.connect", "127.0.0.1:2181");// 2181是zookeeper默认端口        props.put("group.id", "test-group");        props.put("zookeeper.session.timeout.ms", "100000");        props.put("zookeeper.sync.time.ms", "200");        props.put("auto.commit.interval.ms", "1000");        props.put("auto.offset.reset", "smallest");        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());        ConsumerConfig config = new ConsumerConfig(props);        ConsumerConnector consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);        Map
     
       topicCountMap = new HashMap
      
       ();        topicCountMap.put(topicName, new Integer(1));        StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());        StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());        Map
       
        >> consumerMap = consumer.createMessageStreams(topicCountMap,                keyDecoder, valueDecoder);        KafkaStream
        
          stream = consumerMap.get(topicName).get(0);        ConsumerIterator
         
           it = stream.iterator();        while (it.hasNext()) {            System.out.println("收到消息：" + it.next().message());        }    }}
         
        
       
      
     

 

 

确保启动运行了zookeeper和kafka

先后启动运行生产者和消费者，可以看到消费者端接收到了消息...