從今天開(kāi)始，我們將進(jìn)入 Spring Boot 中另一個(gè)重要話題的討論，即消息通信。

消息通信是 Web 應(yīng)用程序中間層組件中的代表性技術(shù)體系，主要用于構(gòu)建復(fù)雜而又靈活的業(yè)務(wù)流程。在互聯(lián)網(wǎng)應(yīng)用中，消息通信被認(rèn)為是實(shí)現(xiàn)系統(tǒng)解耦和高并發(fā)的關(guān)鍵技術(shù)體系。本節(jié)課我們將在 SpringCSS 案例中引入消息通信機(jī)制來(lái)實(shí)現(xiàn)多個(gè)服務(wù)之間的異步交互。

消息通信機(jī)制與 SpringCSS 案例

在引入消息通信機(jī)制及消息中間件之前，我們先來(lái)梳理下 SpringCSS 中的應(yīng)用場(chǎng)景。

SpringCSS 案例中的消息通信場(chǎng)景

在 SpringCSS 案例中，一個(gè)用戶(hù)的賬戶(hù)信息變動(dòng)并不會(huì)太頻繁。因?yàn)?account-service 和 customer-service 分別位于兩個(gè)服務(wù)中，為了降低遠(yuǎn)程交互的成本，很多時(shí)候我們會(huì)想到先在 customer-service 本地存放一份用戶(hù)賬戶(hù)的拷貝信息，并在客戶(hù)工單生成過(guò)程時(shí)直接從本地?cái)?shù)據(jù)庫(kù)中獲取用戶(hù)賬戶(hù)。

在這樣的設(shè)計(jì)和實(shí)現(xiàn)方式下，如果某個(gè)用戶(hù)的賬戶(hù)信息發(fā)生變化，我們應(yīng)該如何正確且高效地應(yīng)對(duì)呢？此時(shí)消息驅(qū)動(dòng)機(jī)制從系統(tǒng)擴(kuò)展性角度為我們提供了一種很好的實(shí)現(xiàn)方案。

在用戶(hù)賬戶(hù)信息變更時(shí)，account-service 首先會(huì)發(fā)送一個(gè)消息告知某個(gè)用戶(hù)賬戶(hù)信息已經(jīng)發(fā)生變化，然后通知所有對(duì)該消息感興趣的服務(wù)。而在 SpringCSS 案例中，這個(gè)服務(wù)就是 customer-service，相當(dāng)于是這個(gè)消息的訂閱者和消費(fèi)者。

通過(guò)這種方式，customer-service 就可以快速獲取用戶(hù)賬戶(hù)變更消息，從而正確且高效地處理本地的用戶(hù)賬戶(hù)數(shù)據(jù)。

整個(gè)場(chǎng)景的示意圖見(jiàn)下圖：

上圖中我們發(fā)現(xiàn)，消息通信機(jī)制使得我們不必花費(fèi)太大代價(jià)即可實(shí)現(xiàn)整個(gè)交互過(guò)程，簡(jiǎn)單而方便。

消息通信機(jī)制簡(jiǎn)介

消息通信機(jī)制的整體工作流程如下圖所示：

上圖中位于流程中間的就是各種消息中間件，消息中間件一般提供了消息的發(fā)送客戶(hù)端和接收客戶(hù)端組件，這些客戶(hù)端組件會(huì)嵌入業(yè)務(wù)服務(wù)中。

消息的生產(chǎn)者負(fù)責(zé)產(chǎn)生消息，在實(shí)際業(yè)務(wù)中一般由業(yè)務(wù)系統(tǒng)充當(dāng)生產(chǎn)者；而消息的消費(fèi)者負(fù)責(zé)消費(fèi)消息，在實(shí)際業(yè)務(wù)中一般是后臺(tái)系統(tǒng)負(fù)責(zé)異步消費(fèi)。

消息通信有兩種基本模型，即發(fā)布-訂閱（Pub-Sub）模型和點(diǎn)對(duì)點(diǎn)（Point to Point）模型，發(fā)布-訂閱支持生產(chǎn)者消費(fèi)者之間的一對(duì)多關(guān)系，而點(diǎn)對(duì)點(diǎn)模型中有且僅有一個(gè)消費(fèi)者。

