[[189769]]

1 背景介紹

消息隊列在互聯(lián)網(wǎng)領(lǐng)域里得到了廣泛的應(yīng)用，它多應(yīng)用在異步處理、模塊之間的解偶和高并發(fā)的消峰等場景，消息隊列中表現(xiàn)最好的當(dāng)屬Apache開源項目Kafka，Kafka使用支持高并發(fā)的Scala語言開發(fā)，利用操作系統(tǒng)的緩存原理達到高性能，并且天生具有可分區(qū)，分布式的特點，而且有不同語言的客戶端，使用起來非常的方便。

Kclient是Kafka生產(chǎn)者客戶端和消費者客戶端的一個簡單易用的框架，它具有高效集成、高性能、高穩(wěn)定的高級特點。

在繼續(xù)閱讀kclient的功能特性、架構(gòu)設(shè)計和使用方法之前，讀者需要對Kafka進行基本的學(xué)習(xí)和了解。如果讀者是Kafka的初學(xué)者，而且英文還不錯，也可以直接參考Kafka官方在線文檔：Kafka 0.8.2 Documentation，如果對英文不感性趣，可以在網(wǎng)上搜索Kakfa的中文資料進行學(xué)習(xí)，內(nèi)容需要涵蓋Kafka的使用向?qū)б约袄貌僮飨到y(tǒng)緩存的“空中接力”、持久化、分片機制、高可用等原理。

本文包含了背景介紹、功能特性、架構(gòu)設(shè)計、使用指南、API簡介、后臺監(jiān)控和管理、消息處理機模板項目、以及性能壓測相關(guān)章節(jié)。如果你想使用kclient快速的構(gòu)建Kafka處理機服務(wù)，請參考消息處理機模板項目章節(jié); 如果你想了解kclient的其他使用方式、功能特性、監(jiān)控和管理等，請參考功能特性、使用指南、API簡介、后臺監(jiān)控和管理等章節(jié); 如果你想更深入的理解kclient的架構(gòu)設(shè)計和性能指標(biāo)，請參考架構(gòu)設(shè)計和性能壓測章節(jié)。

2 功能特性

2.1 簡單易用

簡化了Kafka客戶端API的使用方法, 特別是對消費端開發(fā)，消費端開發(fā)者只需要實現(xiàn)MessageHandler接口或者相關(guān)子類，在實現(xiàn)中處理消息完成業(yè)務(wù)邏輯，并且在主線程中啟動封裝的消費端服務(wù)器即可。它提供了各種常用的MessageHandler，框架自動轉(zhuǎn)換消息到領(lǐng)域?qū)ο竽Ｐ突蛘逬SON對象等數(shù)據(jù)結(jié)構(gòu)，讓開發(fā)者更專注于業(yè)務(wù)處理。如果使用服務(wù)源碼注解的方式聲明消息處理機的后臺，可以將一個通用的服務(wù)方法直接轉(zhuǎn)變成具有完善功能的處理Kafka消息隊列的處理機，使用起來極其簡單，代碼看起來一目了然，在框架級別通過多種線程池技術(shù)保證了處理機的高性能。

在使用方面，它提供了多種使用方式：

1. 直接使用Java API;
2. 與Spring環(huán)境無縫集成;
3. 服務(wù)源碼注解，通過注解聲明方式啟動Kafka消息隊列的處理機。

除此之外，它基于注解提供了消息處理機的模板項目，可以根據(jù)模板項目通過配置快速開發(fā)Kafka的消息處理機。

2.2 高性能

為了在不同的業(yè)務(wù)場景下實現(xiàn)高性能, 它提供不同的線程模型：

適合輕量級服務(wù)的同步線程模型;

適合IO密集型服務(wù)的異步線程模型(細(xì)分為所有消費者流共享線程池和每個流獨享線程池)。

另外，異步模型中的線程池也支持確定數(shù)量線程的線程池和線程數(shù)量可伸縮的線程池。

2.3 高穩(wěn)定性

