KClient——kafka消息中間件源碼解讀
從使用角度上開始入手學習
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 =
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消息中間件源碼解讀