上述概念構(gòu)成了消息通信系統(tǒng)最基本的模型，圍繞這個(gè)模型，業(yè)界已經(jīng)有了一些實(shí)現(xiàn)規(guī)范和工具，代表性的規(guī)范有 JMS 、AMQP ，以及它們的實(shí)現(xiàn)框架 ActiveMQ 和 RabbitMQ 等，而 Kafka 等工具并不遵循特定的規(guī)范，但也提供了消息通信的設(shè)計(jì)和實(shí)現(xiàn)方案。

本節(jié)課我們重點(diǎn)關(guān)注 Kafka，后續(xù)的兩個(gè)課時(shí)中我們?cè)俜謩e介紹 ActiveMQ 和 RabbitMQ。

與前面介紹的 JdbcTemplate 和 RestTemplate 類(lèi)似，Spring Boot 作為一款支持快速開(kāi)發(fā)的集成性框架，同樣提供了一批以 -Template 命名的模板工具類(lèi)用于實(shí)現(xiàn)消息通信。對(duì)于 Kafka 而言，這個(gè)工具類(lèi)就是 KafkaTemplate。

使用 KafkaTemplate 集成 Kafka

在討論如何使用 KafkaTemplate 實(shí)現(xiàn)與 Kafka 之間的集成方法之前，我們先來(lái)簡(jiǎn)單了解 Kafka 的基本架構(gòu)，再引出 Kafka 中的幾個(gè)核心概念。

Kafka 基本架構(gòu)

Kafka 基本架構(gòu)參考下圖，從中我們可以看到 Broker、Producer、Consumer、Push、Pull 等消息通信系統(tǒng)常見(jiàn)概念在 Kafka 中都有所體現(xiàn)，生產(chǎn)者使用 Push 模式將消息發(fā)布到 Broker，而消費(fèi)者使用 Pull 模式從 Broker 訂閱消息。

在上圖中我們注意到，Kafka 架構(gòu)圖中還使用了 Zookeeper。

Zookeeper 中存儲(chǔ)了 Kafka 的元數(shù)據(jù)及消費(fèi)者消費(fèi)偏移量（Offset），其作用在于實(shí)現(xiàn) Broker 和消費(fèi)者之間的負(fù)載均衡。因此，如果我們想要運(yùn)行 Kafka，首先需要啟動(dòng) Zookeeper，再啟動(dòng) Kafka 服務(wù)器。

在 Kafka 中還存在 Topic 這么一個(gè)核心概念，它是 Kafka 數(shù)據(jù)寫(xiě)入操作的基本單元，每一個(gè) Topic 可以存在多個(gè)副本（Replication）以確保其可用性。每條消息屬于且僅屬于一個(gè) Topic，因此開(kāi)發(fā)人員通過(guò) Kafka 發(fā)送消息時(shí)，必須指定將該消息發(fā)布到哪個(gè) Topic。同樣，消費(fèi)者訂閱消息時(shí)，也必須指定訂閱來(lái)自哪個(gè) Topic 的信息。

另一方面，從組成結(jié)構(gòu)上講，一個(gè) Topic 中又可以包含一個(gè)或多個(gè)分區(qū)（Partition），因此在創(chuàng)建 Topic 時(shí)我們可以指定 Partition 個(gè)數(shù)。

KafkaTemplate 是 Spring 中提供的基于 Kafka 完成消息通信的模板工具類(lèi)，而要想使用這個(gè)模板工具類(lèi)，我們必須在消息的發(fā)送者和消費(fèi)者應(yīng)用程序中都添加如下 Maven 依賴(lài)：

<dependency>
??? <groupId>org.springframework.kafka</groupId>
??? <artifactId>spring-kafka</artifactId>
</dependency>

使用 KafkaTemplate 發(fā)送消息

KafkaTemplate 提供了一系列 send 方法用來(lái)發(fā)送消息，典型的 send 方法定義如下代碼所示：

@Override
public ListenableFuture<SendResult<K, V>> send(String topic, @Nullable V data) {
}

