1. 程式人生 > >KClient——kafka消息中間件源碼解讀

KClient——kafka消息中間件源碼解讀

等待 技術 列表 extra 結果 default 返回 isa info

kclient消息中間件
從使用角度上開始入手學習

kclient-processor

該項目使用springboot調用kclient庫,程序目錄如下:

domain
Cat : 定義了一個cat對象
Dog : 定義了一個Dog對象
handler : 消息處理器
AnimalsHandler : 定義了Cat和Dog的具體行為
KClientApplication.java : Spring boot的主函數——程序執行入口
KClientController.java : Controller 文件
top.ninwoo.kclient.app.KClientApplication

1.啟動Spring Boot

ApplicationContext ctxBackend = SpringApplication.run(
KClientApplication.class, args);
2.啟動程序後將自動加載KClientController(@RestController)

top.ninwoo.kclient.app.KClientController

1.通過@RestController,使@SpringBootApplication,可以自動加載該Controller

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

private ApplicationContext ctxKafkaProcessor =

new ClassPathXmlApplicationContext("kafka-application.xml");
kafka-application.xml聲明了一個kclient bean,並設置其初始化執行init方法,具體實現見下章具體實現。

<bean name="kClientBoot" class="top.ninwoo.kafka.kclient.boot.KClientBoot" init-method="init"/>
另外聲明了一個掃描消息處理器的bean

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

具體內容在下一節介紹

使用@RequestMapping定義/,/status,/stop,/restart定義了不同的接口
這些接口實現比較簡單,需要註意的是他們調用的getKClientBoot()函數。

上文,我們已經通過xml中,添加了兩個Bean,調用Bean的具體實現方法如下:

private KClientBoot getKClientBoot() {
return (KClientBoot) ctxKafkaProcessor.getBean("kClientBoot");
}
通過Bean獲取到KClient獲取到了KClientBoot對象,便可以調用其具體方法。

top.ninwoo.kclient.app.handler.AnimalsHandler

消息處理函數

1.使用@KafkaHandlers進行聲明bean,關於其具體實現及介紹在具體實現中進行介紹

2.定義了三個處理函數

dogHandler
catHandler
ioExceptionHandler
dogHandler

具體處理很簡單,主要分析@InputConsumer和@Consumer的作用,具體實現將在後續進行介紹。

@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根據輸入參數定義了一個Consumer,通過該Consumer傳遞具體值給dog,作為該處理函數的
br/>}
@InputConsumer根據輸入參數定義了一個Consumer,通過該Consumer傳遞具體值給dog,作為該處理函數的
br/>@OutputProducer根據輸入參數定義一個Producer,而該處理函數最後返回的Cat對象,將通過該Producer最終傳遞到Kafka中

@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對象,不做贅述。

總結

到這裏,總結下我們都實現了哪些功能?

程序啟動
調用KClientBoot.init()方法
AnimalsHandler定義了消費者和生產者的具體方法。
kclient-core

kclient消息中間件的主體部分,該部分將會涉及

kafka基本操作
反射
項目結構如下:

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
在接下來的源碼閱讀中,我將按照程序執行的順序進行解讀。如果其中涉及到沒有討論過的模塊,讀者可以向下翻閱。這麽

做的唯一原因,為了保證思維的連續性,盡可能不被繁雜的程序打亂。

top.ninwoo.kafka.kclient.boot.KClientBoot

如果讀者剛剛閱讀上一章節,那麽可能記得,我們註冊了一個kClientBoot的bean,並設置了初始化函數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.該函數,首先獲取了一個HandlerMeta,我們可以簡單理解,在這個數據元中,存儲了全部的Handler信息,這個Handler信息指的是上一章節中通過@KafkaHandlers定義的處理函數,

具體實現見top.ninwoo.kafka.kclient.boot.KafkaHandlerMeta。

2.獲取數據元之後,通過循環,創建對應的處理函數。

for (final KafkaHandlerMeta kafkaHandlerMeta : meta) {
createKafkaHandler(kafkaHandlerMeta);
}
3.getKafkaHandlerMeta函數的具體實現

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.構建mapData數據結構,具體構建見top.ninwoo.kafka.kclient.reflection.util.AnnotationTranversor

Map<Class<? extends Annotation>, Map<Method, Annotation>> mapData = extractAnnotationMaps(kafkaHandlerBeanClazz);
d.map轉數據元並添加到數據元meta list中。

meta.addAll(convertAnnotationMaps2Meta(mapData, kafkaHandlerBean));
4.循環遍歷創建kafkaHandler

for (final KafkaHandlerMeta kafkaHandlerMeta : meta) {
createKafkaHandler(kafkaHandlerMeta);
}
createKafkaHandler()函數的具體實現:

a.通過meta獲取clazz中的參數類型

Class<? extends Object> paramClazz = kafkaHandlerMeta
.getParameterType()
b.創建kafkaProducer

