發(fā)布-訂閱消息系統(tǒng)在任何企業(yè)體系結構中都起著重要作用，因為它可以實現(xiàn)可靠的集成而無需緊密耦合應用程序。 在解耦的系統(tǒng)之間共享數(shù)據(jù)的能力并不是一個容易解決的問題。

考慮一個企業(yè)，其中具有使用不同語言和平臺獨立構建的多個應用程序。 它需要以響應方式共享數(shù)據(jù)和流程。 我們可以使用Messaging通過使用可自定義的格式頻繁，立即，可靠且異步地傳輸數(shù)據(jù)包來實現(xiàn)此目的。 從根本上說，異步消息傳遞是對分布式系統(tǒng)問題的務實反應。 發(fā)送消息不需要兩個系統(tǒng)同時啟動和就緒。

發(fā)布-訂閱頻道

從簡單的角度來看，通過添加用于傳達事件通知的事件通道的概念，對這種模式的理解依賴于它對觀察者模式的擴展。 觀察者模式描述了將觀察者與他們的主題脫鉤的必要性，這樣，無論有多少觀察者，主題都可以輕松地向所有感興趣的觀察者提供事件通知。

每個訂閱者都需要一次被通知特定事件，但不應重復通知同一事件。 在通知所有訂戶之前，不能認為事件已消耗。 但是，一旦所有訂戶都得到通知，該事件就可以視為已消耗，應該從通道中消失[2]。

代理，隊列，主題和訂閱

代理消息傳遞支持真實時間解耦系統(tǒng)的場景，在這種情況下，不能保證消息生產(chǎn)者或使用者的可用性。 對于代理消息傳遞，隊列是保留由生產(chǎn)者創(chuàng)建的消息的代理，并且消費者在準備就緒時可以在其中檢索消息。

隊列提供了最簡單的郵件傳遞選項。 隊列中的消息是按照先進先出（FIFO）進行組織的，并且每個消息都應由單個使用者處理。 但是，主題和訂閱構成了一種發(fā)布/訂閱模式，允許N個消費者處理同一條消息。

可以將單個消息添加到主題，并且對于滿足的每個訂閱規(guī)則，將消息的副本添加到該訂閱。 在這種情況下，每個訂閱都會成為隊列，消費者可以在其中單獨處理訂閱上的消息。

行業(yè)領先者正在使用的可靠且成熟的項目之一是Apache Kafka，它為我們提供了每秒處理大量消息的能力，而不是傳統(tǒng)的消息傳遞系統(tǒng)，該系統(tǒng)在傳統(tǒng)場景中非常有用，但效率和價值卻不高在處理大數(shù)據(jù)場景中。

除消息傳遞外，Apache Kafka還可以應用于流處理，網(wǎng)站活動跟蹤，日志聚合，指標，基于時間的消息存儲，提交日志和事件源。 在下一節(jié)中，我們將深入介紹Apache Kafka的組件和特征。

卡夫卡

Kafka是一個分布式的發(fā)布-訂閱消息系統(tǒng)，通過其設計，分區(qū)和復制的提交日志服務，其本質(zhì)上是快速，可伸縮的并且分布式的。 它與傳統(tǒng)消息傳遞系統(tǒng)的不同之處在于，它非常易于擴展，提供高吞吐量，支持多訂戶并在故障期間自動平衡使用者，并具有允許實時應用程序或ETL將其用作批處理消耗的能力磁盤上的持久消息數(shù)。

組件[1]

• 生產(chǎn)者 –生產(chǎn)者是將消息發(fā)布給Kafka經(jīng)紀人的任何應用程序/程序。

• 使用者 -使用者是使用來自Kafka經(jīng)紀人的消息的應用程序。 這些使用者可以是簡單的應用程序，實時流處理引擎等。
• 主題和分區(qū) – Apache Kafka支持消息主題的概念，這些主題允許對消息進行分類。 它使我們能夠為不同類型的消息創(chuàng)建不同的主題，并讓不同的消費者使用消息。 此外，Apache Kafka允許在一個Topic中創(chuàng)建多個分區(qū)，以允許并行使用消息，因為我們可以同時從不同的分區(qū)中消費不同的使用者。 每個分區(qū)都有一個領導節(jié)點，負責處理來自消費者/生產(chǎn)者對該分區(qū)的讀/寫請求。