在上述方法實(shí)際傳入了兩個(gè)參數(shù)，一個(gè)是消息對(duì)應(yīng)的 Topic，另一個(gè)是消息體的內(nèi)容。通過(guò)該方法，我們就能完成最基本的消息發(fā)送過(guò)程。

請(qǐng)注意，在使用 Kafka 時(shí)，我們推薦事先創(chuàng)建好 Topic 供消息生產(chǎn)者和消費(fèi)者使用， 通過(guò)命令行創(chuàng)建 Topic 的方法如下代碼所示：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic springcss.account.topic

這里創(chuàng)建了一個(gè)名為“springcss.account.topic”的 Topic，并指定它的副本數(shù)量和分區(qū)數(shù)量都是 3。

事實(shí)上，我們?cè)谡{(diào)用 KafkaTemplate 的 send 方法時(shí)，如果 Kafka 中不存在該方法中指定的 Topic，它就會(huì)自動(dòng)創(chuàng)建一個(gè)新的 Topic。

另一方面，KafkaTemplate 也提供了一組 sendDefault 方法，它通過(guò)使用默認(rèn)的 Topic 來(lái)發(fā)送消息，如下代碼所示:

@Override
public ListenableFuture<SendResult<K, V>> sendDefault(V data) {
??? return send(this.defaultTopic, data);
}

從代碼中我們可以看到，在上述 sendDefault 方法內(nèi)部中也是使用了 send 方法完成消息的發(fā)送過(guò)程。

那么，如何指定這里的 defaultTopic 呢？在 Spring Boot 中，我們可以使用如下配置項(xiàng)完成這個(gè)工作。

spring:
? kafka:
??? bootstrap-servers:
??? - localhost:9092
??? template:
????? default-topic: demo.topic

現(xiàn)在，我們已經(jīng)了解了通過(guò) KafkaTemplate 發(fā)送消息的實(shí)現(xiàn)方式，KafkaTemplate 高度抽象了消息的發(fā)送過(guò)程，整個(gè)過(guò)程非常簡(jiǎn)單。

接下來(lái)我們切換下視角，看看如何消費(fèi)所發(fā)送的消息。

使用 @KafkaListener 注解消費(fèi)消息

首先需要強(qiáng)調(diào)一點(diǎn)，通過(guò)翻閱 KafkaTemplate 提供的類(lèi)定義，我們并未找到有關(guān)接收消息的任何方法，這實(shí)際上與 Kafka 的設(shè)計(jì)思想有很大關(guān)系。

這點(diǎn)也與本課程后續(xù)要介紹的 JmsTemplate 和 RabbitTemplate 存在很大區(qū)別，因?yàn)樗鼈兌继峁┝嗣鞔_的 receive 方法來(lái)接收消息。

從前面提供的 Kafka 架構(gòu)圖中我們可以看出，在 Kafka 中消息通過(guò)服務(wù)器推送給各個(gè)消費(fèi)者，而 Kafka 的消費(fèi)者在消費(fèi)消息時(shí)，需要提供一個(gè)監(jiān)聽(tīng)器（Listener）對(duì)某個(gè) Topic 實(shí)現(xiàn)監(jiān)聽(tīng)，從而獲取消息，這也是 Kafka 消費(fèi)消息的唯一方式。

在 Spring 中提供了一個(gè) @KafkaListener 注解實(shí)現(xiàn)監(jiān)聽(tīng)器，該注解定義如下代碼所示：

@Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)
@MessageMapping
@Documented
@Repeatable(KafkaListeners.class)
public @interface KafkaListener {
??? String id() default "";
??? String containerFactory() default "";
??? //消息 Topic
??? String[] topics() default {};
??? //Topic 的模式匹配表達(dá)式
??? String topicPattern() default "";
	//Topic 分區(qū)
??? TopicPartition[] topicPartitions() default {};
??? String containerGroup() default "";
??? String errorHandler() default "";
	//消息分組 Id
??? String groupId() default "";
??? boolean idIsGroup() default true;
??? String clientIdPrefix() default "";
??? String beanRef() default "__listener";
}

上述代碼中我們可以看到 @KafkaListener 的定義比較復(fù)雜，我把日常開(kāi)發(fā)中常見(jiàn)的幾個(gè)配置項(xiàng)做了注釋。