框架級別處理了常見的異常，計入錯誤日志，可用于錯誤手工恢復(fù)或者洗數(shù)據(jù)，并實現(xiàn)了優(yōu)雅關(guān)機和重啟等功能。

3 架構(gòu)設(shè)計

3.1 線程模型

1. 同步線程模型

在這種線程模型中，客戶端為每一個消費者流使用一個線程，每個線程負(fù)責(zé)從Kafka隊列里消費消息，并且在同一個線程里進行業(yè)務(wù)處理。我們把這些線程稱為消費線程，把這些線程所在的線程池叫做消息消費線程池。這種模型之所以在消息消費線程池處理業(yè)務(wù)，是因為它多用于處理輕量級別的業(yè)務(wù)，例如：緩存查詢、本地計算等。

2. 異步線程模型

在這種線程模型中，客戶端為每一個消費者流使用一個線程，每個線程負(fù)責(zé)從Kafka隊列里消費消息，并且傳遞消費得到的消息到后端的異步線程池，在異步線程池中處理業(yè)務(wù)。我們?nèi)匀话亚懊尕?fù)責(zé)消費消息的線程池稱為消息消費線程池，把后面的異步線程池稱為異步業(yè)務(wù)線程池。這種線程模型適合重量級的業(yè)務(wù)，例如：業(yè)務(wù)中有大量的IO操作、網(wǎng)絡(luò)IO操作、復(fù)雜計算、對外部系統(tǒng)的調(diào)用等。

后端的異步業(yè)務(wù)線程池又細(xì)分為所有消費者流共享線程池和每個流獨享線程池。

1)所有消費者流共享線程池

所有消費者流共享線程池對比每個流獨享線程池，創(chuàng)建更少的線程池對象，能節(jié)省些許的內(nèi)存，但是，由于多個流共享同一個線程池，在數(shù)據(jù)量較大的時候，流之間的處理可能互相影響。例如，一個業(yè)務(wù)使用2個區(qū)和兩個流，他們一一對應(yīng)，通過生產(chǎn)者指定定制化的散列函數(shù)替換默認(rèn)的key-hash, 實現(xiàn)一個流(區(qū))用來處理普通用戶，另外一個流(區(qū))用來處理VIP用戶，如果兩個流共享一個線程池，當(dāng)普通用戶的消息大量產(chǎn)生的時候，VIP用戶只有少量，并且排在了隊列的后頭，就會產(chǎn)生餓死的情況。這個場景是可以使用多個topic來解決，一個普通用戶的topic，一個VIP用戶的topic，但是這樣又要多維護一個topic，客戶端發(fā)送的時候需要顯式的進行判斷topic目標(biāo)，也沒有多少好處。

2)每個流獨享線程池

每個流獨享線程池使用不同的異步業(yè)務(wù)線程池來處理不同的流里面的消息，互相隔離，互相獨立，不互相影響，對于不同的流(區(qū))的優(yōu)先級不同的情況，或者消息在不同流(區(qū))不均衡的情況下表現(xiàn)會更好，當(dāng)然，創(chuàng)建多個線程池會多使用些許內(nèi)存，但是這并不是一個大問題。

另外，異步業(yè)務(wù)線程池支持確定數(shù)量線程的線程池和線程數(shù)量可伸縮的線程池。

1. 核心業(yè)務(wù)硬件資源有保證，核心服務(wù)有專享的資源池，或者線上流量可預(yù)測，請使用固定數(shù)量的線程池。
2. 非核心業(yè)務(wù)一般混布，資源互相調(diào)配，線上流量不固定等情況請使用線程數(shù)量可伸縮的線程池。

3.2 異常處理

對于消息處理過程中產(chǎn)生的業(yè)務(wù)異常，當(dāng)前在業(yè)務(wù)處理的上層捕捉了Throwable, 在專用的錯誤恢復(fù)日志中記錄出錯的消息，后續(xù)可根據(jù)錯誤恢復(fù)日志人工處理錯誤消息，也可重做或者洗數(shù)據(jù)。TODO：考慮實現(xiàn)異常Listener體系結(jié)構(gòu), 對異常處理實現(xiàn)監(jiān)聽者模式，異常處理器可插拔等，默認(rèn)打印錯誤日志。