• 代理 – Kafka代理通常是指安裝了Kafka的計算機。 但是，可以在非生產(chǎn)設置中在一臺計算機上設置多個代理。 Kafka經(jīng)紀人負責管理消息日志并接受生產(chǎn)者/消費者的請求。 卡夫卡經(jīng)紀人是無國籍的。 這意味著消費者必須保持已經(jīng)消費了多少。 消費者自己維護它，經(jīng)紀人不會做任何事情。
• 存儲 – Kafka具有非常簡單的存儲布局。 主題的每個分區(qū)都對應一個邏輯日志。 從物理上講，日志是作為一組大小相等的段文件實現(xiàn)的。 每次生產(chǎn)者將消息發(fā)布到分區(qū)時，代理都將消息簡單地附加到最后一個段文件。 在發(fā)布了可配置數(shù)量的消息或經(jīng)過一定時間后，段文件將刷新到磁盤。 消息在清除后會暴露給使用者。
• 集群 – Kafka集群是Kafka經(jīng)紀人的集合。 集群中的所有Kafka經(jīng)紀人共同工作，以管理已配置的消息及其副本。

動物園管理員

ZooKeeper用于管理和協(xié)調(diào)Kafka經(jīng)紀人。 每個Kafka經(jīng)紀人都使用ZooKeeper與其他Kafka經(jīng)紀人進行協(xié)調(diào)。 ZooKeeper服務會通知生產(chǎn)者和消費者有關Kafka系統(tǒng)中是否存在新代理或代理失敗的信息。 從Zookeeper收到的有關經(jīng)紀人存在或失敗的通知中，生產(chǎn)者和消費者做出決定并開始與其他經(jīng)紀人協(xié)調(diào)工作。 同樣，它負責為分區(qū)選擇新的領導者。

案例分析

稍作調(diào)整后，專注于練習。 因此，我們的案例研究使用Apache Kafka 2.3.1作為消息系統(tǒng)，在發(fā)布-訂閱上下文中模擬了使用Spring Boot微框架v2.1.8.RELEASE構建的兩個微服務之間的通信。 為了驗證我們的研究，我們將設置并執(zhí)行一個集成測試，該測試重點在于使用JUnit 4/5測試框架在端到端場景中集成應用程序的不同層。

生產(chǎn)者API是一個模塊，用于實現(xiàn)業(yè)務實體服務的操作，以協(xié)調(diào)和協(xié)調(diào)與企業(yè)，機構和實體組有關的經(jīng)濟信息。 消費者API是同一解決方案中的另一個模塊，旨在集中所有業(yè)務實體統(tǒng)計信息，并接收來自不同來源的數(shù)據(jù)輸入。

為了簡單起見，API使用H2內(nèi)存數(shù)據(jù)庫。 項目結構由三個模塊組成。 Producer和Consumer這兩個主要模塊都具有Common模塊的依賴性，在Common模塊中，它與系統(tǒng)的其余部分共享諸如錯誤處理和輔助類之類的內(nèi)容。

可從GitHub存儲庫訪問該示例； 要下載它，請點擊此鏈接 。

讓我們開始吧。

將Spring Kafka與Apache Kafka消息系統(tǒng)集成

用于Apache Kafka的Spring項目將核心Spring概念應用于基于Kafka的消息傳遞解決方案的開發(fā)。 它提供了一個“模板”作為發(fā)送消息的高級抽象。 它還通過@KafkaListener批注和“偵聽器容器”為消息驅(qū)動的POJO提供支持。 這些庫促進了依賴注入和聲明性[3]的使用。

生產(chǎn)者API

