目錄
最近在拜讀李艷鵬的《可伸縮服務架構——框架與中間件》，該篇隨筆，針對第二章的KClient（kafka消息中間件）源碼解讀項目，進行學習。
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，作為該處理函數的
輸入。@OutputProducer根據輸入參數定義一個Producer，而該處理函數最后返回的Cat對象，將通過該Producer最終傳遞到Kafka中
以下的功能與上述相同，唯一需要注意的是 @InputConsumer和@OutputProducer可以單獨存在。
@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
protected void handleMessage(String message) {
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
protected void handleMessage(final String message) {
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創建並發送過來的，所以實現了最終的消費。
總結：
到此全部的項目解讀完畢，如果仍有疑惑，可以參看李艷鵬老師的《可伸縮服務架構框架與中間件》一書，同時也可以與我聯系交流問題。