【原创】Kafka console consumer源代码分析(一)

上一篇中分析了Scala版的console producer代码，这篇文章为读者带来一篇console consumer工做原理分析的随笔。其实不管是哪一个consumer，大部分的工做原理都是相似的。本文利用console consumer做为切入点，既容易理解又不失通常性。

 

本文使用的Kafka环境是0.8.2.1版本，这也是当前最新的版本。(注：Kafka 0.9版本听说会用Java从新设计并编写consumer代码，对此咱们拭目以待) 因为主要目的是分析consumer原理，所以本文并不过多纠结于console consumer特定的使用方法。一条最简单的命令足以做为咱们的开始：

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test-topic

 

kafka-console-consumer.sh脚本内容简洁明了： exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer $@

 

很显然，该shell脚本调用了kafka.tools包下的ConsoleConsumer类，并将提供的命令行参数所有传给该类。由此可知，咱们须要从这个类开始分析。不过在此以前，简单说一下console consumer总体的启动流程，以下图所示：

上图流程具体展开以下：

1. 加载并解析命令行参数，惟一的必要参数(Required)是zookeeper

2. 若是没有传入group.id，ConsoleConsumer将生成本身的group.id，即console-consumer-[10万之内的一个随机数]

3. 建立ConsumerConfig用于封装consumer的各类配置

4. 建立默认的消息格式化类，其定义的writeTo方法会默认将消息输出到控制台

5. 建立ZookeeperConsumerConnector。Kafka使用它来建立KafkaStream消费流

5.1 建立本地缓存， 保存topic下每一个分区的信息，包括该分区底层的阻塞队列，已消费的位移、已获取到的最新位移以及获取大小等

5.2 建立本地缓存，保存每一个topic分区当前在zookeeper中保存的位移值

5.3 建立本地缓存，保存topic的每一个读取线程底层对应的阻塞队列，主要用于关闭Connector时能够批量关闭底层的阻塞队列

5.4 生成consumer id，规则为[group.id]_[主机名]_[时间戳]_[随机产生的一个UUID的前8位]。其中主机名就是运行ConsoleConsumer所在broker节点的主机名

5.5 建立获取线程管理器(ConsumerFetcherManager)

5.6 启动一个特定线程，用于定时地(默认是1分钟)向Zookeeper提交更改过的位移 

6. 增长JVM关闭钩子，确保JVM关闭后资源也可以被释放

7. 建立KafkaStream并经过迭代器不断遍历该stream， KafkaStream的迭代器的底层实现包含一个阻塞队列，若是没有新的消息到来，该迭代器会一直阻塞，除非你显式设置了consumer.timeout.ms参数(默认是-1表示consumer会一直等待新消息的带来)

8. 每接收到一条新的消息，默认的消息格式化类会将其输出到控制台上。而后再次等待迭代器传过来的下一条消息

本质上来讲，console consumer启动时会建立一个KafkaStream(能够简单翻译成Kafak流)，该stream会不停地等待可消费的新消息——具体作法就是经过LinkedBlockingQueue阻塞队列来实现，后续会有详细描述。针对上面启动的顺序列表，咱们在ConsoleConsumer.scala中逐一进行代码走读：

1. 加载必要参数 zookeeper

ConsoleConsumer.scala类定义了main方法，说明这是个可执行的类。类的前100多行几乎都在处理命令行参数的解析。其中真正必要的参数只有zookeeper.connect一个，以下面代码所示：

1 // REQUIRED表示这是一个必需要指定的参数
2 val zkConnectOpt = parser.accepts("zookeeper", "REQUIRED: The connection string for the zookeeper connection in the form host:port. " +
3     "Multiple URLS can be given to allow fail-over.").withRequiredArg.describedAs("urls").ofType(classOf[String])

2. 生成group.id

乍一看和官网上要求的配置不匹配，由于官网中说过consumer真正必要的参数实际上有两个：zookeeper.connect和group.id。由此能够推断console consumer应该会生成group.id的值，且它本质上也是一个consumer，必然属于一个消费组，所以也必然定义了consumer id。下面的代码中即展现了console consumer如何生成本身的group id： (consumer id是如何生成的后面再说)