我們需要兩個步驟來配置生產(chǎn)者。 第一個是config類，其中定義生產(chǎn)者Map對象，生產(chǎn)者工廠和Kafka模板。 當我們將消息生成器設置為在Kafka代理中發(fā)布時，第二種尊重服務類。

生產(chǎn)者配置

在配置類中，在application.properties中設置了常數(shù)“ bootstrapServers” （即Kafka服務器）。 使用@Value（“ $ {spring.kafka.bootstrap-servers}”）批注表示受影響參數(shù)的默認值表達式。

要創(chuàng)建Kafka生產(chǎn)者，我們定義一些屬性，這些屬性將傳遞給Kafka生產(chǎn)者的構造函數(shù)。 在“ producerconfigs ” @Bean中，我們將BOOTSTRAP_SERVERS_CONFIG屬性設置為我們先前在application.properties中定義的代理地址列表。 BOOTSTRAP_SERVERS_CONFIG值是主機/端口對的逗號分隔列表，生產(chǎn)者用來建立與Kafka群集的初始連接。

package com.BusinessEntityManagementSystem;import ...@Configuration public class KafkaProducerConfig {@Value("${spring.kafka.bootstrap-servers}")private String bootstrapServers;@Beanpublic Map<String, Object> producerConfigs() {Map<String, Object> props = new HashMap<>();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);return props;}@Beanpublic ProducerFactory<String, BusinessEntity> producerFactory() {return new DefaultKafkaProducerFactory<>(producerConfigs());}@Beanpublic KafkaTemplate<String, BusinessEntity> kafkaTemplate() {return new KafkaTemplate<String, BusinessEntity>(producerFactory());} }

KEY_SERIALIZER_CLASS_CONFIG是用于Kafka記錄鍵的Kafka序列化器類，該類實現(xiàn)了Kafka序列化器接口。 注意，我們將其設置為StringSerializer.class作為消息ID。 VALUE_SERIALIZER_CLASS_CONFIG是一個Kafka序列化程序類，我們將其設置為JsonSerializer.class作為消息主體。

要創(chuàng)建消息，首先，我們需要配置一個ProducerFactory，該工廠設置創(chuàng)建Kafka Producer實例的策略。 然后，我們需要一個KafkaTemplate，它包裝一個Producer實例，并提供使用數(shù)據(jù)傳輸對象“ BusinessEntity ”將消息發(fā)送到Kafka主題的便捷方法。

生產(chǎn)者服務

在Kafka Producer服務類中， @ Service注釋表示帶注釋的類是“服務”。 在此類中，我們實現(xiàn)了將消息發(fā)送到Kafka代理的方法，在application.properties中預定義的標頭上聲明了topic屬性。

package com.BusinessEntityManagementSystem.kafka;import ...@Service public class KafkaProducer {@Autowiredprivate KafkaTemplate<String, BusinessEntity> kafkaTemplate;@Value("${statistics.kafka.topic}")String kafkaTopic;public void send(BusinessEntity payload) {Message<BusinessEntity> message = MessageBuilder.withPayload(payload).setHeader(KafkaHeaders.TOPIC, kafkaTopic).build();kafkaTemplate.send(message);} }

消費者API

在Consumer中，我們需要添加適當?shù)腄eserializer，該Deserializer可以將JSON byte []轉換為Java Object。 要設置它，我們需要使用config和帶有@components注釋的類，當使用基于注釋的配置和類路徑掃描時，它們將自動檢測此類以進行依賴項注入。

消費者配置

同樣，當我們指定KEY_SERIALIZER_CLASS_CONFIG，VALUE_SERIALIZER_CLASS_CONFIG來序列化生產(chǎn)者發(fā)布的消息時，我們還需要通知Spring Kafka有關反序列化的常量值，例如KEY_DESERIALIZER_CLASS_CONFIG和VALUE_DESERIALIZER_CLASS_CONFIG。 除了上面引用的常量之外，我們還指定了GROUP_ID_CONFIG和AUTO_OFFSET_RESET_CONFIG作為最早的常量，從而允許使用者讀取代理中最后插入的消息。