KafkaProducer kafkaProducer = createProducer(kafkaHandlerMeta);
c.創建ExceptionHandler

List<ExceptionHandler> excepHandlers = createExceptionHandlers(kafkaHandlerMeta);
d.根據clazz的參數類型,選擇消息轉換函數

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.創建kafkaConsumer,並啟動

KafkaConsumer kafkaConsumer = createConsumer(kafkaHandlerMeta,
beanMessageHandler);
kafkaConsumer.startup();
f.創建KafkaHanlder,並添加到列表中

KafkaHandler kafkaHandler = new KafkaHandler(kafkaConsumer,
kafkaProducer, excepHandlers, kafkaHandlerMeta);
kafkaHandlers.add(kafkaHandler);
createExceptionHandlers的具體實現

1.創建一個異常處理列表

List<ExceptionHandler> excepHandlers = new ArrayList<ExceptionHandler>();
2.從kafkaHandlerMeta獲取異常處理的註解

for (final Map.Entry<ErrorHandler, Method> errorHandler : kafkaHandlerMeta
.getErrorHandlers().entrySet()) {
3.創建一個異常處理對象

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 same
return errorHandler.getKey().exception() == t.getClass();
}
handler方法,進一步對異常進行處理

1.獲取異常處理方法

Method excepHandlerMethod = errorHandler.getValue();
2.使用Method.invoke執行異常處理方法

excepHandlerMethod.invoke(kafkaHandlerMeta.getBean(),
t, message);
這裏用到了一些反射原理,以下對invoke做簡單介紹

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

obj 從底層方法被調用的對象
args 用於方法的參數
在該項目中的實際情況如下:

Method實際對應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);
}
參數方面:

kafkaHandlerMeta.getBean() : AninmalsHandler
t
message
invoke完成之後,將會執行ioExceptionHandler函數

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

excepHandlers.add(exceptionHandler);
createObjectHandler

createObjectsHandler

createDocumentHandler

createBeanHandler

createBeansHandler

以上均實現了類似的功能,只是創建了不同類型的對象,然後重寫了不同的執行函數。

實現原理和異常處理相同,底層都是調用了invoke函數,通過反射機制啟動了對應的函數。

下一節對此做了詳細介紹

invokeHandler

1.獲取對應Method方法

Method kafkaHandlerMethod = kafkaHandlerMeta.getMethod();
2.執行接收返回結果

Object result = kafkaHandlerMethod.invoke(
kafkaHandlerMeta.getBean(), parameter);
3.如果生產者非空,意味著需要通過生產者程序將結果發送到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());
else
kafkaProducer.send(JSON.toJSONString(result));
生產者和消費者創建方法

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 null
return kafkaProducer;
}
這兩部分比較簡單,不做贅述。

小結

KClientBoot.java實現了:

獲取使用KafkaHandlers中定義註釋的方法及其它信息
基於反射機制,生成處理函數。
執行處理函數
創建對應Producer和Consumer
還剩余幾個比較簡單的部分,比如shutdownAll()等方法,將在具體實現處進行補充介紹。

到此,整個項目的主體功能都已經實現。接下來,將分析上文中出現頻率最高的kafkaHandlerMeta與生產者消費者的具體實現。

top.ninwoo.kafka.kclient.boot.KafkaHandlerMeta

KafkaHandlerMeta存儲了全部的可用信息,該類實現比較簡單,主要分析其成員對象。

Object bean : 存儲底層的bean對象
Method method : 存儲方法對象
Class<? extends Object> parameterType : 存儲參數的類型
InputConsumer inputConsumer : 輸入消費者註解對象,其中存儲著創建Consumer需要的配置
OutputProducer outputProducer : 輸出生產者註解對象,其中存儲著創建Producer需要的配置
Map<ErrorHandler, Method> errorHandlers = new HashMap<ErrorHandler, Method>() 異常處理函數與其方法組成的Map
top.ninwoo.kafka.kclient.core.KafkaProducer

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

top.ninwoo.kafka.kclient.core.KafkaConsumer

該類的核心功能是:

加載配置文件
初始化線程池
初始化GracefullyShutdown函數
初始化kafka連接
在這裏跳過構造函數,但在進入核心問題前,先明確幾個成員變量的作用。

streamNum : 創建消息流的數量
fixedThreadNum : 異步線程池中的線程數量
minThreadNum : 異步線程池的最小線程數
maxThreadNum : 異步線程池的最大線程數
stream : kafka消息流
streamThreadPool : kafka消息處理線程池
在每個構造函數後都調用了init()方法,所以我們從init()入手。另外一個核心方法startup()將在介紹完init()函數進行介紹。

init()

在執行核心代碼前,進行了一系列的驗證,這裏跳過該部分。

1.加載配置文件

properties = loadPropertiesfile();
2.如果共享異步線程池,則初始化異步線程池

sharedAsyncThreadPool = initAsyncThreadPool();
3.初始化優雅關閉

