目錄

• kclient消息中間件
  • kclient-processor
    • top.ninwoo.kclient.app.KClientApplication
    • top.ninwoo.kclient.app.KClientController
    • top.ninwoo.kclient.app.handler.AnimalsHandler
    • top.ninwoo.kclient.app.domain
    • 總結(jié)
  • kclient-core
    • top.ninwoo.kafka.kclient.boot.KClientBoot
    • createObjectHandler
    • createObjectsHandler
    • createDocumentHandler
    • createBeanHandler
    • createBeansHandler
    • invokeHandler
    • 生產(chǎn)者和消費(fèi)者創(chuàng)建方法
    • 小結(jié)
  • top.ninwoo.kafka.kclient.boot.KafkaHandlerMeta
  • top.ninwoo.kafka.kclient.core.KafkaProducer
  • top.ninwoo.kafka.kclient.core.KafkaConsumer
    • init()
    • initAsyncThreadPool()
  • initKafka
  • startup()
  • AbstractMessageTask
    • SequentialMessageTask && SequentialMessageTask
  • 總結(jié)：

最近在拜讀李艷鵬的《可伸縮服務(wù)架構(gòu)——框架與中間件》，該篇隨筆，針對第二章的KClient（kafka消息中間件）源碼解讀項(xiàng)目，進(jìn)行學(xué)習(xí)。

kclient消息中間件

從使用角度上開始入手學(xué)習(xí)

kclient-processor

該項(xiàng)目使用springboot調(diào)用kclient庫，程序目錄如下:

• domain
  • Cat ： 定義了一個(gè)cat對象
  • Dog ： 定義了一個(gè)Dog對象
• handler ： 消息處理器
  • AnimalsHandler ： 定義了Cat和Dog的具體行為
• KClientApplication.java ： Spring boot的主函數(shù)——程序執(zhí)行入口
• KClientController.java : Controller 文件

top.ninwoo.kclient.app.KClientApplication

1.啟動(dòng)Spring Boot

ApplicationContext ctxBackend = SpringApplication.run(KClientApplication.class, args);

2.啟動(dòng)程序后將自動(dòng)加載KClientController(@RestController)

top.ninwoo.kclient.app.KClientController

1.通過@RestController，使@SpringBootApplication,可以自動(dòng)加載該Controller

2.通過kafka-application.xml加載Beans

private ApplicationContext ctxKafkaProcessor =new ClassPathXmlApplicationContext("kafka-application.xml");

kafka-application.xml聲明了一個(gè)kclient bean，并設(shè)置其初始化執(zhí)行init方法，具體實(shí)現(xiàn)見下章具體實(shí)現(xiàn)。

<bean name="kClientBoot" class="top.ninwoo.kafka.kclient.boot.KClientBoot" init-method="init"/>

另外聲明了一個(gè)掃描消息處理器的bean

<context:component-scan base-package="top.ninwoo.kclient.app.handler" />

具體內(nèi)容在下一節(jié)介紹

使用@RequestMapping定義/,/status,/stop,/restart定義了不同的接口

這些接口實(shí)現(xiàn)比較簡單，需要注意的是他們調(diào)用的getKClientBoot()函數(shù)。

上文，我們已經(jīng)通過xml中，添加了兩個(gè)Bean，調(diào)用Bean的具體實(shí)現(xiàn)方法如下：

private KClientBoot getKClientBoot() {return (KClientBoot) ctxKafkaProcessor.getBean("kClientBoot");}

通過Bean獲取到KClient獲取到了KClientBoot對象，便可以調(diào)用其具體方法。

top.ninwoo.kclient.app.handler.AnimalsHandler

消息處理函數(shù)

1.使用@KafkaHandlers進(jìn)行聲明bean，關(guān)于其具體實(shí)現(xiàn)及介紹在具體實(shí)現(xiàn)中進(jìn)行介紹

2.定義了三個(gè)處理函數(shù)

• dogHandler
• catHandler
• ioExceptionHandler

dogHandler

具體處理很簡單，主要分析@InputConsumer和@Consumer的作用，具體實(shí)現(xiàn)將在后續(xù)進(jìn)行介紹。

