Kafka的生产者和消费者代码解析-白红宇

Kafka的生产者和消费者代码解析

阅读量：6862 次

发布时间：2019-06-26

本文共 22037 字，大约阅读时间需要 73 分钟。

1:Kafka名词解释和工作方式    1.1:Producer ：消息生产者，就是向kafka broker发消息的客户端。    1.2:Consumer ：消息消费者，向kafka broker取消息的客户端    1.3:Topic ：可以理解为一个队列。    1.4:Consumer Group （CG）：这是kafka用来实现一个topic消息的广播（发给所有的consumer）和单播（发给任意一个consumer）的手段。一个topic可以有多个CG。topic的消息会复制（不是真的复制，是概念上的）到所有的CG，但每个partion只会把消息发给该CG中的一个consumer。如果需要实现广播，只要每个consumer有一个独立的CG就可以了。要实现单播只要所有的consumer在同一个CG。用CG还可以将consumer进行自由的分组而不需要多次发送消息到不同的topic。    1.5:Broker ：一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。    1.6:Partition：为了实现扩展性，一个非常大的topic可以分布到多个broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体（多个partition间）的顺序。    1.7:Offset：kafka的存储文件都是按照offset.kafka来命名，用offset做名字的好处是方便查找。例如你想找位于2049的位置，只要找到2048.kafka的文件即可。当然the first offset就是00000000000.kafka。 2:Consumer与topic关系?本质上kafka只支持Topic。   2.1:每个group中可以有多个consumer，每个consumer属于一个consumer group； 通常情况下，一个group中会包含多个consumer，这样不仅可以提高topic中消息的并发消费能力，而且还能提高"故障容错"性，如果group中的某个consumer失效那么其消费的partitions将会有其他consumer自动接管。   2.2:对于Topic中的一条特定的消息，只会被订阅此Topic的每个group中的其中一个consumer消费，此消息不会发送给一个group的多个consumer； 那么一个group中所有的consumer将会交错的消费整个Topic，每个group中consumer消息消费互相独立，我们可以认为一个group是一个"订阅"者。   2.3:在kafka中,一个partition中的消息只会被group中的一个consumer消费(同一时刻)； 一个Topic中的每个partions，只会被一个"订阅者"中的一个consumer消费，不过一个consumer可以同时消费多个partitions中的消息。   2.4:kafka的设计原理决定,对于一个topic，同一个group中不能有多于partitions个数的consumer同时消费，否则将意味着某些consumer将无法得到消息。   2.5：kafka只能保证一个partition中的消息被某个consumer消费时是顺序的；事实上，从Topic角度来说,当有多个partitions时,消息仍不是全局有序的。 3：Kafka消息的分发，Producer客户端负责消息的分发。 　 3.1:kafka集群中的任何一个broker都可以向producer提供metadata信息,这些metadata中包含"集群中存活的servers列表"/"partitions leader列表"等信息；   3.2:当producer获取到metadata信息之后, producer将会和Topic下所有partition leader保持socket连接；   3.3:消息由producer直接通过socket发送到broker，中间不会经过任何"路由层"，事实上，消息被路由到哪个partition上由producer客户端决定；       比如可以采用"random""key-hash""轮询"等,如果一个topic中有多个partitions,那么在producer端实现"消息均衡分发"是必要的。   3.4:在producer端的配置文件中,开发者可以指定partition路由的方式。   3.5:Producer消息发送的应答机制：     设置发送数据是否需要服务端的反馈,三个值0,1,-1。     0: producer不会等待broker发送ack。     1: 当leader接收到消息之后发送ack。     -1: 当所有的follower都同步消息成功后发送ack。         request.required.acks=0。 4：Consumer的负载均衡： 　　当一个group中,有consumer加入或者离开时,会触发partitions均衡.均衡的最终目的,是提升topic的并发消费能力：

步骤如下：

a、假如topic1,具有如下partitions: P0,P1,P2,P3。