由于默認(rèn)的異常處理中，捕捉異常，在專用的錯誤回復(fù)日志中記錄錯誤，并且繼續(xù)處理下一個消息?？紤]到可能上線失敗，或者上游消息格式出錯，導(dǎo)致所有消息處理都出錯，堆滿錯誤恢復(fù)日志的情況，我們需要訴諸于報警和監(jiān)控系統(tǒng)來解決。

3.3 優(yōu)雅關(guān)機

由于消費者本身是一個事件驅(qū)動的服務(wù)器，類似Tomcat，Tomcat接收HTTP請求返回HTTP響應(yīng)，Consumer則接收Kafka消息，然后處理業(yè)務(wù)后返回，也可以將處理結(jié)果發(fā)送到下一個消息隊列。所以消費者本身是非常復(fù)雜的，除了線程模型，異常處理，性能，穩(wěn)定性，可用性等都是需要思考點。既然消費者是一個后臺的服務(wù)器，我們需要考慮如何優(yōu)雅的關(guān)機，也就是在消費者服務(wù)器在處理消息的時候，如果關(guān)機才能不導(dǎo)致處理的消息中斷而丟失。

優(yōu)雅關(guān)機的重點在于解決如下3個問題：

1. 如何知道JVM要退出?
2. 如何阻止Daemon的線程在JVM退出被殺掉而導(dǎo)致消息丟失?
3. 如果Worker線程在阻塞，如何喚起并退出?

第一個問題：如果一個后臺程序運行在控制臺的前臺，通過Ctrl + C可以發(fā)送退出信號給JVM，也可以通過kill -2 PS_IS 或者 kill -15 PS_IS發(fā)送退出信號，但是不能發(fā)送kill -9 PS_IS, 否則進程會無條件強制退出。JVM收到退出信號后，會調(diào)用注冊的鉤子，我們通過的注冊的JVM退出鉤子進行優(yōu)雅關(guān)機。

第二個問題：線程分為Daemon線程和非Daemon線程，一個線程默認(rèn)繼承父線程的Daemon屬性，如果當(dāng)前線程池是由Daemon線程創(chuàng)建的，則Worker線程是Daemon線程。如果Worker線程是Daemon線程，我們需要在JVM退出鉤子中等待Worker線程完成當(dāng)前手頭處理的消息，再退出JVM。如果不是Daemon線程，即使JVM收到退出信號，也得等待Worker線程退出后再退出，不會丟掉正在處理的消息。

第三個問題：在Worker線程從Kafka服務(wù)器消費消息的時候，Worker線程可能處于阻塞，這時需要中斷線程以退出，沒有消息被丟掉。在Worker線程處理業(yè)務(wù)時有可能有阻塞，例如：IO，網(wǎng)絡(luò)IO，在指定退出時間內(nèi)沒有完成，我們也需要中斷線程退出，這時會產(chǎn)生一個InterruptedException, 在異常處理的默認(rèn)處理器中被捕捉，并寫入錯誤日志，Worker線程隨后退出。

4 使用指南

kclient提供了三種使用方法，對于每一種方法，按照下面的步驟可快速構(gòu)建Kafka生產(chǎn)者和消費者程序。

4.1 前置步驟

1.下載源代碼后在項目根目錄執(zhí)行如下命令安裝打包文件到你的Maven本地庫。

mvn install 

2.在你的項目pom.xml文件中添加對kclient的依賴。

<dependency> 
    <groupId>com.robert.kafka</groupId> 
    <artifactId>kclient-core</artifactId> 
    <version>0.0.1</version> 
</dependency> 

3.根據(jù)Kafka官方文檔搭建Kafka環(huán)境，并創(chuàng)建兩個Topic， test1和test2。

