?Apache Kafka是一種基于發(fā)布者-訂閱者模型的分布式消息傳遞系統(tǒng)。它由Apache軟件基金會(huì)開發(fā)，用Java和Scala編寫。Kafka的初衷是為了克服傳統(tǒng)消息傳遞系統(tǒng)的分布和可擴(kuò)展性面臨的問(wèn)題。它可以處理和存儲(chǔ)大量數(shù)據(jù)，具有延遲低、吞吐量高的優(yōu)點(diǎn)。因此，它適用于構(gòu)建實(shí)時(shí)數(shù)據(jù)處理應(yīng)用程序和流媒體服務(wù)。它目前已開源，被Netflix、沃爾瑪和領(lǐng)英等許多組織使用。

消息傳遞系統(tǒng)使多個(gè)應(yīng)用程序可以相互收發(fā)數(shù)據(jù)，不用擔(dān)心數(shù)據(jù)的傳輸和共享。點(diǎn)到點(diǎn)和發(fā)布者-訂閱者是兩種廣泛使用的消息傳遞系統(tǒng)。在點(diǎn)到點(diǎn)模型中，發(fā)送方將數(shù)據(jù)推送到隊(duì)列，接收方從隊(duì)列中彈出數(shù)據(jù)，就像遵循先進(jìn)先出（FIFO）原則的標(biāo)準(zhǔn)隊(duì)列系統(tǒng)一樣。此外，一旦讀取數(shù)據(jù)，數(shù)據(jù)就被刪除，并且每次只允許一個(gè)接收方。接收方讀取消息時(shí)不存在時(shí)間依賴關(guān)系。

圖1. 點(diǎn)對(duì)點(diǎn)消息系統(tǒng)

在發(fā)布者-訂閱者模型中，發(fā)送方稱為發(fā)布者，接收方稱為訂閱者。在這種模型中，多個(gè)發(fā)送方和接收方可以同時(shí)讀取或?qū)懭霐?shù)據(jù)。但是它有時(shí)間依賴關(guān)系。消費(fèi)者必須在一定的時(shí)間之前消費(fèi)消息，因?yàn)榇撕笙⒈粍h除，即使它沒(méi)有被讀取。這個(gè)時(shí)間限制可能是一天、一周或一個(gè)月，視用戶的配置而定。

圖2. 發(fā)布者-訂閱者消息系統(tǒng)

一、Kafka的架構(gòu)

Kafka架構(gòu)由幾個(gè)關(guān)鍵組件組成：

1. 主題

2. 分區(qū)

3. 代理

4. 生產(chǎn)者

5. 消費(fèi)者

6. Kafka集群

7. Zookeeper

圖3. Kafka的架構(gòu)

不妨簡(jiǎn)單了解一下每個(gè)組件。

Kafka將消息存儲(chǔ)在不同的主題中。主題是一個(gè)組，含有特定類別的消息。它類似數(shù)據(jù)庫(kù)中的表。主題由名稱作為唯一標(biāo)識(shí)符。不能創(chuàng)建名稱相同的兩個(gè)主題。

主題進(jìn)一步劃分為分區(qū)。這些分區(qū)的每個(gè)記錄都與一個(gè)名為Offset的唯一標(biāo)識(shí)符相關(guān)聯(lián)，該標(biāo)識(shí)符表示了記錄在該分區(qū)中的位置。

除此之外，系統(tǒng)中還有生產(chǎn)者和消費(fèi)者。生產(chǎn)者使用Producing API編寫或發(fā)布主題中的數(shù)據(jù)。這些生產(chǎn)者可以在主題或分區(qū)層面寫入數(shù)據(jù)。

消費(fèi)者使用Consumer API從主題中讀取或消費(fèi)數(shù)據(jù)。它們還可以在主題或分區(qū)層面讀取數(shù)據(jù)。執(zhí)行類似任務(wù)的消費(fèi)者將組成一個(gè)組，名為消費(fèi)者組。

還有其他系統(tǒng)，比如代理（Broker）和Zookeeper，它們?cè)贙afka服務(wù)器的后臺(tái)運(yùn)行。代理是維護(hù)和保存已發(fā)布消息記錄的軟件。它還負(fù)責(zé)使用offset以正確的順序?qū)⒄_的消息傳遞給正確的使用者。相互之間進(jìn)行集體通信的代理集可以稱為Kafka集群。代理可以動(dòng)態(tài)添加到Kafka集群中或從集群中動(dòng)態(tài)刪除，系統(tǒng)不會(huì)遇到任何停機(jī)。Kafka集群中的其中一個(gè)代理名為控制器。它負(fù)責(zé)管理集群內(nèi)的狀態(tài)和副本，并執(zhí)行管理任務(wù)。

另一方面，Zookeeper負(fù)責(zé)維護(hù)Kafka集群的健康狀態(tài)，并與該集群的每個(gè)代理進(jìn)行協(xié)調(diào)。它以鍵值對(duì)的形式維護(hù)每個(gè)集群的元數(shù)據(jù)。

本教程主要介紹實(shí)際實(shí)現(xiàn)Apache Kafka的例子。

二、出租車預(yù)訂應(yīng)用程序：實(shí)際用例

以優(yōu)步之類的出租車預(yù)訂服務(wù)這一用例為例。這個(gè)應(yīng)用程序使用Apache Kafka通過(guò)各種服務(wù)（比如事務(wù)、電子郵件、分析等）發(fā)送和接收消息。

圖4出租車應(yīng)用程序架構(gòu)圖

架構(gòu)由幾個(gè)服務(wù)組成。Rides服務(wù)接收來(lái)自客戶的打車請(qǐng)求，并將打車詳細(xì)信息寫入到Kafka消息系統(tǒng)上。

然后，Transaction服務(wù)讀取這些訂單詳細(xì)信息，確認(rèn)訂單和支付狀態(tài)。在確認(rèn)這趟打車之后，該Transaction服務(wù)將再次在消息系統(tǒng)中寫入確認(rèn)的打車信息，并添加一些額外的詳細(xì)信息。最后，電子郵件或數(shù)據(jù)分析等其他服務(wù)讀取已確認(rèn)的打車細(xì)節(jié)，并向客戶發(fā)送確認(rèn)郵件，并對(duì)其進(jìn)行一些分析。

我們可以以非常高的吞吐量和極小的延遲實(shí)時(shí)執(zhí)行所有這些進(jìn)程。此外，由于Apache Kafka能夠橫向擴(kuò)展，我們可以擴(kuò)展這個(gè)應(yīng)用程序以處理數(shù)百萬(wàn)用戶。

三、上述用例的實(shí)際實(shí)現(xiàn)

本節(jié)包含在我們的應(yīng)用程序中實(shí)現(xiàn)Kafka消息系統(tǒng)的快速教程。它包括下載和配置Kafka、創(chuàng)建生產(chǎn)者-消費(fèi)者函數(shù)的步驟。

注意：本教程基于Python編程語(yǔ)言，使用Windows機(jī)器。

1.Apache Kafka下載步驟

1）從這個(gè)鏈接（https://kafka.apache.org/downloads）下載最新版本的Apache Kafka。Kafka基于JVM語(yǔ)言，所以必須在系統(tǒng)中安裝Java 7或更高版本。

2） 從計(jì)算機(jī)的C:驅(qū)動(dòng)器解壓已下載的zip文件，并將文件夾重命名為/apache-kafka。

3）父目錄包含兩個(gè)子目錄：/bin和/config，分別含有Zookeeper和Kafka服務(wù)器的可執(zhí)行文件和配置文件。

2.配置步驟

首先，我們需要為Kafka和Zookeeper服務(wù)器創(chuàng)建日志目錄。這些目錄將存儲(chǔ)這些集群的所有元數(shù)據(jù)以及主題和分區(qū)的消息。

注意：默認(rèn)情況下，這些日志目錄創(chuàng)建在/tmp目錄中，這是一個(gè)易變目錄：當(dāng)系統(tǒng)關(guān)閉或重啟時(shí)，該目錄中的所有數(shù)據(jù)都會(huì)消失。我們需要為日志目錄設(shè)置永久路徑來(lái)解決這個(gè)問(wèn)題。不妨看看怎么做。

導(dǎo)航到apache-kafka >> config，打開server.properties文件。在這里您可以配置Kafka的許多屬性，比如日志目錄路徑、日志保留時(shí)間和分區(qū)數(shù)量等。

在server.properties文件中，我們必須將日志目錄文件的路徑從臨時(shí)/tmp目錄改為永久目錄。日志目錄含有Kafka Server中的生成或?qū)懭氲臄?shù)據(jù)。若要更改路徑，將log.dirs變量由/tmp/kafka-logs改為c:/apache-kafka/kafka-logs。這將使您的日志永久存儲(chǔ)。

log.dirs = c: / apache-kafka / kafka-logs

Zookeeper服務(wù)器還包含一些日志文件，用于存儲(chǔ)Kafka服務(wù)器的元數(shù)據(jù)。若要更改路徑，重復(fù)上面的步驟，即打開zookeeper.properties文件，并按如下方式替換路徑。

dataDir = c: / apache-kafka / zookeeper-logs

該Zookeeper服務(wù)器將充當(dāng)Kafka服務(wù)器的資源管理器。

四、運(yùn)行Kafka和Zookeeper服務(wù)器

若要運(yùn)行Zookeeper服務(wù)器，在父目錄中打開一個(gè)新的cmd提示符，并運(yùn)行以下命令。

$ .\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

圖5

保持Zookeeper實(shí)例運(yùn)行。

若要運(yùn)行Kafka服務(wù)器，打開一個(gè)單獨(dú)的cmd提示符，并執(zhí)行以下代碼：

$ .\bin\windows\kafka-server-start.bat .\config\server.properties

保持Kafka和Zookeeper服務(wù)器運(yùn)行；在下一節(jié)中，我們將創(chuàng)建生產(chǎn)者和消費(fèi)者函數(shù)，它們用于讀取數(shù)據(jù)并將數(shù)據(jù)寫入到Kafka服務(wù)器。

五、創(chuàng)建生產(chǎn)者和消費(fèi)者函數(shù)

為了創(chuàng)建生產(chǎn)者和消費(fèi)者函數(shù)，我們將以前面討論的電子商務(wù)應(yīng)用程序?yàn)槔?。“訂單”服?wù)將充當(dāng)生產(chǎn)者，將訂單細(xì)節(jié)寫入到Kafka服務(wù)器，而電子郵件和分析服務(wù)將充當(dāng)消費(fèi)者，從服務(wù)器讀取該數(shù)據(jù)。交易服務(wù)將充當(dāng)消費(fèi)者和生產(chǎn)者。它讀取訂單詳細(xì)信息，并在交易確認(rèn)后再次將它們寫回來(lái)。

但首先我們需要安裝Kafka Python庫(kù)，該庫(kù)含有生產(chǎn)者和消費(fèi)者的內(nèi)置函數(shù)。

$ pip install kafka-python

現(xiàn)在，創(chuàng)建一個(gè)名為kafka-tutorial的新目錄。我們將在該目錄中創(chuàng)建含有所需函數(shù)的Python文件。

$ mkdir kafka-tutorial
$ CD .\kafka-tutorial\

生產(chǎn)者函數(shù)：

現(xiàn)在，創(chuàng)建一個(gè)名為' rides.py '的Python文件，并將以下代碼粘貼到其中。

rides.py

import kafka
import json
import time
import random

topicName = "ride_details"
producer = kafka.KafkaProducer(bootstrap_servers="localhost:9092")

for i in range(1, 10):
ride = {
"id": i,
"customer_id": f"user_{i}",
"location": f"Lat: {random.randint(-90, 90)}, Long: {random.randint(-90, 90)}",
}
producer.send(topicName, json.dumps(ride).encode("utf-8"))
print(f"Ride Details Send Succesfully!")
time.sleep(5)

解釋：

首先，我們導(dǎo)入了所有必要的庫(kù)，包括Kafka。然后，定義主題名稱和各項(xiàng)目的列表。記住，主題是一個(gè)含有類似類型消息的組。在本例中，該主題將包含所有訂單。

然后，我們創(chuàng)建一個(gè)KafkaProducer函數(shù)的實(shí)例，并將其連接到在localhost:9092上運(yùn)行的Kafka服務(wù)器。如果您的Kafka服務(wù)器在不同的地址和端口上運(yùn)行，那么您必須在那里提及服務(wù)器的IP和端口號(hào)。

之后，我們將生成一些JSON格式的訂單，并根據(jù)定義的話題名稱將它們寫入到Kafka服務(wù)器。睡眠函數(shù)用于生成后續(xù)訂單之間的間隔。

消費(fèi)者函數(shù)：

transaction.py

import json
import kafka
import random

RIDE_DETAILS_KAFKA_TOPIC = "ride_details"
RIDES_CONFIRMED_KAFKA_TOPIC = "ride_confirmed"

consumer = kafka.KafkaConsumer(
RIDE_DETAILS_KAFKA_TOPIC, bootstrap_servers="localhost:9092"
)
producer = kafka.KafkaProducer(bootstrap_servers="localhost:9092")

print("Listening Ride Details")
while True:
for data in consumer:
print("Loading Transaction..")
message = json.loads(data.value.decode())
customer_id = message["customer_id"]
location = message["location"]
confirmed_ride = {
"customer_id": customer_id,
"customer_email": f"{customer_id}@xyz.com",
"location": location,
"alloted_driver": f"driver_{customer_id}",
"pickup_time": f"{random.randint(1, 20)}mins",
}
print(f"Transaction Completed..({customer_id})")
producer.send(
RIDES_CONFIRMED_KAFKA_TOPIC, json.dumps(confirmed_ride).encode("utf-8")
)

解釋：

transaction.py文件用于確認(rèn)用戶所做的交易，并為他們分配司機(jī)和估計(jì)的載客時(shí)間。它從Kafka服務(wù)器讀取打車細(xì)節(jié)，并在確認(rèn)打車后將其再次寫入到Kafka服務(wù)器。

現(xiàn)在，創(chuàng)建兩個(gè)名為email.py和analysis .py的Python文件，分別用于向客戶發(fā)送電子郵件以確認(rèn)打車和執(zhí)行一些分析。創(chuàng)建這些文件只是為了表明甚至多個(gè)消費(fèi)者都可以同時(shí)從Kafka服務(wù)器讀取數(shù)據(jù)。

email.py

import kafka
import json

RIDES_CONFIRMED_KAFKA_TOPIC = "ride_confirmed"
consumer = kafka.KafkaConsumer(
RIDES_CONFIRMED_KAFKA_TOPIC, bootstrap_servers="localhost:9092"
)

print("Listening Confirmed Rides!")
while True:
for data in consumer:
message = json.loads(data.value.decode())
email = message["customer_email"]
print(f"Email sent to {email}!")


analysis.py

import kafka
import json

RIDES_CONFIRMED_KAFKA_TOPIC = "ride_confirmed"
consumer = kafka.KafkaConsumer(
RIDES_CONFIRMED_KAFKA_TOPIC, bootstrap_servers="localhost:9092"
)

print("Listening Confirmed Rides!")
while True:
for data in consumer:
message = json.loads(data.value.decode())
id = message["customer_id"]
driver_details = message["alloted_driver"]
pickup_time = message["pickup_time"]
print(f"Data sent to ML Model for analysis ({id})!")

現(xiàn)在，我們已完成了應(yīng)用程序。在下一節(jié)中，我們將同時(shí)運(yùn)行所有服務(wù)并檢查性能。

六、測(cè)試應(yīng)用程序

在四個(gè)單獨(dú)的命令提示符中逐一運(yùn)行每個(gè)文件。

$ python transaction.py

$ python email.py

$ python analysis.py

$ python ride.py

圖6

當(dāng)打車詳細(xì)信息被推送到服務(wù)器時(shí)，您可以同時(shí)接收來(lái)自所有文件的輸出。您還可以通過(guò)刪除rides.py文件中的延遲函數(shù)來(lái)提高處理速度。'rides.py'文件將數(shù)據(jù)推送到Kafka服務(wù)器，另外三個(gè)文件同時(shí)從Kafka服務(wù)器讀取數(shù)據(jù)，并發(fā)揮相應(yīng)的作用。

但愿您對(duì)Apache Kafka以及如何實(shí)現(xiàn)它已有了基本的了解。

七、結(jié)語(yǔ)

我們?cè)诒疚闹辛私饬薃pache Kafka工作原理及實(shí)際實(shí)現(xiàn)該架構(gòu)的出租車預(yù)訂應(yīng)用程序用例。使用Kafka設(shè)計(jì)一條可擴(kuò)展的管道需要認(rèn)真計(jì)劃和實(shí)施。您可以增加代理和分區(qū)的數(shù)量，提高這些應(yīng)用程序的可擴(kuò)展性。每個(gè)分區(qū)都獨(dú)立處理，這樣負(fù)載可以在它們之間予以分配。此外，您還可以通過(guò)設(shè)置緩存大小、緩沖區(qū)大小或線程數(shù)量來(lái)優(yōu)化Kafka配置。

本文中使用的完整代碼的GitHub鏈接如下：https://github.com/aryan0141/apache-kafka-tutorial/tree/master。

原文鏈接：https://www.kdnuggets.com/2023/04/build-scalable-data-architecture-apache-kafka.html