1 // 若是没有显式指定group.id，那么代码就本身合成一个 2 // 具体格式： console-consumer-[10万之内的一个随机数] 3 // 10万是一个很大的数，所以只有很是低的概率会碰到多个console consumer的group id相同的状况
4 if(!consumerProps.containsKey("group.id")) { 5       consumerProps.put("group.id","console-consumer-" + new Random().nextInt(100000)) 6       groupIdPassed=false
7 }

3. 建立ConsumerConfig对象封装配置

肯定了consumer的group.id以后console consumer须要把传入参数封装进ConsumerConfig类中并把后者传给Consumer的create方法以构造一个ConsumerConnector——即初始化consumer了，具体逻辑见下面的代码：

1 val config = new ConsumerConfig(consumerProps) // 封装ConsumerConfig配置类
2 val skipMessageOnError = if (options.has(skipMessageOnErrorOpt)) true else false

4. 建立默认的消息格式化类，其定义的writeTo方法会默认将消息输出到控制台

1 val messageFormatterClass = Class.forName(options.valueOf(messageFormatterOpt))  // 建立消息格式类，用于最后的输出显示
2 val formatterArgs = CommandLineUtils.parseKeyValueArgs(options.valuesOf(messageFormatterArgOpt)) 3 val maxMessages = if(options.has(maxMessagesOpt)) options.valueOf(maxMessagesOpt).intValue else -1

5. 建立ZookeeperConsumerConnector

ZookeeperConsumerConnector很是重要，它实现了ConsumerConnector接口(该接口定义了建立KafkaStream和提交位移的操做，如createMessageStreams、commitOffsets等)。Kakfa官网把这个接口称为high level的consumer API。对于大多数consumer来讲，这个high level的consumer API提供的功能已经足够了。不过不少用户可能须要对位移有更大的控制，这个时候Kafka推荐用户使用被称为low level的consumer API—— SimpleConsumer。你们参考这篇文章来深刻学习high level API的用法。目前为止，咱们只须要知道Kafka经过下面的语句构建了ConsumerConnector这个consumer的核心接口：

1 val connector = Consumer.create(config) // 建立ConsumerConnector，Consumer核心接口

6. 构建JVM关闭钩子线程 

这部分很是简单，就是在线程中关闭上一步建立的connector，并根据传入的参数决定是否删除zookeeper下/consumers/[group.id]节点

7. 建立KafkaStream，经过迭代器等待消息到来

因为console consumer支持同时消费多个topic的消息，所以它提供了相似于过滤器这样的实现，这也是为何connector调用createMessageStreamsByFilter来建立KafkaStream的缘由，以下面的代码所示。

1 val stream = connector.createMessageStreamsByFilter(filterSpec, 1, new DefaultDecoder(), new DefaultDecoder()).get(0) 2 val iter = if(maxMessages >= 0) 3     stream.slice(0, maxMessages) 4 else
5     stream

createMessageStreamsByFilter方法返回的是一组KafkaStream，但console consumer默认只是建立了1个stream，因此这里直接调用get(0)取到这个stream就能够了。

8. 经过迭代器以阻塞等待的方式消费消息

建立好KafkaStream以后，console consumer经过迭代器遍历KafkaStream。这里值得注意的是，该迭代器底层实现依赖一个阻塞队列。若是没有显式配置过consumer.timeout.ms参数(默认是-1表示consumer会一直等待新消息)，那么迭代器会一直处于阻塞状态等待可供消费的消息——具体的实现细节参见下一篇。迭代器每收到一条消息后，它就会使用默认的消息格式化类DefaultMessageFormatter将消息输出到控制台，这也是console consumer名字的由来，以下面的代码所示：

1 for(messageAndTopic <- iter) { 2     try { 3         formatter.writeTo(messageAndTopic.key, messageAndTopic.message, System.out) // 输出到控制台
4         numMessages += 1
5     } catch { ... } 6  ... 7 }

好了，至此咱们按照启动顺序概述了console consumer启动时的各个阶段。不过，ZookeeperConsumerConnector和建立和迭代器的实现咱们并未详细展开，这部份内容将做为后面续篇的内容呈现给你们。敬请期待！