b、加入group中,有如下consumer: C1,C2。

c、首先根据partition索引号对partitions排序: P0,P1,P2,P3。

d、根据consumer.id排序: C0,C1。

e、计算倍数: M = [P0,P1,P2,P3].size / [C0,C1].size,本例值M=2(向上取整)。

f、然后依次分配partitions: C0 = [P0,P1],C1=[P2,P3],即Ci = [P(i * M),P((i + 1) * M -1)]。

6：Kafka文件存储基本结构：

　　6.1：在Kafka文件存储中，同一个topic下有多个不同partition，每个partition为一个目录，partiton命名规则为topic名称+有序序号，第一个partiton序号从0开始，序号最大值为partitions数量减1。

　　6.2：每个partion(目录)相当于一个巨型文件被平均分配到多个大小相等segment(段)数据文件中。但每个段segment file消息数量不一定相等，这种特性方便old segment file快速被删除。默认保留7天的数据。

　　6.3：每个partiton只需要支持顺序读写就行了，segment文件生命周期由服务端配置参数决定。（什么时候创建，什么时候删除）。

1：使用Idea进行开发，源码如下所示，首先加入Kafka必须依赖的包，这句话意味着你必须要先在Idea上面搭建好的你的maven环境：

pom.xml如下所示内容：

1 
      2 
     
       5     
      
       4.0.0
       6  7     
      
       com.bie
       8     
      
       storm
       9     
      
       1.0-SNAPSHOT
      10 11     
      12     
      
       13         
       14         
       
        15             
        
         org.apache.storm
        16             
        
         storm-core
        17             
        
         0.9.5
        18             
        19         
       20         
       21         
       
        22             
        
         org.apache.kafka
        23             
        
         kafka_2.8.2
        24             
        
         0.8.1
        25             
        
         26                 
         
          27                     
          
           jmxtools
          28                     
          
           com.sun.jdmk
          29                 
         30                 
         
          31                     
          
           jmxri
          32                     
          
           com.sun.jmx
          33                 
         34                 
         
          35                     
          
           jms
          36                     
          
           javax.jms
          37                 
         38                 
         
          39                     
          
           org.apache.zookeeper
          40                     
          
           zookeeper
          41                 
         42                 
         
          43                     
          
           org.slf4j
          44                     
          
           slf4j-log4j12
          45                 
         46                 
         
          47                     
          
           org.slf4j
          48                     
          
           slf4j-api
          49                 
         50             
        51         
       52     
      53 54     
      55     
      
       56         
       
        57             
        
         58                 
         59                 
         
          maven-assembly-plugin
         60                 
         
          61                     
          
           62 
           
            jar-with-dependencies
           63 
          64                     
          
           65 
           
            66 
            
             com.bie.wordcount.WordCountTopologyMain
            67 
           68 
          69                 
         70                 
         
          71                     
          
           72 
           
            make-assembly
           73 
           
            package
           74 
           
            75 
            
             single
            76 
           77 
          78                 
         79             
        80             
        
         81                 
         
          org.apache.maven.plugins
         82                 
         
          maven-compiler-plugin
         83                 
         
          84                     
          1.785                     
          
           1.7
          86                 
         87             
        88         
       89     
      90 91 92

然后呢，书写你的生产者源码，如下所示：