@InputConsumer(propertiesFile = "kafka-consumer.properties", topic = "test", streamNum = 1)@OutputProducer(propertiesFile = "kafka-producer.properties", defaultTopic = "test1")public Cat dogHandler(Dog dog) {System.out.println("Annotated dogHandler handles: " + dog);return new Cat(dog);}

• @InputConsumer根據(jù)輸入?yún)?shù)定義了一個(gè)Consumer，通過該Consumer傳遞具體值給dog，作為該處理函數(shù)的
  輸入。
• @OutputProducer根據(jù)輸入?yún)?shù)定義一個(gè)Producer，而該處理函數(shù)最后返回的Cat對象，將通過該P(yáng)roducer最終傳遞到Kafka中

以下的功能與上述相同，唯一需要注意的是 @InputConsumer和@OutputProducer可以單獨(dú)存在。

@InputConsumer(propertiesFile = "kafka-consumer.properties", topic = "test1", streamNum = 1)public void catHandler(Cat cat) throws IOException {System.out.println("Annotated catHandler handles: " + cat);throw new IOException("Man made exception.");}@ErrorHandler(exception = IOException.class, topic = "test1")public void ioExceptionHandler(IOException e, String message) {System.out.println("Annotated excepHandler handles: " + e);}

top.ninwoo.kclient.app.domain

只是定義了Cat和Dog對象，不做贅述。

總結(jié)

到這里，總結(jié)下我們都實(shí)現(xiàn)了哪些功能？

程序啟動(dòng)

調(diào)用KClientBoot.init()方法

AnimalsHandler定義了消費(fèi)者和生產(chǎn)者的具體方法。

kclient-core

kclient消息中間件的主體部分，該部分將會(huì)涉及

• kafka基本操作
• 反射

項(xiàng)目結(jié)構(gòu)如下:

• boot
  • ErrorHandler
  • InputConsumer
  • OutputProducer
  • KafkaHandlers
  • KClientBoot
  • KafkaHandler
  • KafkaHandlerMeta
• core
  • KafkaConsumer
  • KafkaProducer
• excephandler
  • DefaultExceptionHandler
  • ExceptionHandler
• handlers
  • BeanMessageHandler
  • BeansMessageHandler
  • DocumentMessageHandler
  • ObjectMessageHandler
  • ObjectsMessageHandler
  • MessageHandler
  • SafelyMessageHandler
• reflection.util
  • AnnotationHandler
  • AnnotationTranversor
  • TranversorContext

在接下來的源碼閱讀中，我將按照程序執(zhí)行的順序進(jìn)行解讀。如果其中涉及到?jīng)]有討論過的模塊，讀者可以向下翻閱。這么
做的唯一原因，為了保證思維的連續(xù)性，盡可能不被繁雜的程序打亂。

top.ninwoo.kafka.kclient.boot.KClientBoot

如果讀者剛剛閱讀上一章節(jié)，那么可能記得，我們注冊了一個(gè)kClientBoot的bean，并設(shè)置了初始化函數(shù)init(),所以，在kclient源碼的閱讀中
，我們將從該文件入手，開始解讀。

public void init() {meta = getKafkaHandlerMeta();if (meta.size() == 0)throw new IllegalArgumentException("No handler method is declared in this spring context.");for (final KafkaHandlerMeta kafkaHandlerMeta : meta) {createKafkaHandler(kafkaHandlerMeta);}}

1.該函數(shù)，首先獲取了一個(gè)HandlerMeta，我們可以簡單理解，在這個(gè)數(shù)據(jù)元中，存儲了全部的Handler信息，這個(gè)Handler信息指的是上一章節(jié)中通過@KafkaHandlers定義的處理函數(shù)，
具體實(shí)現(xiàn)見top.ninwoo.kafka.kclient.boot.KafkaHandlerMeta。

2.獲取數(shù)據(jù)元之后，通過循環(huán)，創(chuàng)建對應(yīng)的處理函數(shù)。

for (final KafkaHandlerMeta kafkaHandlerMeta : meta) {createKafkaHandler(kafkaHandlerMeta);}

3.getKafkaHandlerMeta函數(shù)的具體實(shí)現(xiàn)

a.通過applicationContext獲取包含kafkaHandlers注解的Bean名稱。

String[] kafkaHandlerBeanNames = applicationContext.getBeanNamesForAnnotation(KafkaHandlers.class);

b.通過BeanName獲取到Bean對象

Object kafkaHandlerBean = applicationContext.getBean(kafkaHandlerBeanName);Class<? extends Object> kafkaHandlerBeanClazz = kafkaHandlerBean.getClass();

c.構(gòu)建mapData數(shù)據(jù)結(jié)構(gòu),具體構(gòu)建見top.ninwoo.kafka.kclient.reflection.util.AnnotationTranversor

Map<Class<? extends Annotation>, Map<Method, Annotation>> mapData = extractAnnotationMaps(kafkaHandlerBeanClazz);

d.map轉(zhuǎn)數(shù)據(jù)元并添加到數(shù)據(jù)元meta list中。

meta.addAll(convertAnnotationMaps2Meta(mapData, kafkaHandlerBean));

4.循環(huán)遍歷創(chuàng)建kafkaHandler

for (final KafkaHandlerMeta kafkaHandlerMeta : meta) {createKafkaHandler(kafkaHandlerMeta);}

createKafkaHandler()函數(shù)的具體實(shí)現(xiàn)：

a.通過meta獲取clazz中的參數(shù)類型

Class<? extends Object> paramClazz = kafkaHandlerMeta.getParameterType()

b.創(chuàng)建kafkaProducer

KafkaProducer kafkaProducer = createProducer(kafkaHandlerMeta);

c.創(chuàng)建ExceptionHandler

List<ExceptionHandler> excepHandlers = createExceptionHandlers(kafkaHandlerMeta);

d.根據(jù)clazz的參數(shù)類型，選擇消息轉(zhuǎn)換函數(shù)

MessageHandler beanMessageHandler = null;if (paramClazz.isAssignableFrom(JSONObject.class)) {beanMessageHandler = createObjectHandler(kafkaHandlerMeta,kafkaProducer, excepHandlers);} else if (paramClazz.isAssignableFrom(JSONArray.class)) {beanMessageHandler = createObjectsHandler(kafkaHandlerMeta,kafkaProducer, excepHandlers);} else if (List.class.isAssignableFrom(Document.class)) {beanMessageHandler = createDocumentHandler(kafkaHandlerMeta,kafkaProducer, excepHandlers);} else if (List.class.isAssignableFrom(paramClazz)) {beanMessageHandler = createBeansHandler(kafkaHandlerMeta,kafkaProducer, excepHandlers);} else {beanMessageHandler = createBeanHandler(kafkaHandlerMeta,kafkaProducer, excepHandlers);}

e.創(chuàng)建kafkaConsumer,并啟動(dòng)

KafkaConsumer kafkaConsumer = createConsumer(kafkaHandlerMeta,beanMessageHandler);kafkaConsumer.startup();

f.創(chuàng)建KafkaHanlder,并添加到列表中

KafkaHandler kafkaHandler = new KafkaHandler(kafkaConsumer,kafkaProducer, excepHandlers, kafkaHandlerMeta);kafkaHandlers.add(kafkaHandler);

createExceptionHandlers的具體實(shí)現(xiàn)

1.創(chuàng)建一個(gè)異常處理列表

List<ExceptionHandler> excepHandlers = new ArrayList<ExceptionHandler>();

2.從kafkaHandlerMeta獲取異常處理的注解

for (final Map.Entry<ErrorHandler, Method> errorHandler : kafkaHandlerMeta.getErrorHandlers().entrySet()) {

3.創(chuàng)建一個(gè)異常處理對象

ExceptionHandler exceptionHandler = new ExceptionHandler() {public boolean support(Throwable t) {}public void handle(Throwable t, String message) {}

support方法判斷異常類型是否和輸入相同

public boolean support(Throwable t) {// We handle the exception when the classes are exactly samereturn errorHandler.getKey().exception() == t.getClass();}

---

handler方法，進(jìn)一步對異常進(jìn)行處理

1.獲取異常處理方法

Method excepHandlerMethod = errorHandler.getValue();

2.使用Method.invoke執(zhí)行異常處理方法

excepHandlerMethod.invoke(kafkaHandlerMeta.getBean(),t, message);

這里用到了一些反射原理，以下對invoke做簡單介紹

public Object invoke(Object obj,Object... args)throws IllegalAccessException,IllegalArgumentException,InvocationTargetException

參數(shù)：

• obj 從底層方法被調(diào)用的對象
• args 用于方法的參數(shù)

在該項(xiàng)目中的實(shí)際情況如下：

Method實(shí)際對應(yīng)top.ninwoo.kclient.app.handler.AnimalsHandler中的：

@ErrorHandler(exception = IOException.class, topic = "test1")public void ioExceptionHandler(IOException e, String message) {System.out.println("Annotated excepHandler handles: " + e);}

參數(shù)方面：

• kafkaHandlerMeta.getBean() ： AninmalsHandler
• t
• message

invoke完成之后，將會(huì)執(zhí)行ioExceptionHandler函數(shù)

---

4.添加異常處理到列表中

excepHandlers.add(exceptionHandler);

createObjectHandler

createObjectsHandler

createDocumentHandler

createBeanHandler

createBeansHandler

以上均實(shí)現(xiàn)了類似的功能，只是創(chuàng)建了不同類型的對象，然后重寫了不同的執(zhí)行函數(shù)。

實(shí)現(xiàn)原理和異常處理相同，底層都是調(diào)用了invoke函數(shù)，通過反射機(jī)制啟動(dòng)了對應(yīng)的函數(shù)。

下一節(jié)對此做了詳細(xì)介紹

invokeHandler

1.獲取對應(yīng)Method方法

Method kafkaHandlerMethod = kafkaHandlerMeta.getMethod();

2.執(zhí)行接收返回結(jié)果

Object result = kafkaHandlerMethod.invoke(kafkaHandlerMeta.getBean(), parameter);

3.如果生產(chǎn)者非空，意味著需要通過生產(chǎn)者程序?qū)⒔Y(jié)果發(fā)送到Kafka中

if (kafkaProducer != null) {if (result instanceof JSONObject)kafkaProducer.send(((JSONObject) result).toJSONString());else if (result instanceof JSONArray)kafkaProducer.send(((JSONArray) result).toJSONString());else if (result instanceof Document)kafkaProducer.send(((Document) result).getTextContent());elsekafkaProducer.send(JSON.toJSONString(result));

生產(chǎn)者和消費(fèi)者創(chuàng)建方法

protected KafkaConsumer createConsumer(final KafkaHandlerMeta kafkaHandlerMeta,MessageHandler beanMessageHandler) {KafkaConsumer kafkaConsumer = null;if (kafkaHandlerMeta.getInputConsumer().fixedThreadNum() > 0) {kafkaConsumer = new KafkaConsumer(kafkaHandlerMeta.getInputConsumer().propertiesFile(), kafkaHandlerMeta.getInputConsumer().topic(), kafkaHandlerMeta.getInputConsumer().streamNum(), kafkaHandlerMeta.getInputConsumer().fixedThreadNum(), beanMessageHandler);} else if (kafkaHandlerMeta.getInputConsumer().maxThreadNum() > 0&& kafkaHandlerMeta.getInputConsumer().minThreadNum() < kafkaHandlerMeta.getInputConsumer().maxThreadNum()) {kafkaConsumer = new KafkaConsumer(kafkaHandlerMeta.getInputConsumer().propertiesFile(), kafkaHandlerMeta.getInputConsumer().topic(), kafkaHandlerMeta.getInputConsumer().streamNum(), kafkaHandlerMeta.getInputConsumer().minThreadNum(), kafkaHandlerMeta.getInputConsumer().maxThreadNum(), beanMessageHandler);} else {kafkaConsumer = new KafkaConsumer(kafkaHandlerMeta.getInputConsumer().propertiesFile(), kafkaHandlerMeta.getInputConsumer().topic(), kafkaHandlerMeta.getInputConsumer().streamNum(), beanMessageHandler);}return kafkaConsumer;}protected KafkaProducer createProducer(final KafkaHandlerMeta kafkaHandlerMeta) {KafkaProducer kafkaProducer = null;if (kafkaHandlerMeta.getOutputProducer() != null) {kafkaProducer = new KafkaProducer(kafkaHandlerMeta.getOutputProducer().propertiesFile(), kafkaHandlerMeta.getOutputProducer().defaultTopic());}// It may return nullreturn kafkaProducer;}

這兩部分比較簡單，不做贅述。

小結(jié)

KClientBoot.java實(shí)現(xiàn)了：

• 獲取使用KafkaHandlers中定義注釋的方法及其它信息
• 基于反射機(jī)制，生成處理函數(shù)。
• 執(zhí)行處理函數(shù)
• 創(chuàng)建對應(yīng)Producer和Consumer

還剩余幾個(gè)比較簡單的部分，比如shutdownAll()等方法，將在具體實(shí)現(xiàn)處進(jìn)行補(bǔ)充介紹。

到此，整個(gè)項(xiàng)目的主體功能都已經(jīng)實(shí)現(xiàn)。接下來，將分析上文中出現(xiàn)頻率最高的kafkaHandlerMeta與生產(chǎn)者消費(fèi)者的具體實(shí)現(xiàn)。

top.ninwoo.kafka.kclient.boot.KafkaHandlerMeta

KafkaHandlerMeta存儲了全部的可用信息，該類實(shí)現(xiàn)比較簡單，主要分析其成員對象。

• Object bean : 存儲底層的bean對象
• Method method ： 存儲方法對象
• Class<? extends Object> parameterType ： 存儲參數(shù)的類型
• InputConsumer inputConsumer ： 輸入消費(fèi)者注解對象，其中存儲著創(chuàng)建Consumer需要的配置
• OutputProducer outputProducer ： 輸出生產(chǎn)者注解對象，其中存儲著創(chuàng)建Producer需要的配置
• Map<ErrorHandler, Method> errorHandlers = new HashMap<ErrorHandler, Method>() 異常處理函數(shù)與其方法組成的Map

top.ninwoo.kafka.kclient.core.KafkaProducer

該類主要通過多態(tài)封裝了kafka Producer的接口，提供了更加靈活豐富的api接口，比較簡單不做贅述。

top.ninwoo.kafka.kclient.core.KafkaConsumer

該類的核心功能是：

加載配置文件

初始化線程池

初始化GracefullyShutdown函數(shù)

初始化kafka連接

在這里跳過構(gòu)造函數(shù)，但在進(jìn)入核心問題前，先明確幾個(gè)成員變量的作用。

• streamNum : 創(chuàng)建消息流的數(shù)量
• fixedThreadNum ： 異步線程池中的線程數(shù)量
• minThreadNum ： 異步線程池的最小線程數(shù)
• maxThreadNum ： 異步線程池的最大線程數(shù)
• stream : kafka消息流
• streamThreadPool : kafka消息處理線程池

在每個(gè)構(gòu)造函數(shù)后都調(diào)用了init()方法，所以我們從init()入手。另外一個(gè)核心方法startup()將在介紹完init()函數(shù)進(jìn)行介紹。

init()

在執(zhí)行核心代碼前，進(jìn)行了一系列的驗(yàn)證，這里跳過該部分。

1.加載配置文件

properties = loadPropertiesfile();

2.如果共享異步線程池，則初始化異步線程池

sharedAsyncThreadPool = initAsyncThreadPool();

3.初始化優(yōu)雅關(guān)閉

initGracefullyShutdown();

4.初始化kafka連接

initKafka();

initAsyncThreadPool()

完整代碼如下：

private ExecutorService initAsyncThreadPool() {ExecutorService syncThreadPool = null;if (fixedThreadNum > 0)syncThreadPool = Executors.newFixedThreadPool(fixedThreadNum);elsesyncThreadPool = new ThreadPoolExecutor(minThreadNum, maxThreadNum,60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());return syncThreadPool;}

首先，如果異步線程數(shù)大于0，則使用該參數(shù)進(jìn)行創(chuàng)建線程池。

syncThreadPool = Executors.newFixedThreadPool(fixedThreadNum);

如果線程數(shù)不大于0，使用minThreadNum,maxThreadNum進(jìn)行構(gòu)造線程池。

syncThreadPool = new ThreadPoolExecutor(minThreadNum, maxThreadNum,60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());

Executors簡介

這里介紹Executors提供的四種線程池

• newCachedThreadPool創(chuàng)建一個(gè)可緩存線程池，如果線程池長度超過處理需要，可靈活回收空閑線程，若無可回收，則新建線程。
• newFixedThreadPool 創(chuàng)建一個(gè)定長線程池，可控制線程最大并發(fā)數(shù)，超出的線程會(huì)在隊(duì)列中等待。
• newScheduledThreadPool 創(chuàng)建一個(gè)定長線程池，支持定時(shí)及周期性任務(wù)執(zhí)行。
• newSingleThreadExecutor 創(chuàng)建一個(gè)單線程化的線程池，它只會(huì)用唯一的工作線程來執(zhí)行任務(wù)，保證所有任務(wù)按照指定順序(FIFO, LIFO, 優(yōu)先級)執(zhí)行。

ThreadPoolExecutor簡介

ThreadPooExecutor與Executor的關(guān)系如下：

構(gòu)造方法：

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue)；ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler)ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory)ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)