在使用 @KafkaListener 時(shí)，最核心的操作是設(shè)置 Topic，而 Kafka 還提供了一個(gè)模式匹配表達(dá)式可以對(duì)目標(biāo) Topic 實(shí)現(xiàn)靈活設(shè)置。

在這里，我們有必要強(qiáng)調(diào)下 groupId 這個(gè)屬性，這就涉及 Kafka 中另一個(gè)核心概念：消費(fèi)者分組（Consumer Group）。

設(shè)計(jì)消費(fèi)者組的目的是應(yīng)對(duì)集群環(huán)境下的多服務(wù)實(shí)例問(wèn)題。顯然，如果采用發(fā)布-訂閱模式會(huì)導(dǎo)致一個(gè)服務(wù)的不同實(shí)例可能會(huì)消費(fèi)到同一條消息。

為了解決這個(gè)問(wèn)題，Kafka 中提供了消費(fèi)者組的概念。一旦我們使用了消費(fèi)組，一條消息只能被同一個(gè)組中的某一個(gè)服務(wù)實(shí)例所消費(fèi)。

消費(fèi)者組的基本結(jié)構(gòu)如下圖所示：

使用 @KafkaListener 注解時(shí)，我們把它直接添加在處理消息的方法上即可，如下代碼所示：

@KafkaListener(topics = “demo.topic”)
public void handlerEvent(DemoEvent event) {
??? //TODO：添加消息處理邏輯
}

當(dāng)然，我們還需要在消費(fèi)者的配置文件中指定用于消息消費(fèi)的配置項(xiàng)，如下代碼所示：

spring:????? 
? kafka:
??? bootstrap-servers:
??? - localhost:9092
??? template:
????? default-topic: demo.topic
??? consumer:
????? group-id: demo.group

可以看到，這里除了指定 template.default-topic 配置項(xiàng)之外，還指定了 consumer. group-id 配置項(xiàng)來(lái)指定消費(fèi)者分組信息。

在 SpringCSS 案例中集成 Kafka

介紹完 KakfaTemplate 的基本原理后，我們將在 SpringCSS 案例中引入 Kafka 實(shí)現(xiàn) account-service 與 customer-service 之間的消息通信。

實(shí)現(xiàn) account-service 消息生產(chǎn)者

首先，我們新建一個(gè) Spring Boot 工程，用來(lái)保存用于多個(gè)服務(wù)之間交互的消息對(duì)象，以供各個(gè)服務(wù)使用。

我們將這個(gè)代碼工程簡(jiǎn)單命名為 message，并添加一個(gè)代表消息主體的事件 AccountChangedEvent，如下代碼所示：

package com.springcss.message;
?
public class AccountChangedEvent implements Serializable {
??? //事件類(lèi)型
??? private String type;
??? //事件所對(duì)應(yīng)的操作（新增、更新和刪除）
??? private String operation;
??? //事件對(duì)應(yīng)的領(lǐng)域模型
??? private AccountMessage accountMessage;
??? //省略 getter/setter
}

上述 AccountChangedEvent 類(lèi)包含了 AccountMessage 對(duì)象本身以及它的操作類(lèi)型，而 AccountMessage 與 Account 對(duì)象的定義完全一致，只不過(guò) AccountMessage 額外實(shí)現(xiàn)了用于序列化的 Serializable 接口而已，如下代碼所示：

public class AccountMessage implements Serializable {
??????? 
??? private Long id;
??? private String accountCode;??? 
??? private String accountName;
}

定義完消息實(shí)體之后，我們?cè)?account-service 中引用了一個(gè) message 工程，并添加了一個(gè) KafkaAccountChangedPublisher 類(lèi)用來(lái)實(shí)現(xiàn)消息的發(fā)布，如下代碼所示：

@Component("kafkaAccountChangedPublisher")
public class KafkaAccountChangedPublisher {
?
??? @Autowired
??? private KafkaTemplate<String, AccountChangedEvent> kafkaTemplate;
??? ? 
??? @Override
??? protected void publishEvent(AccountChangedEvent event) {
??????? kafkaTemplate.send(AccountChannels.SPRINGCSS_ACCOUNT_TOPIC, event);
??? }
}