package com.bie.kafka;import kafka.javaapi.producer.Producer;import kafka.producer.KeyedMessage;import kafka.producer.ProducerConfig;import java.util.Properties;import java.util.UUID;/** * 这是一个简单的Kafka producer代码 * 包含两个功能: * 1、数据发送 * 2、数据按照自定义的partition策略进行发送 * * * KafkaSpout的类 */public class KafkaProducerSimple {    public static void main(String[] args) {        /**         * 1、指定当前kafka producer生产的数据的目的地         *  创建topic可以输入以下命令，在kafka集群的任一节点进行创建。         *  bin/kafka-topics.sh --create --zookeeper master:2181         *  --replication-factor 1 --partitions 1 --topic orderMq         */        String TOPIC = "orderMq8";        /**         * 2、读取配置文件         */        Properties props = new Properties();        /*         * key.serializer.class默认为serializer.class         */        props.put("serializer.class", "kafka.serializer.StringEncoder");        /*         * kafka broker对应的主机，格式为host1:port1,host2:port2         */        props.put("metadata.broker.list", "master:9092,slaver1:9092,slaver2:9092");        /*         * request.required.acks,设置发送数据是否需要服务端的反馈,有三个值0,1,-1         * 0，意味着producer永远不会等待一个来自broker的ack，这就是0.7版本的行为。         * 这个选项提供了最低的延迟，但是持久化的保证是最弱的，当server挂掉的时候会丢失一些数据。         * 1，意味着在leader replica已经接收到数据后，producer会得到一个ack。         * 这个选项提供了更好的持久性，因为在server确认请求成功处理后，client才会返回。         * 如果刚写到leader上，还没来得及复制leader就挂了，那么消息才可能会丢失。         * -1，意味着在所有的ISR都接收到数据后，producer才得到一个ack。         * 这个选项提供了最好的持久性，只要还有一个replica存活，那么数据就不会丢失         */        props.put("request.required.acks", "1");        /*         * 可选配置，如果不配置，则使用默认的partitioner partitioner.class         * 默认值：kafka.producer.DefaultPartitioner         * 用来把消息分到各个partition中，默认行为是对key进行hash。         */        props.put("partitioner.class", "com.bie.kafka.MyLogPartitioner");        //props.put("partitioner.class", "kafka.producer.DefaultPartitioner");        /**         * 3、通过配置文件，创建生产者         */        Producer
     
       producer = new Producer
      
       (new ProducerConfig(props));        /**         * 4、通过for循环生产数据         */        for (int messageNo = 1; messageNo < 100000; messageNo++) {            String messageStr = new String(messageNo + "注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，" +                    "注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发" +                    "注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发" +                    "注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发" +                    "注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发" +                    "注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发" +                    "注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发" +                    "注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发" +                    "用来配合自定义的MyLogPartitioner进行数据分发");            /**             * 5、调用producer的send方法发送数据             * 注意：这里需要指定 partitionKey，用来配合自定义的MyLogPartitioner进行数据分发             */            producer.send(new KeyedMessage
       
        (TOPIC, messageNo + "", "appid" + UUID.randomUUID() + messageStr));            //producer.send(new KeyedMessage
        
         (TOPIC, messageNo + "", "appid" + UUID.randomUUID() + "biexiansheng"));        }    }}

生产者需要的Partitioner如下所示内容：

package com.bie.kafka;import kafka.producer.Partitioner;import kafka.utils.VerifiableProperties;import org.apache.log4j.Logger;public class MyLogPartitioner implements Partitioner {    private static Logger logger = Logger.getLogger(MyLogPartitioner.class);    public MyLogPartitioner(VerifiableProperties props) {    }    public int partition(Object obj, int numPartitions) {        return Integer.parseInt(obj.toString())%numPartitions;//        return 1;    }}

生产者运行效果如下所示：

消费者代码如下所示：

package com.bie.kafka;import kafka.consumer.Consumer;import kafka.consumer.ConsumerConfig;import kafka.consumer.ConsumerIterator;import kafka.consumer.KafkaStream;import kafka.javaapi.consumer.ConsumerConnector;import kafka.message.MessageAndMetadata;import java.util.HashMap;import java.util.List;import java.util.Map;import java.util.Properties;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;public class KafkaConsumerSimple implements Runnable {    public String title;    public KafkaStream
     
       stream;    public KafkaConsumerSimple(String title, KafkaStream
      
        stream) {        this.title = title;        this.stream = stream;    }    @Override    public void run() {        System.out.println("开始运行 " + title);        ConsumerIterator
       
         it = stream.iterator();        /**         * 不停地从stream读取新到来的消息，在等待新的消息时，hasNext()会阻塞         * 如果调用 `ConsumerConnector#shutdown`，那么`hasNext`会返回false         * */        while (it.hasNext()) {            MessageAndMetadata
        
          data = it.next();            Object topic = data.topic();            int partition = data.partition();            long offset = data.offset();            String msg = new String(data.message());            System.out.println(String.format(                    "Consumer: [%s],  Topic: [%s],  PartitionId: [%d], Offset: [%d], msg: [%s]",                    title, topic, partition, offset, msg));        }        System.out.println(String.format("Consumer: [%s] exiting ...", title));    }    public static void main(String[] args) throws Exception{        Properties props = new Properties();        props.put("group.id", "biexiansheng");        props.put("zookeeper.connect", "master:2181,slaver1:2181,slaver2:2181");        props.put("auto.offset.reset", "largest");        props.put("auto.commit.interval.ms", "1000");        props.put("partition.assignment.strategy", "roundrobin");        ConsumerConfig config = new ConsumerConfig(props);        String topic1 = "orderMq8";        //String topic2 = "paymentMq";        //只要ConsumerConnector还在的话，consumer会一直等待新消息，不会自己退出        ConsumerConnector consumerConn = Consumer.createJavaConsumerConnector(config);        //定义一个map        Map
         
           topicCountMap = new HashMap<>();        topicCountMap.put(topic1, 3);        //Map
          
           > 中String是topic， List
           
            
             是对应的流 Map
             
              >> topicStreamsMap = consumerConn.createMessageStreams(topicCountMap); //取出 `kafkaTest` 对应的 streams List
              
               
                > streams = topicStreamsMap.get(topic1); //创建一个容量为4的线程池 ExecutorService executor = Executors.newFixedThreadPool(3); //创建20个consumer threads for (int i = 0; i < streams.size(); i++) { executor.execute(new KafkaConsumerSimple("消费者" + (i + 1), streams.get(i))); } }}

消费者运行如下所示：

运行消费者出现下面的错误，解决方法将pomx.ml里面的zookeeper配置注释了即可：

错误如下所示：

1 D:\soft\Java\jdk1.7.0_80\bin\java -javaagent:E:\360Downloads\idea\lib\idea_rt.jar=61635:E:\360Downloads\idea\bin -Dfile.encoding=UTF-8 -classpath D:\soft\Java\jdk1.7.0_80\jre\lib\charsets.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\deploy.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\ext\access-bridge-64.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\ext\dnsns.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\ext\jaccess.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\ext\localedata.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\ext\sunec.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\ext\sunjce_provider.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\ext\sunmscapi.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\ext\zipfs.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\javaws.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\jce.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\jfr.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\jfxrt.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\jsse.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\management-agent.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\plugin.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\resources.jar;D:\soft\Java\jdk1.7.0_80\jre\lib\rt.jar;E:\360Downloads\idea\storm\target\classes;E:\maven\repository\org\apache\storm\storm-core\0.9.5\storm-core-0.9.5.jar;E:\maven\repository\org\clojure\clojure\1.5.1\clojure-1.5.1.jar;E:\maven\repository\clj-time\clj-time\0.4.1\clj-time-0.4.1.jar;E:\maven\repository\joda-time\joda-time\2.0\joda-time-2.0.jar;E:\maven\repository\compojure\compojure\1.1.3\compojure-1.1.3.jar;E:\maven\repository\org\clojure\core.incubator\0.1.0\core.incubator-0.1.0.jar;E:\maven\repository\org\clojure\tools.macro\0.1.0\tools.macro-0.1.0.jar;E:\maven\repository\clout\clout\1.0.1\clout-1.0.1.jar;E:\maven\repository\ring\ring-core\1.1.5\ring-core-1.1.5.jar;E:\maven\repository\commons-fileupload\commons-fileupload\1.2.1\commons-fileupload-1.2.1.jar;E:\maven\repository\javax\servlet\servlet-api\2.5\servlet-api-2.5.jar;E:\maven\repository\hiccup\hiccup\0.3.6\hiccup-0.3.6.jar;E:\maven\repository\ring\ring-devel\0.3.11\ring-devel-0.3.11.jar;E:\maven\repository\clj-stacktrace\clj-stacktrace\0.2.2\clj-stacktrace-0.2.2.jar;E:\maven\repository\ring\ring-jetty-adapter\0.3.11\ring-jetty-adapter-0.3.11.jar;E:\maven\repository\ring\ring-servlet\0.3.11\ring-servlet-0.3.11.jar;E:\maven\repository\org\mortbay\jetty\jetty\6.1.26\jetty-6.1.26.jar;E:\maven\repository\org\mortbay\jetty\jetty-util\6.1.26\jetty-util-6.1.26.jar;E:\maven\repository\org\clojure\tools.logging\0.2.3\tools.logging-0.2.3.jar;E:\maven\repository\org\clojure\math.numeric-tower\0.0.1\math.numeric-tower-0.0.1.jar;E:\maven\repository\org\clojure\tools.cli\0.2.4\tools.cli-0.2.4.jar;E:\maven\repository\commons-io\commons-io\2.4\commons-io-2.4.jar;E:\maven\repository\org\apache\commons\commons-exec\1.1\commons-exec-1.1.jar;E:\maven\repository\commons-lang\commons-lang\2.5\commons-lang-2.5.jar;E:\maven\repository\com\googlecode\json-simple\json-simple\1.1\json-simple-1.1.jar;E:\maven\repository\com\twitter\carbonite\1.4.0\carbonite-1.4.0.jar;E:\maven\repository\com\esotericsoftware\kryo\kryo\2.21\kryo-2.21.jar;E:\maven\repository\com\esotericsoftware\reflectasm\reflectasm\1.07\reflectasm-1.07-shaded.jar;E:\maven\repository\org\ow2\asm\asm\4.0\asm-4.0.jar;E:\maven\repository\com\esotericsoftware\minlog\minlog\1.2\minlog-1.2.jar;E:\maven\repository\org\objenesis\objenesis\1.2\objenesis-1.2.jar;E:\maven\repository\com\twitter\chill-java\0.3.5\chill-java-0.3.5.jar;E:\maven\repository\org\yaml\snakeyaml\1.11\snakeyaml-1.11.jar;E:\maven\repository\commons-logging\commons-logging\1.1.3\commons-logging-1.1.3.jar;E:\maven\repository\commons-codec\commons-codec\1.6\commons-codec-1.6.jar;E:\maven\repository\com\googlecode\disruptor\disruptor\2.10.1\disruptor-2.10.1.jar;E:\maven\repository\org\jgrapht\jgrapht-core\0.9.0\jgrapht-core-0.9.0.jar;E:\maven\repository\ch\qos\logback\logback-classic\1.0.13\logback-classic-1.0.13.jar;E:\maven\repository\ch\qos\logback\logback-core\1.0.13\logback-core-1.0.13.jar;E:\maven\repository\org\slf4j\slf4j-api\1.7.5\slf4j-api-1.7.5.jar;E:\maven\repository\org\slf4j\log4j-over-slf4j\1.6.6\log4j-over-slf4j-1.6.6.jar;E:\maven\repository\jline\jline\2.11\jline-2.11.jar;E:\maven\repository\org\apache\kafka\kafka_2.8.2\0.8.1\kafka_2.8.2-0.8.1.jar;E:\maven\repository\org\scala-lang\scala-library\2.8.2\scala-library-2.8.2.jar;E:\maven\repository\com\yammer\metrics\metrics-annotation\2.2.0\metrics-annotation-2.2.0.jar;E:\maven\repository\com\yammer\metrics\metrics-core\2.2.0\metrics-core-2.2.0.jar;E:\maven\repository\org\xerial\snappy\snappy-java\1.0.5\snappy-java-1.0.5.jar;E:\maven\repository\net\sf\jopt-simple\jopt-simple\3.2\jopt-simple-3.2.jar;E:\maven\repository\com\101tec\zkclient\0.3\zkclient-0.3.jar;E:\maven\repository\log4j\log4j\1.2.14\log4j-1.2.14.jar com.bie.kafka.KafkaConsumerSimple 2 260  [main] INFO  kafka.utils.VerifiableProperties - Verifying properties 3 311  [main] INFO  kafka.utils.VerifiableProperties - Property auto.commit.interval.ms is overridden to 1000 4 311  [main] INFO  kafka.utils.VerifiableProperties - Property auto.offset.reset is overridden to largest 5 311  [main] INFO  kafka.utils.VerifiableProperties - Property group.id is overridden to biexiansheng 6 312  [main] WARN  kafka.utils.VerifiableProperties - Property partition.assignment.strategy is not valid 7 312  [main] INFO  kafka.utils.VerifiableProperties - Property zookeeper.connect is overridden to master:2181,slaver1:2181,slaver2:2181 8 448  [main] INFO  kafka.consumer.ZookeeperConsumerConnector - [biexiansheng_HY-201707051724-1516692275031-bffb9bfb], Connecting to zookeeper instance at master:2181,slaver1:2181,slaver2:2181 9 Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/zookeeper/Watcher10     at java.lang.ClassLoader.defineClass1(Native Method)11     at java.lang.ClassLoader.defineClass(ClassLoader.java:800)12     at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)13     at java.net.URLClassLoader.defineClass(URLClassLoader.java:449)14     at java.net.URLClassLoader.access$100(URLClassLoader.java:71)15     at java.net.URLClassLoader$1.run(URLClassLoader.java:361)16     at java.net.URLClassLoader$1.run(URLClassLoader.java:355)17     at java.security.AccessController.doPrivileged(Native Method)18     at java.net.URLClassLoader.findClass(URLClassLoader.java:354)19     at java.lang.ClassLoader.loadClass(ClassLoader.java:425)20     at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)21     at java.lang.ClassLoader.loadClass(ClassLoader.java:358)22     at kafka.consumer.ZookeeperConsumerConnector.connectZk(ZookeeperConsumerConnector.scala:156)23     at kafka.consumer.ZookeeperConsumerConnector.
     
      (ZookeeperConsumerConnector.scala:114)24     at kafka.javaapi.consumer.ZookeeperConsumerConnector.
      
       (ZookeeperConsumerConnector.scala:65)25     at kafka.javaapi.consumer.ZookeeperConsumerConnector.
       
        (ZookeeperConsumerConnector.scala:67)26     at kafka.consumer.Consumer$.createJavaConsumerConnector(ConsumerConnector.scala:100)27     at kafka.consumer.Consumer.createJavaConsumerConnector(ConsumerConnector.scala)28     at com.bie.kafka.KafkaConsumerSimple.main(KafkaConsumerSimple.java:58)29 Caused by: java.lang.ClassNotFoundException: org.apache.zookeeper.Watcher30     at java.net.URLClassLoader$1.run(URLClassLoader.java:366)31     at java.net.URLClassLoader$1.run(URLClassLoader.java:355)32     at java.security.AccessController.doPrivileged(Native Method)33     at java.net.URLClassLoader.findClass(URLClassLoader.java:354)34     at java.lang.ClassLoader.loadClass(ClassLoader.java:425)35     at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)36     at java.lang.ClassLoader.loadClass(ClassLoader.java:358)37     ... 19 more38 39 Process finished with exit code 1

运行效果如下所示：

待续......

转载地址：http://txqyl.baihongyu.com/

你可能感兴趣的文章

缺乏规模效应腾讯视频或借道短视频突围竞争