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



最近在拜讀李艷鵬的《可伸縮服務架構——框架與中間件》,該篇隨筆,針對第二章的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" />

具體內容在下一節介紹

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

總結

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

  1. 程序啟動
  2. 調用KClientBoot.init()方法
  3. 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

該類的核心功能是:

  1. 加載配置文件
  2. 初始化線程池
  3. 初始化GracefullyShutdown函數
  4. 初始化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創建並發送過來的,所以實現了最終的消費。

總結:

到此全部的項目解讀完畢,如果仍有疑惑,可以參看李艷鵬老師的《可伸縮服務架構框架與中間件》一書,同時也可以與我聯系交流問題。


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM