當(dāng)前位置：首頁(yè) > 编程语言 > java >内容正文

java

java程序连接kafka_Java的Kafka：构建安全，可扩展的消息传递应用程序

發(fā)布時(shí)間：2023/12/3 java 25 豆豆

生活随笔收集整理的這篇文章主要介紹了 java程序连接kafka_Java的Kafka：构建安全，可扩展的消息传递应用程序小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

java程序連接kafka

使用Okta的身份管理平臺(tái)輕松部署您的應(yīng)用程序 使用Okta的API在幾分鐘之內(nèi)即可對(duì)任何應(yīng)用程序中的用戶進(jìn)行身份驗(yàn)證，管理和保護(hù)。今天嘗試Okta。

當(dāng)今的用戶希望可以通過(guò)其計(jì)算機(jī)，手機(jī)，平板電腦或任何其他設(shè)備訪問(wèn)您的應(yīng)用程序！這種向軟件即服務(wù)（SaaS）規(guī)范的過(guò)渡要求開(kāi)發(fā)人員有效地與強(qiáng)大的工具集成，這些工具可以擴(kuò)展為每秒處理數(shù)千（甚至數(shù)百萬(wàn)）個(gè)請(qǐng)求。 Apache Kafka是處理那些高吞吐量環(huán)境的最有效工具之一。

在本教程中，您將學(xué)習(xí)Apache Kafka的基本概念，并構(gòu)建一個(gè)功能齊全的Java應(yīng)用程序，該應(yīng)用程序能夠產(chǎn)生和使用來(lái)自Kafka的消息。

先決條件： Java 8+，互聯(lián)網(wǎng)連接和免費(fèi)的Okta開(kāi)發(fā)人員帳戶。

Apache Kafka的簡(jiǎn)要概述

Apache Kafka是一個(gè)分布式流媒體平臺(tái)，它利用發(fā)布/訂閱消息模式與應(yīng)用程序進(jìn)行交互，并且旨在創(chuàng)建持久消息。

讓我們更詳細(xì)地分解這些概念。

分布式流媒體平臺(tái)

當(dāng)您要運(yùn)行Kafka時(shí)，您需要啟動(dòng)其代理：與其他任何服務(wù)器一樣，在計(jì)算機(jī)上運(yùn)行的Kafka的簡(jiǎn)單實(shí)例。代理負(fù)責(zé)將消息發(fā)送，接收和存儲(chǔ)到磁盤中。

一個(gè)經(jīng)紀(jì)人不足以確保Kafka可以處理高吞吐量的消息。該目標(biāo)是通過(guò)許多經(jīng)紀(jì)人同時(shí)合作，相互溝通和協(xié)調(diào)來(lái)實(shí)現(xiàn)的。

Kafka集群將一個(gè)或多個(gè)經(jīng)紀(jì)人組合在一起。您的應(yīng)用程序連接到一個(gè)群集，該群集為您管理所有分布式詳細(xì)信息，而不是連接到單個(gè)節(jié)點(diǎn)。

具有持久消息的發(fā)布/訂閱消息系統(tǒng)

發(fā)布/訂閱是分布式系統(tǒng)中的常見(jiàn)模式。下圖說(shuō)明了Kafka中此模式的基本結(jié)構(gòu)：

該圖像包括到目前為止尚未提及的兩個(gè)組件：生產(chǎn)者和消費(fèi)者。

生產(chǎn)者是將消息發(fā)送到群集的應(yīng)用程序。在此示例中，生產(chǎn)者1、2和3正在發(fā)送消息。然后，集群選擇應(yīng)由哪個(gè)代理存儲(chǔ)它們，并將其發(fā)送給選定的代理。

另一方面，您有消費(fèi)者。使用者是連接到集群并接收生產(chǎn)者發(fā)布的消息的應(yīng)用程序。任何有興趣使用生產(chǎn)者發(fā)送的消息的應(yīng)用程序都必須連接到Kafka消費(fèi)者。

由于Kafka會(huì)長(zhǎng)時(shí)間存儲(chǔ)消息（默認(rèn)值為7天），因此即使發(fā)送消息時(shí)不在場(chǎng)，您也可以讓許多使用者收到相同的消息！

卡夫卡主題

將消息發(fā)送到Kafka代理時(shí)，需要通過(guò)指定主題來(lái)指定將消息發(fā)送到的位置。主題是消費(fèi)者可以訂閱的消息類別。該機(jī)制確保使用者僅接收與其相關(guān)的消息，而不是接收發(fā)布到集群的每條消息。

現(xiàn)在您已經(jīng)了解了Kafka的基本架構(gòu)，讓我們下載并安裝它。

安裝并運(yùn)行Kafka

要下載Kafka，請(qǐng)?jiān)L問(wèn)Kafka網(wǎng)站。將此壓縮文件的內(nèi)容提取到您喜歡的文件夾中。

在Kafka目錄中，轉(zhuǎn)到bin文件夾。在這里，您會(huì)發(fā)現(xiàn)許多bash腳本，這些腳本對(duì)于運(yùn)行Kafka應(yīng)用程序很有用。如果使用的是Windows，則windows文件夾中也有相同的腳本。本教程使用Linux命令，但是如果您正在運(yùn)行Microsoft OS，則只需使用等效的Windows版本。

啟動(dòng)Zookeeper管理您的Kafka群集

Apache Kafka始終作為分布式應(yīng)用程序運(yùn)行。這意味著您的集群必須在同步配置或選舉負(fù)責(zé)人的過(guò)程中處理一些分布式挑戰(zhàn)。

Kafka使用Zookeeper跟蹤這些細(xì)節(jié)。不過(guò)，不必?fù)?dān)心下載它。 Kafka已經(jīng)與Zookeeper一起提供，可以讓您快速啟動(dòng)并運(yùn)行。

讓我們啟動(dòng)一個(gè)Zookeeper實(shí)例！在您的Kafka目錄中的bin文件夾中，運(yùn)行以下命令：

./zookeeper-server-start.sh ../config/zookeeper.properties

默認(rèn)情況下，此命令在端口2181上啟動(dòng)Zookeeper服務(wù)器。 Zookeeper負(fù)責(zé)協(xié)調(diào)集群內(nèi)的Kafka經(jīng)紀(jì)人。您將在本教程的Kafka項(xiàng)目中使用默認(rèn)配置，但始終可以根據(jù)需要更改這些值。

運(yùn)行卡夫卡經(jīng)紀(jì)人

下一步是運(yùn)行代理本身。在另一個(gè)終端上，從bin文件夾運(yùn)行以下命令：

./kafka-server-start.sh ../config/server.properties

您可能已經(jīng)猜到了，此命令在默認(rèn)端口9092上以默認(rèn)配置運(yùn)行Kafka服務(wù)器。

創(chuàng)建一個(gè)Kafka主題

現(xiàn)在您已運(yùn)行代理和Zookeeper，現(xiàn)在可以指定一個(gè)主題以開(kāi)始從生產(chǎn)者發(fā)送消息。您將在bin文件夾中運(yùn)行命令，就像前面的步驟一樣：

./kafka-topics.sh --create --topic myTopic -zookeeper \localhost:2181 --replication-factor 1 --partitions 1

此命令創(chuàng)建一個(gè)名為myTopic ，該myTopic指向您使用第一個(gè)命令啟動(dòng)的Zookeeper實(shí)例。您還必須指定兩個(gè)不同的參數(shù)： replication-factor和partitions 。現(xiàn)在不用擔(dān)心它們–它們用于控制與Kafka中的分布式系統(tǒng)相關(guān)的特定方面。在運(yùn)行簡(jiǎn)單設(shè)置時(shí)，您可以為兩個(gè)參數(shù)指定“ 1”。

現(xiàn)在您已完成所有事情并可以運(yùn)行，您可以開(kāi)始將Kafka與Java應(yīng)用程序集成！

創(chuàng)建一個(gè)Java + Kafka應(yīng)用程序

讓我們從項(xiàng)目結(jié)構(gòu)開(kāi)始，使用Spring Initializer創(chuàng)建應(yīng)用程序。

轉(zhuǎn)到https://start.spring.io并填寫以下信息：

項(xiàng)目：Maven項(xiàng)目
語(yǔ)言：Java
群組： com.okta.javakafka
工件： kafka-java
依存關(guān)系：
- Spring網(wǎng)

您也可以使用命令行生成項(xiàng)目。將以下命令粘貼到您的終端中，它將下載與上面定義的配置相同的項(xiàng)目：

curl https://start.spring.io/starter.zip -d language=java \-d dependencies=web,kafka \-d packageName=com.okta.javakafka \-d name=kafka-java \-d type=maven-project \-o kafka-java.zip

本教程使用Maven，但是您可以根據(jù)需要輕松地使用Gradle進(jìn)行學(xué)習(xí)。

而已！現(xiàn)在，您的Java項(xiàng)目結(jié)構(gòu)已創(chuàng)建，您可以開(kāi)始開(kāi)發(fā)應(yīng)用程序了。

在Java應(yīng)用程序中將消息推送到Kafka主題

創(chuàng)建可以推送消息的生產(chǎn)者的第一步是在Java應(yīng)用程序中配置生產(chǎn)者。讓我們創(chuàng)建一個(gè)配置類來(lái)做到這一點(diǎn)。

創(chuàng)建一個(gè)src/main/java/com/okta/javakafka/configuration文件夾，并在其中創(chuàng)建一個(gè)ProducerConfiguration類：

import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.serialization.StringSerializer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory;import java.util.HashMap; import java.util.Map;@Configuration public class ProducerConfiguration {private static final String KAFKA_BROKER = "localhost:9092";@Beanpublic ProducerFactory<String, String> producerFactory() {return new DefaultKafkaProducerFactory<>(producerConfigurations());}@Beanpublic Map<String, Object> producerConfigurations() {Map<String, Object> configurations = new HashMap<>();configurations.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BROKER);configurations.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);configurations.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);return configurations;}@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {return new KafkaTemplate<>(producerFactory());}}

此類創(chuàng)建一個(gè)ProducerFactory ，該P(yáng)roducerFactory知道如何根據(jù)您提供的配置來(lái)創(chuàng)建生產(chǎn)者。您還指定了連接到本地Kafka代理，并使用String序列化密鑰和值。

您還聲明了一個(gè)KafkaTemplate bean在生產(chǎn)者上執(zhí)行高級(jí)操作。換句話說(shuō)，模板能夠執(zhí)行諸如將消息發(fā)送到主題之類的操作，并有效地向您隱藏了后臺(tái)信息。

下一步是創(chuàng)建端點(diǎn)，以將消息發(fā)送給生產(chǎn)者。在src/main/java/com/okta/javakafka/controller軟件包中，創(chuàng)建以下類：

import org.springframework.kafka.core.KafkaTemplate; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController;import java.util.List;@RestController public class KafkaController {private KafkaTemplate<String, String> template;public KafkaController(KafkaTemplate<String, String> template) {this.template = template;}@GetMapping("/kafka/produce")public void produce(@RequestParam String message) {template.send("myTopic", message);}}

注意：由于您要發(fā)送要處理的數(shù)據(jù)，所以produce()方法實(shí)際上應(yīng)該是POST。出于演示目的，將其保留為GET更容易，因此您可以在瀏覽器中進(jìn)行練習(xí)。

如您所見(jiàn)，此端點(diǎn)非常簡(jiǎn)單。它噴射KafkaTemplate前面配置和發(fā)送一個(gè)消息給myTopic當(dāng)GET請(qǐng)求到由/kafka/produce 。

讓我們測(cè)試一切是否按預(yù)期工作。在JavaKafkaApplication類中運(yùn)行main方法。要從命令行運(yùn)行，請(qǐng)執(zhí)行以下命令：

./mvnw spring-boot:run

您的服務(wù)器應(yīng)該在端口8080上運(yùn)行，并且您已經(jīng)可以對(duì)它發(fā)出API請(qǐng)求！

轉(zhuǎn)到網(wǎng)絡(luò)瀏覽器，然后訪問(wèn)http：// localhost：8080 / kafka / produce？message =這是我的消息。

當(dāng)您使用上述命令進(jìn)行調(diào)用時(shí)，您的應(yīng)用程序?qū)?zhí)行/kafka/produce端點(diǎn)，該端點(diǎn)將消息發(fā)送到Kafka中的myTopic主題。

但是您如何知道該命令已成功向該主題發(fā)送消息？現(xiàn)在，您不會(huì)在應(yīng)用程序內(nèi)使用消息，這意味著您不能確定！

幸運(yùn)的是，有一種簡(jiǎn)單的方法可以創(chuàng)建一個(gè)立即進(jìn)行測(cè)試的消費(fèi)者。在您的Kafka目錄的bin文件夾中，運(yùn)行以下命令：

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic myTopic

訪問(wèn)http：// localhost：8080 / kafka / produce？message =這是我的消息，再次在運(yùn)行Kafka使用者的終端中看到以下消息：

This is my message

很好！您可以暫時(shí)停止此命令。

讓我們添加一些Java代碼來(lái)使用應(yīng)用程序中的消息，而不是從終端執(zhí)行。

在Java App中使用來(lái)自Kafka主題的消息

與生產(chǎn)者一樣，您需要添加配置以使消費(fèi)者能夠找到Kafka經(jīng)紀(jì)人。

在src/main/java/com/okta/javakafka/configuration創(chuàng)建以下類：

import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.serialization.StringDeserializer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory;import java.util.HashMap; import java.util.Map;@Configuration public class ConsumerConfiguration {private static final String KAFKA_BROKER = "localhost:9092";private static final String GROUP_ID = "kafka-sandbox";@Beanpublic ConsumerFactory<String, String> consumerFactory() {return new DefaultKafkaConsumerFactory<>(consumerConfigurations());}@Beanpublic Map<String, Object> consumerConfigurations() {Map<String, Object> configurations = new HashMap<>();configurations.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BROKER);configurations.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);configurations.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);configurations.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);return configurations;}@BeanConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory());return factory;}}

上面的代碼創(chuàng)建了一個(gè)工廠，該工廠知道如何連接到本地代理。它還將您的使用者配置為針對(duì)鍵和值對(duì)String反序列化，以匹配生產(chǎn)者配置。

組ID是必需的，Kafka使用組ID來(lái)允許并行數(shù)據(jù)消耗。 ConcurrentKafkaListenerContainerFactory bean使您的??應(yīng)用程序可以在多個(gè)線程中使用消息。

現(xiàn)在，您的Java應(yīng)用已配置為在Kafka經(jīng)紀(jì)人中查找使用者，讓我們開(kāi)始收聽(tīng)發(fā)送給該主題的消息。

創(chuàng)建一個(gè)src/main/java/com/okta/javakafka/consumer目錄，并在其中創(chuàng)建以下類：

import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component;import java.util.ArrayList; import java.util.List;@Component public class MyTopicConsumer {private final List<String> messages = new ArrayList<>();@KafkaListener(topics = "myTopic", groupId = "kafka-sandbox")public void listen(String message) {synchronized (messages) {messages.add(message);}}public List<String> getMessages() {return messages;}}

此類負(fù)責(zé)偵聽(tīng)myTopic主題內(nèi)的更改。它通過(guò)使用KafkaListener注釋來(lái)實(shí)現(xiàn)。每當(dāng)生產(chǎn)者向主題發(fā)送新消息時(shí)，您的應(yīng)用程序都會(huì)在此類內(nèi)接收到一條消息。它將一條消息添加到接收到的消息列表中，通過(guò)getMessages()方法使其可用于其他類。

接下來(lái)，讓我們創(chuàng)建一個(gè)顯示已消費(fèi)消息列表的端點(diǎn)。返回KafkaController以添加MyTopicConsumer作為依賴項(xiàng)和getMessages()方法。

import com.okta.javakafka.consumer.MyTopicConsumer; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController;import java.util.List;@RestController public class KafkaController {private KafkaTemplate<String, String> template;private MyTopicConsumer myTopicConsumer;public KafkaController(KafkaTemplate<String, String> template, MyTopicConsumer myTopicConsumer) {this.template = template;this.myTopicConsumer = myTopicConsumer;}@GetMapping("/kafka/produce")public void produce(@RequestParam String message) {template.send("myTopic", message);}@GetMapping("/kafka/messages")public List<String> getMessages() {return myTopicConsumer.getMessages();}}

此類現(xiàn)在具有一個(gè)新的終結(jié)點(diǎn)，以顯示存儲(chǔ)在使用者中的消息。調(diào)用此端點(diǎn)時(shí)，它將發(fā)送已從Kafka主題處理過(guò)的當(dāng)前消息。

您的Java應(yīng)用程序現(xiàn)在同時(shí)具有Kafka生產(chǎn)者和使用者，因此讓我們一起進(jìn)行測(cè)試！重新啟動(dòng)您的應(yīng)用程序，然后轉(zhuǎn)到http：// localhost：8080 / kafka / messages 。

目前，沒(méi)有信息被返回。原因很簡(jiǎn)單：您的使用者僅配置為接收新消息，而您尚未發(fā)送新消息。讓我們通過(guò)訪問(wèn)網(wǎng)絡(luò)瀏覽器并訪問(wèn)http：// localhost：8080 / kafka / produce？message =我的應(yīng)用程序發(fā)送的消息來(lái)解決此問(wèn)題！。

當(dāng)Kafka收到該消息時(shí)，它將立即讓您的消費(fèi)者知道它。繼續(xù)并在瀏覽器中轉(zhuǎn)到http：// localhost：8080 / kafka / messages 。現(xiàn)在，您將看到您的消息已成功收到！

很好！您有一個(gè)Java應(yīng)用程序，能夠產(chǎn)生和使用來(lái)自Kafka的消息！但是，在我們將其稱為“一天”之前，還有最后一步，這是非常重要的一步。

保護(hù)您的Java Kafka應(yīng)用程序

您的應(yīng)用目前不是很安全。盡管您已經(jīng)準(zhǔn)備好在分布式環(huán)境中處理許多消息，但是那些可以找到指向您的端點(diǎn)的鏈接的人仍然可以使用這些消息。這是一個(gè)關(guān)鍵漏洞，因此請(qǐng)確保已正確解決此漏洞。

您將使用OAuth 2.0來(lái)確保只有經(jīng)過(guò)身份驗(yàn)證的用戶才能看到您的端點(diǎn)。最好的部分？使用Okta驗(yàn)證用戶身份，只需5分鐘即可在您的應(yīng)用中添加此功能！

創(chuàng)建一個(gè)Okta帳戶

如果您還沒(méi)有Okta帳戶，請(qǐng)繼續(xù)創(chuàng)建一個(gè) 。完成注冊(cè)后，請(qǐng)執(zhí)行以下步驟：

登錄到您的帳戶
轉(zhuǎn)到應(yīng)用程序 > 添加應(yīng)用程序 。您將被重定向到以下頁(yè)面：

選擇網(wǎng)站 ，然后單擊下一步。
在表格中填寫以下選項(xiàng)：
- 姓名： Bootiful Kafka
點(diǎn)擊完成

有了Okta應(yīng)用程序之后，就可以使用它來(lái)在Java + Kafka應(yīng)用程序中對(duì)用戶進(jìn)行身份驗(yàn)證。

使用用戶身份驗(yàn)證保護(hù)Java應(yīng)用安全

首先，將Okta的庫(kù)添加到您的項(xiàng)目中。打開(kāi)您的pom.xml并在<dependencies>標(biāo)記內(nèi)添加以下依賴項(xiàng)：

<dependency><groupId>com.okta.spring</groupId><artifactId>okta-spring-boot-starter</artifactId><version>1.3.0</version> </dependency>

該庫(kù)將與您剛創(chuàng)建的Okta應(yīng)用集成。它還會(huì)將Spring Security添加到您當(dāng)前的應(yīng)用程序中。在src/main/resources/application.properties使用以下變量配置它：

okta.oauth2.issuer: https://{yourOktaDomain}/oauth2/default okta.oauth2.client-id: {yourClientID} okta.oauth2.client-secret: {yourClientSecret}

重要說(shuō)明 ：此文件只能在本地使用。不要將客戶的機(jī)密泄露給Git或任何其他版本控制系統(tǒng)。

為避免意外暴露這些憑據(jù)，您還可以將Okta應(yīng)用程序的值指定為環(huán)境變量。使用以下環(huán)境變量在應(yīng)用程序的根目錄中創(chuàng)建okta.env文件。然后在啟動(dòng)您的應(yīng)用程序之前運(yùn)行source okta.env

export OKTA_OAUTH2_ISSUER=https://{yourOktaDomain}/oauth2/default export OKTA_OAUTH2_CLIENT_ID={yourClientID} export OKTA_OAUTH2_CLIENT_SECRET={yourClientSecret}

您可以在Okta UI的應(yīng)用程序頁(yè)面中找到{yourClientID}和{yourClientSecret} 。要訪問(wèn)它，請(qǐng)按照以下步驟操作：

在您的Okta菜單中，轉(zhuǎn)到“ 應(yīng)用程序”
選擇Bootiful Kafka應(yīng)用程序
單擊常規(guī)選項(xiàng)卡

您應(yīng)該在“客戶端憑據(jù)”區(qū)域中看到兩個(gè)值。

值{yourOktaDomain}將在Okta儀表板中可見(jiàn)，只需單擊菜單上的Dashboard。您將在右上角看到組織URL。

而已！

重新啟動(dòng)Spring Boot應(yīng)用程序，然后轉(zhuǎn)到http：// localhost：8080 / kafka / messages 。您的應(yīng)用程序現(xiàn)在會(huì)將您重定向到登錄頁(yè)面：

注意：如果未提示您登錄，那是因?yàn)槟呀?jīng)登錄。在隱身窗口中打開(kāi)應(yīng)用程序，您將看到上面顯示的登錄屏幕。

輸入您的用戶名和密碼。如果登錄嘗試成功，您將再次被重定向回您的應(yīng)用程序。

恭喜你！您現(xiàn)在有了一個(gè)安全的Java應(yīng)用程序，該應(yīng)用程序可以生成和使用來(lái)自Kafka的消息。

如果您想查看本教程的完整源代碼，請(qǐng)轉(zhuǎn)到 GitHub上的oktadeveloper / okta-java-kafka-example 。

想更多地了解Java，安全性和OAuth 2.0？以下是您可能感興趣的一些鏈接：

OAuth 2.0 Java指南：5分鐘保護(hù)您的應(yīng)用程序安全
OAuth和OpenID Connect圖解指南
使用Spring Boot和Kotlin構(gòu)建應(yīng)用程序
帶有Spring Boot和Spring Cloud的Java微服務(wù)
使用Spring Cloud Gateway保護(hù)React式微服務(wù)

有關(guān)此類文章的更多信息，請(qǐng)?jiān)赥witter上關(guān)注@oktadev 。我們還會(huì)定期將截屏視頻發(fā)布到我們的YouTube頻道！

翻譯自: https://www.javacodegeeks.com/2020/01/kafka-with-java-build-a-secure-scalable-messaging-app.html