參數(shù)說明：

• corePoolSize

核心線程數(shù)，默認(rèn)情況下核心線程會(huì)一直存活，即使處于閑置狀態(tài)也不會(huì)受存keepAliveTime限制。除非將allowCoreThreadTimeOut設(shè)置為true。

• maximumPoolSize

線程池所能容納的最大線程數(shù)。超過這個(gè)數(shù)的線程將被阻塞。當(dāng)任務(wù)隊(duì)列為沒有設(shè)置大小的LinkedBlockingDeque時(shí)，這個(gè)值無效。

• keepAliveTime

非核心線程的閑置超時(shí)時(shí)間，超過這個(gè)時(shí)間就會(huì)被回收。

• unit

指定keepAliveTime的單位，如TimeUnit.SECONDS。當(dāng)將allowCoreThreadTimeOut設(shè)置為true時(shí)對corePoolSize生效。

• workQueue

線程池中的任務(wù)隊(duì)列.

常用的有三種隊(duì)列，SynchronousQueue,LinkedBlockingDeque,ArrayBlockingQueue。

• SynchronousQueue

線程工廠，提供創(chuàng)建新線程的功能。

• RejectedExecutionHandler

當(dāng)線程池中的資源已經(jīng)全部使用，添加新線程被拒絕時(shí)，會(huì)調(diào)用RejectedExecutionHandler的rejectedExecution方法。

