KafkaKafka

kafka

package com.lin.demo.consumer;  
  
import java.util.HashMap;  
import java.util.List;  
import java.util.Map;  
import java.util.Properties;  
  
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;  
  
import com.lin.demo.producer.KafkaProducer;  
  
public class KafkaConsumer {  
  
    private final ConsumerConnector consumer;  
  
    private KafkaConsumer（） {  
        Properties props = new Properties（）;  
        // zookeeper 配置  
        props.put（“zookeeper.connect“ “127.0.0.1:2181“）;  
  
        // group 代表一个消费组  
        props.put（“group.id“ “lingroup“）;  
  
        // zk连接超时  
        props.put（“zookeeper.session.timeout.ms“ “4000“）;  
        props.put（“zookeeper.sync.time.ms“ “200“）;  
        props.put（“rebalance.max.retries“ “5“）;  
        props.put（“rebalance.backoff.ms“ “1200“）;  
          
      
        props.put（“auto.commit.interval.ms“ “1000“）;  
        props.put（“auto.offset.reset“ “smallest“）;  
        // 序列化类  
        props.put（“serializer.class“ “kafka.serializer.StringEncoder“）;  
  
        ConsumerConfig config = new ConsumerConfig（props）;  
  
        consumer = kafka.consumer.Consumer.createJavaConsumerConnector（config）;  
    }  
  
    void consume（） {  
        Map topicCountMap = new HashMap（）;  
        topicCountMap.put（KafkaProducer.TOPIC 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（KafkaProducer.TOPIC）.get（0）;  
        ConsumerIterator it = stream.iterator（）;  
        while （it.hasNext（）） {
        	System.out.println（“<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<“ + it.next（）.message（） + “<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<“）;  
        /*	KafkaProducer kafkaProducer = new KafkaProducer（）;
        	kafkaProducer.produce（）;*/
        }
    }  
  
    public static void main（String[] args） {  
        new KafkaConsumer（）.consume（）;  
    }  
}  

 属性            大小     日期    时间   名称
----------- ---------  ---------- -----  ----
     目录           0  2017-06-19 14:15  Kafka-Demo\
     文件         596  2017-06-19 14:15  Kafka-Demo\.classpath
     目录           0  2017-06-19 14:15  Kafka-Demo\.myeclipse\
     文件         304  2017-06-19 14:15  Kafka-Demo\.mymetadata
     文件        1537  2017-06-19 14:15  Kafka-Demo\.project
     目录           0  2017-06-19 14:15  Kafka-Demo\.settings\
     文件         500  2017-06-19 14:15  Kafka-Demo\.settings\.jsdtscope
     文件         364  2017-06-19 14:15  Kafka-Demo\.settings\org.eclipse.jdt.core.prefs
     文件         568  2017-06-19 14:15  Kafka-Demo\.settings\org.eclipse.wst.common.component
     文件         252  2017-06-19 14:15  Kafka-Demo\.settings\org.eclipse.wst.common.project.facet.core.xml
     文件          49  2017-06-19 14:15  Kafka-Demo\.settings\org.eclipse.wst.jsdt.ui.superType.container
     文件           6  2017-06-19 14:15  Kafka-Demo\.settings\org.eclipse.wst.jsdt.ui.superType.name
     目录           0  2017-06-19 14:15  Kafka-Demo\src\
     目录           0  2017-06-19 14:15  Kafka-Demo\src\main\
     目录           0  2017-06-19 14:15  Kafka-Demo\src\main\java\
     目录           0  2017-06-19 14:15  Kafka-Demo\src\main\java\com\
     目录           0  2017-06-19 14:15  Kafka-Demo\src\main\java\com\lin\
     目录           0  2017-06-19 14:15  Kafka-Demo\src\main\java\com\lin\demo\
     目录           0  2017-06-19 14:15  Kafka-Demo\src\main\java\com\lin\demo\consumer\
     文件        2474  2017-06-19 14:15  Kafka-Demo\src\main\java\com\lin\demo\consumer\KafkaConsumer.java
     目录           0  2017-06-19 14:15  Kafka-Demo\src\main\java\com\lin\demo\producer\
     文件        1521  2017-06-19 14:15  Kafka-Demo\src\main\java\com\lin\demo\producer\KafkaProducer.java
     目录           0  2017-06-19 14:15  Kafka-Demo\src\main\resources\
     文件         292  2017-06-19 14:15  Kafka-Demo\src\main\resources\log4j.properties
     目录           0  2017-06-19 14:15  Kafka-Demo\WebRoot\
     文件         834  2017-06-19 14:15  Kafka-Demo\WebRoot\index.jsp
     目录           0  2017-06-19 14:15  Kafka-Demo\WebRoot\meta-INF\
     文件          36  2017-06-19 14:15  Kafka-Demo\WebRoot\meta-INF\MANIFEST.MF
     目录           0  2017-06-19 14:15  Kafka-Demo\WebRoot\WEB-INF\
     目录           0  2017-06-19 14:15  Kafka-Demo\WebRoot\WEB-INF\classes\
     目录           0  2017-06-19 14:15  Kafka-Demo\WebRoot\WEB-INF\classes\com\
............此处省略70个文件信息