initGracefullyShutdown();
4.初始化kafka連接

initKafka();
initAsyncThreadPool()

完整代碼如下:

private ExecutorService initAsyncThreadPool() {
ExecutorService syncThreadPool = null;
if (fixedThreadNum > 0)
syncThreadPool = Executors.newFixedThreadPool(fixedThreadNum);
else
syncThreadPool = new ThreadPoolExecutor(minThreadNum, maxThreadNum,
60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
return syncThreadPool;
}
首先,如果異步線程數大於0,則使用該參數進行創建線程池。

syncThreadPool = Executors.newFixedThreadPool(fixedThreadNum);
如果線程數不大於0,使用minThreadNum,maxThreadNum進行構造線程池。

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

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

newCachedThreadPool創建一個可緩存線程池,如果線程池長度超過處理需要,可靈活回收空閑線程,若無可回收,則新建線程。
newFixedThreadPool 創建一個定長線程池,可控制線程最大並發數,超出的線程會在隊列中等待。
newScheduledThreadPool 創建一個定長線程池,支持定時及周期性任務執行。
newSingleThreadExecutor 創建一個單線程化的線程池,它只會用唯一的工作線程來執行任務,保證所有任務按照指定順序(FIFO, LIFO, 優先級)執行。
ThreadPoolExecutor簡介

ThreadPooExecutor與Executor的關系如下:

技術分享圖片
構造方法:

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)
參數說明:

corePoolSize
核心線程數,默認情況下核心線程會一直存活,即使處於閑置狀態也不會受存keepAliveTime限制。除非將allowCoreThreadTimeOut設置為true。

maximumPoolSize
線程池所能容納的最大線程數。超過這個數的線程將被阻塞。當任務隊列為沒有設置大小的LinkedBlockingDeque時,這個值無效。

keepAliveTime
非核心線程的閑置超時時間,超過這個時間就會被回收。

unit
指定keepAliveTime的單位,如TimeUnit.SECONDS。當將allowCoreThreadTimeOut設置為true時對corePoolSize生效。

workQueue
線程池中的任務隊列.

常用的有三種隊列,SynchronousQueue,LinkedBlockingDeque,ArrayBlockingQueue。

SynchronousQueue
線程工廠,提供創建新線程的功能。

RejectedExecutionHandler
當線程池中的資源已經全部使用,添加新線程被拒絕時,會調用RejectedExecutionHandler的rejectedExecution方法。

initKafka

由於kafka API已經改動很多,所以這裏關於Kafka的操作僅做參考,不會詳細介紹。

1.加載Consumer配置

ConsumerConfig config = new ConsumerConfig(properties);
2.創建consumerConnector連接

consumerConnector = Consumer.createJavaConsumerConnector(config);
3.存儲kafka topic與對應設置的消息流數量

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.創建消息處理線程池

startup()

上述init()主要介紹了kafka消費者的初始化,而startup()則是kafkaConsumer作為消費者進行消費動作的核心功能代碼。

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

for (KafkaStream<String, String> stream : streams) {
2.創建消息任務

AbstractMessageTask abstractMessageTask = (fixedThreadNum == 0 ? new SequentialMessageTask(
stream, handler) : new ConcurrentMessageTask(stream, handler, fixedThreadNum));
3.添加到tasks中,以方便關閉進程

tasks.add(abstractMessageTask);
4.執行任務

streamThreadPool.execute(abstractMessageTask);
AbstractMessageTask

任務執行的抽象類,核心功能如下從消息線程池中不斷獲取消息,進行消費。

下面是完整代碼,不再詳細介紹:

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 seconds
hasNext = 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
// exception
if (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 manually
    if (!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

或許您還比較迷惑如何在這個抽象類中實現我們具體的消費方法,實際上是通過子類實現handleMessage方法進行綁定我們具體的消費方法。

class SequentialMessageTask extends AbstractMessageTask {
SequentialMessageTask(KafkaStream<String, String> stream,
MessageHandler messageHandler) {
super(stream, messageHandler);}
@Override
br/>}
@Override
messageHandler.execute(message);
}
}
在該子類中,handleMessage直接執行了messageHandler.execute(message),而沒有調用線程池,所以是順序消費消息。

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();}
}
@Override
br/>}
}
@Override
asyncThreadPool.submit(new Runnable() {
public void run() {
// if it blows, how to recover
messageHandler.execute(message);
}
});
}
protected void shutdown() {
if (!isSharedAsyncThreadPool)
shutdownThreadPool(asyncThreadPool, "async-pool-"

  • Thread.currentThread().getId());
    }
    }
    在ConcurrentMessageTask中, handleMessage調用asyncThreadPool.submit()提交了任務到異步線程池中,是一個並發消費。

而messageHandler是通過KClientBoot的createKafkaHandler創建並發送過來的,所以實現了最終的消費。

KClient——kafka消息中間件源碼解讀