initKafka

由于kafka API已經(jīng)改動(dòng)很多，所以這里關(guān)于Kafka的操作僅做參考，不會(huì)詳細(xì)介紹。

1.加載Consumer配置

ConsumerConfig config = new ConsumerConfig(properties);

2.創(chuàng)建consumerConnector連接

consumerConnector = Consumer.createJavaConsumerConnector(config);

3.存儲kafka topic與對應(yīng)設(shè)置的消息流數(shù)量

Map<String, Integer> topics = new HashMap<String, Integer>(); topics.put(topic, streamNum);

4.從kafka獲取消息流

Map<String, List<KafkaStream<String, String>>> streamsMap = consumerConnector.createMessageStreams(topics, keyDecoder, valueDecoder); streams = streamsMap.get(topic);

5.創(chuàng)建消息處理線程池

startup()

上述init()主要介紹了kafka消費(fèi)者的初始化，而startup()則是kafkaConsumer作為消費(fèi)者進(jìn)行消費(fèi)動(dòng)作的核心功能代碼。

1.依次處理消息線程streams中的消息

for (KafkaStream<String, String> stream : streams) {

2.創(chuàng)建消息任務(wù)

AbstractMessageTask abstractMessageTask = (fixedThreadNum == 0 ? new SequentialMessageTask(stream, handler) : new ConcurrentMessageTask(stream, handler, fixedThreadNum));

3.添加到tasks中，以方便關(guān)閉進(jìn)程

tasks.add(abstractMessageTask);

4.執(zhí)行任務(wù)

streamThreadPool.execute(abstractMessageTask);

AbstractMessageTask

任務(wù)執(zhí)行的抽象類，核心功能如下從消息線程池中不斷獲取消息，進(jìn)行消費(fèi)。
下面是完整代碼，不再詳細(xì)介紹：

abstract class AbstractMessageTask implements Runnable {protected KafkaStream<String, String> stream;protected MessageHandler messageHandler;AbstractMessageTask(KafkaStream<String, String> stream,MessageHandler messageHandler) {this.stream = stream;this.messageHandler = messageHandler;}public void run() {ConsumerIterator<String, String> it = stream.iterator();while (status == Status.RUNNING) {boolean hasNext = false;try {// When it is interrupted if process is killed, it causes some duplicate message processing, because it commits the message in a chunk every 30 secondshasNext = it.hasNext();} catch (Exception e) {// hasNext() method is implemented by scala, so no checked// exception is declared, in addtion, hasNext() may throw// Interrupted exception when interrupted, so we have to// catch Exception here and then decide if it is interrupted// exceptionif (e instanceof InterruptedException) {log.info("The worker [Thread ID: {}] has been interrupted when retrieving messages from kafka broker. Maybe the consumer is shutting down.",Thread.currentThread().getId());log.error("Retrieve Interrupted: ", e);if (status != Status.RUNNING) {it.clearCurrentChunk();shutdown();break;}} else {log.error("The worker [Thread ID: {}] encounters an unknown exception when retrieving messages from kafka broker. Now try again.",Thread.currentThread().getId());log.error("Retrieve Error: ", e);continue;}}if (hasNext) {MessageAndMetadata<String, String> item = it.next();log.debug("partition[" + item.partition() + "] offset["+ item.offset() + "] message[" + item.message()+ "]");handleMessage(item.message());// if not auto commit, commit it manuallyif (!isAutoCommitOffset) {consumerConnector.commitOffsets();}}}protected void shutdown() {// Actually it doesn't work in auto commit mode, because kafka v0.8 commits once per 30 seconds, so it is bound to consume duplicate messages.stream.clear();}protected abstract void handleMessage(String message);}

SequentialMessageTask && SequentialMessageTask

或許您還比較迷惑如何在這個(gè)抽象類中實(shí)現(xiàn)我們具體的消費(fèi)方法，實(shí)際上是通過子類實(shí)現(xiàn)handleMessage方法進(jìn)行綁定我們具體的消費(fèi)方法。

class SequentialMessageTask extends AbstractMessageTask {SequentialMessageTask(KafkaStream<String, String> stream,MessageHandler messageHandler) {super(stream, messageHandler);}@Overrideprotected void handleMessage(String message) {messageHandler.execute(message);}}

在該子類中，handleMessage直接執(zhí)行了messageHandler.execute(message),而沒有調(diào)用線程池，所以是順序消費(fèi)消息。

class ConcurrentMessageTask extends AbstractMessageTask {private ExecutorService asyncThreadPool;ConcurrentMessageTask(KafkaStream<String, String> stream,MessageHandler messageHandler, int threadNum) {super(stream, messageHandler);if (isSharedAsyncThreadPool)asyncThreadPool = sharedAsyncThreadPool;else {asyncThreadPool = initAsyncThreadPool();}}@Overrideprotected void handleMessage(final String message) {asyncThreadPool.submit(new Runnable() {public void run() {// if it blows, how to recovermessageHandler.execute(message);}});}protected void shutdown() {if (!isSharedAsyncThreadPool)shutdownThreadPool(asyncThreadPool, "async-pool-"+ Thread.currentThread().getId());}}

在ConcurrentMessageTask中， handleMessage調(diào)用asyncThreadPool.submit()提交了任務(wù)到異步線程池中，是一個(gè)并發(fā)消費(fèi)。

而messageHandler是通過KClientBoot的createKafkaHandler創(chuàng)建并發(fā)送過來的，所以實(shí)現(xiàn)了最終的消費(fèi)。

總結(jié)：

到此全部的項(xiàng)目解讀完畢，如果仍有疑惑，可以參看李艷鵬老師的《可伸縮服務(wù)架構(gòu)框架與中間件》一書，同時(shí)也可以與我聯(lián)系交流問題。

轉(zhuǎn)載于:https://www.cnblogs.com/NinWoo/p/9798270.html

總結(jié)

以上是生活随笔為你收集整理的KClient——kafka消息中间件源码解读的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。