4.然后，從Kafka安裝目錄的config目錄下拷貝kafka-consumer.properties和kafka-producer.properties到你的項目類路徑下，通常是src/main/resources目錄。

4.2 Java API

Java API提供了最直接，最簡單的使用kclient的方法。

構(gòu)建Producer示例：

KafkaProducer kafkaProducer = new KafkaProducer("kafka-producer.properties", "test"); 
 
for (int i = 0; i < 10; i++) { 
    Dog dog = new Dog(); 
    dog.setName("Yours " + i); 
    dog.setId(i); 
    kafkaProducer.sendBean2Topic("test", dog); 
 
    System.out.format("Sending dog: %d \n", i + 1); 
 
    Thread.sleep(100); 
} 

構(gòu)建Consumer示例：

DogHandler mbe = new DogHandler(); 
 
KafkaConsumer kafkaConsumer = new KafkaConsumer("kafka-consumer.properties", "test", 1, mbe); 
try { 
    kafkaConsumer.startup(); 
 
    try { 
        System.in.read(); 
    } catch (IOException e) { 
        e.printStackTrace(); 
    } 
} finally { 
    kafkaConsumer.shutdownGracefully(); 
} 

public class DogHandler extends BeanMessageHandler<Dog> { 
    public DogHandler() { 
        super(Dog.class); 
    } 
 
    protected void doExecuteBean(Dog dog) { 
        System.out.format("Receiving dog: %s\n", dog); 
    } 
} 

4.3 Spring環(huán)境集成

kclient可以與Spring環(huán)境無縫集成，你可以像使用Spring Bean一樣來使用KafkaProducer和KafkaConsumer。

構(gòu)建Producer示例：

ApplicationContext ac = new ClassPathXmlApplicationContext("kafka-producer.xml"); 
 
KafkaProducer kafkaProducer = (KafkaProducer) ac.getBean("producer"); 
 
for (int i = 0; i < 10; i++) { 
    Dog dog = new Dog(); 
    dog.setName("Yours " + i); 
    dog.setId(i); 
    kafkaProducer.send2Topic("test", JSON.toJSONString(dog)); 
 
    System.out.format("Sending dog: %d \n", i + 1); 
 
    Thread.sleep(100); 
} 

<bean name="producer" class="com.robert.kafka.kclient.core.KafkaProducer" init-method="init"> 
    <property name="propertiesFile" value="kafka-producer.properties"/> 
    <property name="defaultTopic" value="test"/> 
</bean> 

構(gòu)建Consumer示例：

ApplicationContext ac = new ClassPathXmlApplicationContext( 
        "kafka-consumer.xml"); 
 
KafkaConsumer kafkaConsumer = (KafkaConsumer) ac.getBean("consumer"); 
try { 
    kafkaConsumer.startup(); 
 
    try { 
        System.in.read(); 
    } catch (IOException e) { 
        e.printStackTrace(); 
    } 
} finally { 
    kafkaConsumer.shutdownGracefully(); 
} 

public class DogHandler extends BeanMessageHandler<Dog> { 
    public DogHandler() { 
        super(Dog.class); 
    } 
 
    protected void doExecuteBean(Dog dog) { 
        System.out.format("Receiving dog: %s\n", dog); 
    } 
} 

<bean name="dogHandler" class="com.robert.kafka.kclient.sample.api.DogHandler" /> 
 
<bean name="consumer" class="com.robert.kafka.kclient.core.KafkaConsumer" init-method="init"> 
    <property name="propertiesFile" value="kafka-consumer.properties" /> 
    <property name="topic" value="test" /> 
    <property name="streamNum" value="1" /> 
    <property name="handler" ref="dogHandler" /> 
</bean> 

4.4 服務(wù)源碼注解

kclient提供了類似Spring聲明式的編程方法，使用注解聲明Kafka處理器方法，所有的線程模型、異常處理、服務(wù)啟動和關(guān)閉等都由后臺服務(wù)自動完成，極大程度的簡化了API的使用方法，提高了開發(fā)者的工作效率。