要啟用Kafka偵聽器，我們使用@EnableKafka批注。 這注釋了由AbstractListenerContainerFactory在后臺創(chuàng)建的端點。 KafkaListenerContainerFactory負責為特定端點創(chuàng)建偵聽器容器。 它可以檢測容器中任何受Spring管理的bean上的KafkaListener批注。

作為典型的實現(xiàn)， ConcurrentKafkaListenerContainerFactory提供了基礎MessageListenerContainer支持的必要配置選項。

package com.BusinessStatisticsUnitFiles;import ...@Configuration @EnableKafka public class KafkaConsumerConfig {@Value("${spring.kafka.bootstrap-servers}")private String bootstrapServers;@Beanpublic Map<String, Object> consumerConfigs() {Map<String, Object> props = new HashMap<>();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);props.put(ConsumerConfig.GROUP_ID_CONFIG, "statistics-BusinessStatisticsUnitFiles-group");props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");return props;}@Beanpublic ConsumerFactory<String, BusinessEntity> consumerFactory() {return new DefaultKafkaConsumerFactory<>(consumerConfigs(),new StringDeserializer(),new JsonDeserializer<>(BusinessEntity.class, false));}@Beanpublic ConcurrentKafkaListenerContainerFactory<String, BusinessEntity> kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactory<String, BusinessEntity> factory =new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory());return factory;} }

在消費者工廠，我們可以禁用標題的使用。 現(xiàn)在，這可以通過將新JsonDeserializer <>（BusinessEntity.class，false））中的第二個參數(shù)設置為false來實現(xiàn)； 。 這使使用者可以信任來自任何程序包的消息。

消費者“服務”

為了使用消息，有必要像上面一樣配置ConsumerFactory和KafkaListenerContainerFactory。 一旦這些bean在Spring bean工廠中可用，就可以使用@KafkaListener注釋配置基于POJO的使用者。

在用@KafkaListener注釋的類中， @ KafkaHandler還需要將方法標記為Kafka消息偵聽器的目標。 重要的是要了解，當消息到達時，選擇的方法取決于有效負載類型。 該類型與單個非注釋參數(shù)匹配，或與@Payload注釋的參數(shù)匹配。 絕不能有歧義-系統(tǒng)必須能夠根據(jù)有效載荷類型選擇恰好一種方法。

package com.BusinessStatisticsUnitFiles.kafka;import ...@Component public class KafkaConsumer {@AutowiredIBusinessEntityRepository businessEntityRepository;private static final Logger LOG = LoggerFactory.getLogger(BusinessEntity.class);@KafkaListener(topics = "${statistics.kafka.topic.create.entity}", groupId = "statistics-BusinessEntityManagementSystem-group")@KafkaHandlerpublic void receiveCreatedEntity(@Payload BusinessEntity data,@Headers MessageHeaders headers) {businessEntityRepository.save(RetrieveConsumerFromReceivedProducerObject.Binding(new BusinessEntityModel(), data));} }

@Payload批注將方法參數(shù)綁定到消息的有效負載。 它還可以用于將有效負載與方法調(diào)用相關聯(lián)。 有效負載可以通過MessageConverter傳遞，以將其從具有特定MIME類型的序列化形式轉換為與目標方法參數(shù)匹配的Object。 用@Payload注釋的類是“ BusinessEntity” DTO。

Spring Boot還支持使用偵聽器中的@Headers批注檢索一個或多個消息頭。 可以為一個主題實現(xiàn)多個偵聽器，每個偵聽器具有不同的組ID。 此外，一個消費者可以收聽來自各種主題的消息。

您可能已經(jīng)注意到，我們創(chuàng)建的主題建筑只有一個分區(qū)。 但是，對于具有多個分區(qū)的主題， @ KafkaListener可以顯式訂閱具有初始偏移量的主題的特定分區(qū)。

Application.properties

最后但并非最不重要的一點是，在我們的配置中，我們指定一些與生產(chǎn)者和消費者之間的通信行為有關的值。

生產(chǎn)者/消費者

在每個Producer和Consumer API上，我們都使用spring.kafka.bootstrap-servers = localhost：9092定義了我們希望微服務連接的Kafka集群。 同樣，有必要定義主題名稱以產(chǎn)生和接收消息，密鑰以及組ID。

... ## Application.properties Kafka config spring.kafka.bootstrap-servers=localhost:9092 statistics.kafka.topic=test statistics.kafka.key=test statistics.kafka.topic.create.entity=test spring.kafka.producer.group-id=statistics-BusinessStatisticsUnitFiles-group spring.kafka.template.default-topic=test ...

準備Kafka和Zookeeper進行集成測試

下面定義的步驟演示了如何在Windows 10操作系統(tǒng)上運行和測試Kafka。

下載帶有嵌入式Zookeeper的Kafka

下載Kafka二進制文件 。 這篇文章基于Kafka 2.3.1，因此我們假設您正在下載Scala 2.12的2.3.1版本。

解壓縮kafka_2.12-2.3.1.tgz文件。

設置zookeeper.properties

為了使其正常工作，我們需要更改Zookeeper數(shù)據(jù)目錄的位置。
打開kafka \ config \ zookeeper.properties文件，然后將Zookeeper數(shù)據(jù)/ log目錄位置配置更改為有效的Windows目錄位置。

dataDir=C:\\kafka\\zookeeper-logs

設置server.properties

我們還需要對Kafka配置進行一些更改。 打開kafka \ config \ server.properties并將主題默認設置為1。 我們將運行一個單節(jié)點Kafka。 另外，為防止Kafka創(chuàng)建不必要的偏移量，我們將副本指定為1。我們在Windows環(huán)境中使用最新的Kafka 2.3.1版本遇到此問題。 這導致Kafka停止，因為內(nèi)存不足，無法處理在啟動服務器的初始階段自動創(chuàng)建的大量數(shù)據(jù)。

############################# Log Basics #############################log.dirs=C:\\kafka\\kafka-logs####################### Internal Topic Settings #####################offsets.topic.replication.factor=1 offsets.topic.num.partitions = 1 min.insync.replicas=1 default.replication.factor = 1 ...

要完成Kafka配置，請將Kafka bin \ windows目錄添加到PATH環(huán)境變量中。

創(chuàng)建和執(zhí)行集成測試

顧名思義，集成測試專注于集成應用程序的不同層，其中不涉及任何模擬。 集成測試需要啟動一個容器來執(zhí)行測試用例。 因此，為此需要一些額外的設置，但是使用spring boot時，使用一些注釋和庫可以很容易地完成這些步驟。

測試班

第一個注釋@RunWith（SpringRunner.class）用于在Spring Boot測試功能和JUnit之間建立橋梁。 SpringRunner.class在測試中全面支持Spring上下文加載和Bean的依賴項注入。 @SpringBootTest通過SpringApplication創(chuàng)建ApplicationContext測試，這些測試將在我們的測試中使用。 自嵌入式服務器以來，它將引導整個容器，并創(chuàng)建一個Web環(huán)境。

在我們的測試中，我們模仿的是真實的Web環(huán)境，將其設置為RANDOM_PORT也會加載WebServerApplicationContext。 嵌入式服務器將啟動并在隨機端口上進行偵聽。

@RunWith(SpringRunner.class) @SpringBootTest(classes = {BusinessEntityManagementApplication.class}, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) class BusinessEntityIntegrationTest {@LocalServerPortprivate int port;@AutowiredTestRestTemplate restTemplate;HttpHeaders headers = new HttpHeaders();

@LocalServerPort批注為我們提供了在運行時分配的注入的HTTP端口。 這是@Value("${local.server.port}")的便捷替代方法。

要訪問Spring應用程序中的第三方REST服務，我們使用Spring RestTemplate或TestRestTemplate ，這是一種適合集成測試的便捷替代方法，方法是將其注入我們的測試類中。 通過在項目中使用spring-boot-starter-test依賴項，我們可以在運行時訪問“ TestRestTemplate”類。

測試方法

在我們的方法測試中，我們使用“ junit-json-params ”，這是一個Junit 5庫，提供注釋以從JSON字符串或參數(shù)化測試中的文件加載數(shù)據(jù)。 我們還使用@ParameterizedTest注釋對方法進行了注釋，以補充下面的庫。 它用于表示帶注釋的方法是參數(shù)化測試方法。 該方法不得為私有或靜態(tài)。 他們還必須通過@ArgumentsSource或相應的組合批注指定至少一個ArgumentsProvider 。

我們的@ArgumentsSource是@ArgumentsSource中的JSON文件@JsonFileSource（resources =“ /business-entity-test-param.json”）。 @JsonFileSource允許您使用類路徑中的JSON文件。 它支持單個對象，對象數(shù)組和JSON原語。

從文件中檢索的JSON對象綁定到方法參數(shù)“對象”，該方法將其轉換為POJO對象，在本例中為我們的實體模型。

@ParameterizedTest @JsonFileSource(resources = "/business-entity-test-param.json") @DisplayName("create business entity with json parameter") void createBusinessEntity(JsonObject object) throws IOException, URISyntaxException {BusinessEntityModel businessEntityModel;businessEntityModel = new BusinessEntityModel();ObjectMapper mapper = new ObjectMapper();businessEntityModel = mapper.readValue(object.toString(), BusinessEntityModel.class);HttpEntity<BusinessEntityModel> request = new HttpEntity<>(businessEntityModel, headers);try {ResponseEntity<String> response = this.restTemplate.postForEntity(createURLWithPort("/api/businessEntityManagementSystem/v1/businessEntity"), request, String.class);assertAll(() -> assertThat(response.getStatusCodeValue()).isEqualTo(HttpStatus.CREATED.value()),() -> assertThat(response.getHeaders().getLocation().getPath()).contains("/v1"));}catch(HttpClientErrorException ex) {assertAll(() -> Assert.assertEquals(HttpStatus.BAD_REQUEST.value(), ex.getRawStatusCode()),() -> Assert.assertEquals(true, ex.getResponseBodyAsString().contains("Missing request header")));} }

在安排并執(zhí)行操作之后，我們斷言對其余API的調(diào)用是否返回了所需的結果。

運行集成測試

在我們的開發(fā)環(huán)境中，我們需要確保我們的Kafka和Zookeeper在兩個不同的控制臺中啟動并運行，如圖所示

Kafka需要Zookeeper，因此我們將首先使用以下命令啟動Zookeeper。

c:\kafka>.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

它應該啟動Zookeeper服務器。 最小化命令窗口，并讓Zookeeper在該窗口中運行。 啟動一個新的命令窗口，并使用以下命令啟動Kafka Broker。

c:\kafka>.\bin\windows\kafka-server-start.bat .\config\server.properties

接下來，我們將按照我們的想法運行Consumer API，或者將其部署在任何兼容的Web服務器中。

最后，我們可以將測試類作為JUnit測試執(zhí)行。 它將按正常方式啟動服務器并部署API。 然后它將執(zhí)行測試。 您可以在“ JUnit”選項卡中驗證測試。

結論

在本文中，我們看到了如何使用發(fā)布-訂閱模式以可響應的方式在兩個不同的微服務之間使用可定制格式頻繁，立即，可靠和異步地共享數(shù)據(jù)，并通過在不同層中的不同層進行集成測試來對其進行驗證。端到端場景。

參考文獻

[1] Kafka 2.3文檔 ；

[2] Gregor Hohpe，Bobby Woolf，企業(yè)集成模式設計，構建和部署消息解決方案，2003年；

[3] 適用于Apache Kafka 2.3.3的Spring 。

翻譯自: https://www.javacodegeeks.com/2019/11/microservices-in-publish-subscribe-communication-using-apache-kafka-as-a-messaging-systems-and-validated-through-integration-test.html

總結

以上是生活随笔為你收集整理的使用Apache Kafka作为消息系统的发布-订阅通信中的微服务，并通过集成测试进行了验证...的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。