在這里可以看到，我們注入了一個(gè) KafkaTemplate 對(duì)象，然后通過(guò)它的 send 方法向目標(biāo) Topic 發(fā)送了消息。

這里的 AccountChannels.SPRINGCSS_ACCOUNT_TOPIC 就是 "springcss.account.topic"，我們需要在 account-service 中的配置文件中指定同一個(gè) Topic，如下代碼所示：

spring:
? kafka:
??? bootstrap-servers:
??? - localhost:9092
??? template:
????? default-topic: springcss.account.topic
??? producer:
????? keySerializer: org.springframework.kafka.support.serializer.JsonSerializer
????? valueSerializer: org.springframework.kafka.support.serializer.JsonSerializer

注意到這里，我們使用了 JsonSerializer 對(duì)發(fā)送的消息進(jìn)行序列化。

實(shí)現(xiàn) customer-service 消息消費(fèi)者

針對(duì)服務(wù)消費(fèi)者 customer-service，我們先來(lái)看它的配置信息，如下代碼所示：

spring:????? 
? kafka:
??? bootstrap-servers:
??? - localhost:9092
??? template:
????? default-topic: springcss.account.topic
??? consumer:
????? value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
????? group-id: springcss_customer
????? properties:
??????? spring.json.trusted.packages: com.springcss.message

相較消息生產(chǎn)者中的配置信息，消息消費(fèi)者的配置信息多了兩個(gè)配置項(xiàng)，其中一個(gè)是 group-id，通過(guò)前面內(nèi)容的介紹，我們已經(jīng)知道這是 Kafka 消費(fèi)者特有的一個(gè)配置項(xiàng)，用于指定消費(fèi)者組。

而另一個(gè)配置項(xiàng)是 spring.json.trusted.packages，用于設(shè)置 JSON 序列化的可行包名稱(chēng)，這個(gè)名稱(chēng)需要與 AccountChangedEvent 類(lèi)所在的包結(jié)構(gòu)一致，即這里指定的 com.springcss.message。

小結(jié)與預(yù)告

消息通信機(jī)制是應(yīng)用程序開(kāi)發(fā)過(guò)程中常用的一種技術(shù)體系。在今天的課程中，我們首先基于 SpringCSS 案例梳理了消息通信機(jī)制的應(yīng)用場(chǎng)景，并給出了這一機(jī)制的一些基本概念。然后，基于 Kafka 這款主流的詳細(xì)中間件，我們使用 Spring Boot 提供的 KafkaTemplate 完成了消息的發(fā)送和消費(fèi)，并將其集成到 SpringCSS 案例中。

這里給你留一道思考題：在 Kafka 中，消費(fèi)者組的作用是什么，如何都消費(fèi)者組進(jìn)行配置？

在下一課時(shí)中，我們將繼續(xù)介紹另一款主流的消息中間件 ActiveMQ，以及 Spring Boot 中提供的模板工具類(lèi) JmsTemplate。

精選評(píng)論

**強(qiáng)：

老師請(qǐng)教個(gè)問(wèn)題，就是我的消費(fèi)者隊(duì)列，如何動(dòng)態(tài)的從數(shù)據(jù)庫(kù)里面加載呀？意思就是我的消費(fèi)者隊(duì)列名從數(shù)據(jù)庫(kù)里面查出來(lái)，之后讓kafka監(jiān)聽(tīng)這個(gè)隊(duì)列，這個(gè)請(qǐng)問(wèn)下怎么做呀？

??? 講師回復(fù)：

??? 這個(gè)就跟普通的查詢(xún)數(shù)據(jù)庫(kù)一樣，把查詢(xún)出來(lái)的隊(duì)列名通過(guò)代碼的方式設(shè)置到Kafka中，Kafka既支持配置的方式使用隊(duì)列，也可以使用代碼的方式動(dòng)態(tài)配置文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-757336.html

到了這里，關(guān)于Springboot實(shí)戰(zhàn)14 消息驅(qū)動(dòng)：如何使用 KafkaTemplate 集成 Kafka？的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！