注解聲明Kafka消息處理器：

@KafkaHandlers 
public class AnnotatedDogHandler { 
    @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(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); 
    } 
} 

注解啟動程序：

public static void main(String[] args) { 
    ApplicationContext ac = new ClassPathXmlApplicationContext( 
            "annotated-kafka-consumer.xml"); 
 
    try { 
        System.in.read(); 
    } catch (IOException e) { 
        e.printStackTrace(); 
    } 
} 

注解Spring環(huán)境配置：

<bean name="kclientBoot" class="com.robert.kafka.kclient.boot.kclientBoot" init-method="init"/> 
 
<context:component-scan base-package="com.robert.kafka.kclient.sample.annotation" /> 

5 API簡介

5.1 Producer API

KafkaProducer類提供了豐富的API來發(fā)送不同類型的消息，它支持發(fā)送字符串消息，發(fā)送一個普通的Bean，以及發(fā)送JSON對象等。在這些API中可以指定發(fā)送到某個Topic，也可以不指定而使用默認(rèn)的Topic。對于發(fā)送的數(shù)據(jù)，支持帶Key值的消息和不帶Key值的消息。

發(fā)送字符串消息:

public void send(String message); 
public void send2Topic(String topicName, String message);  
public void send(String key, String message);  
public void send2Topic(String topicName, String key, String message);  
public void send(Collection<String> messages);  
public void send2Topic(String topicName, Collection<String> messages);  
public void send(Map<String, String> messages);  
public void send2Topic(String topicName, Map<String, String> messages); 

發(fā)送Bean消息:

public <T> void sendBean(T bean);  
public <T> void sendBean2Topic(String topicName, T bean);  
public <T> void sendBean(String key, T bean);  
public <T> void sendBean2Topic(String topicName, String key, T bean);  
public <T> void sendBeans(Collection<T> beans);  
public <T> void sendBeans2Topic(String topicName, Collection<T> beans);  
public <T> void sendBeans(Map<String, T> beans);  
public <T> void sendBeans2Topic(String topicName, Map<String, T> beans); 

發(fā)送JSON對象消息:

public void sendObject(JSONObject jsonObject);  
public void sendObject2Topic(String topicName, JSONObject jsonObject);  
public void sendObject(String key, JSONObject jsonObject);  
public void sendObject2Topic(String topicName, String key, JSONObject jsonObject);  
public void sendObjects(JSONArray jsonArray);  
public void sendObjects2Topic(String topicName, JSONArray jsonArray);  
public void sendObjects(Map<String, JSONObject> jsonObjects);  
public void sendObjects2Topic(String topicName, Map<String, JSONObject> jsonObjects); 

5.2 Consumer API

KafkaConsumer類提供了豐富的構(gòu)造函數(shù)用來指定Kafka消費者服務(wù)器的各項參數(shù)，包括線程池策略，線程池類型，流數(shù)量等等。

使用PROPERTIES文件初始化:

public KafkaConsumer(String propertiesFile, String topic, int streamNum, MessageHandler handler); 
public KafkaConsumer(String propertiesFile, String topic, int streamNum, int fixedThreadNum, MessageHandler handler); 
public KafkaConsumer(String propertiesFile, String topic, int streamNum, int fixedThreadNum, boolean isSharedThreadPool, MessageHandler handler); 
public KafkaConsumer(String propertiesFile, String topic, int streamNum, int minThreadNum, int maxThreadNum, MessageHandler handler); 
public KafkaConsumer(String propertiesFile, String topic, int streamNum, int minThreadNum, int maxThreadNum, boolean isSharedThreadPool,MessageHandler handler); 

使用PROPERTIES對象初始化:

public KafkaConsumer(Properties properties, String topic, int streamNum, MessageHandler handler); 
public KafkaConsumer(Properties properties, String topic, int streamNum, int fixedThreadNum, MessageHandler handler); 
public KafkaConsumer(Properties properties, String topic, int streamNum, int fixedThreadNum, boolean isSharedThreadPool, MessageHandler handler); 
public KafkaConsumer(Properties properties, String topic, int streamNum, int minThreadNum, int maxThreadNum, MessageHandler handler); 
public KafkaConsumer(Properties properties, String topic, int streamNum, int minThreadNum, int maxThreadNum, boolean isSharedThreadPool,MessageHandler handler); 

5.3 消息處理器

消息處理器結(jié)構(gòu)提供了一個基本接口，并且提供了不同的抽象類實現(xiàn)不同層次的功能，讓功能得到最大化的重用，并且互相解偶，開發(fā)者可以根據(jù)需求選擇某一個抽象類來繼承和使用。

接口定義:

public interface MessageHandler { 
    public void execute(String message); 
} 

安全處理異常抽象類:

public abstract class SafelyMessageHandler implements MessageHandler { 
    public void execute(String message) { 
        try { 
            doExecute(message); 
        } catch (Throwable t) { 
            handleException(t, message); 
        } 
    } 
 
    protected void handleException(Throwable t, String message) { 
        for (ExceptionHandler excepHandler : excepHandlers) { 
            if (t.getClass() == IllegalStateException.class 
                    && t.getCause() != null 
                    && t.getCause().getClass() == InvocationTargetException.class 
                    && t.getCause().getCause() != null) 
                t = t.getCause().getCause(); 
 
            if (excepHandler.support(t)) { 
                try { 
                    excepHandler.handle(t, message); 
                } catch (Exception e) { 
                    log.error( 
                            "Exception hanppens when the handler {} is handling the exception {} and the message {}. Please check if the exception handler is configured properly.", 
                            excepHandler.getClass(), t.getClass(), message); 
                    log.error( 
                            "The stack of the new exception on exception is, ", 
                            e); 
                } 
            } 
        } 
    } 
} 
 
protected abstract void doExecute(String message); 

面向類型的抽象類:

public abstract class BeanMessageHandler<T> extends SafelyMessageHandler {...} 
public abstract class BeansMessageHandler<T> extends SafelyMessageHandler {...} 
public abstract class DocumentMessageHandler extends SafelyMessageHandler {...} 
public abstract class ObjectMessageHandler extends SafelyMessageHandler {...} 
public abstract class ObjectsMessageHandler extends SafelyMessageHandler {...} 

5.4 消息處理器注解

正如上面使用指南第三部分服務(wù)源碼注解所講述的那樣，kclient可以通過注解來聲明Kafka消息處理器，kclient提供了@KafkaHandlers、@InputConsumer、@OutputProducer和@ErrorHandler等注解。

@KafkaHandlers:

@Target({ ElementType.TYPE }) 
@Retention(RetentionPolicy.RUNTIME) 
@Documented 
@Component 
public @interface KafkaHandlers { 
} 

@InputConsumer:

@Target({ ElementType.METHOD }) 
@Retention(RetentionPolicy.RUNTIME) 
@Documented 
public @interface InputConsumer { 
    String propertiesFile() default ""; 
 
    String topic() default ""; 
 
    int streamNum() default 1; 
 
    int fixedThreadNum() default 0; 
 
    int minThreadNum() default 0; 
 
    int maxThreadNum() default 0; 
} 

@OutputProducer:

@Target({ ElementType.METHOD }) 
@Retention(RetentionPolicy.RUNTIME) 
@Documented 
public @interface OutputProducer { 
    String propertiesFile() default ""; 
 
    String defaultTopic() default ""; 
} 

@ErrorHandler:

@Target({ ElementType.METHOD }) 
@Retention(RetentionPolicy.RUNTIME) 
@Documented 
public @interface ErrorHandler { 
    Class<? extends Throwable> exception() default Throwable.class; 
 
    String topic() default ""; 
} 

6 消息處理機模板項目

6.1 快速開發(fā)向?qū)?/strong>

@KafkaHandlers 
public class AnimalsHandler { 
    @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(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); 
    } 
}