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

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

圖1. 點對點消息系統(tǒng)

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

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

一、Kafka的架構(gòu)

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

1. 主題

2. 分區(qū)

3. 代理

4. 生產(chǎn)者

5. 消費者

6. Kafka集群

7. Zookeeper

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

不妨簡單了解一下每個組件。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

注意：本教程基于Python編程語言，使用Windows機器。

1.Apache Kafka下載步驟

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

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

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

2.配置步驟

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

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

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

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

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

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

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

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

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

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

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

圖5

保持Zookeeper實例運行。

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

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

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

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

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

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

$ pip install kafka-python

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

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

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

現(xiàn)在，創(chuàng)建一個名為' 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)入了所有必要的庫，包括Kafka。然后，定義主題名稱和各項目的列表。記住，主題是一個含有類似類型消息的組。在本例中，該主題將包含所有訂單。

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

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

消費者函數(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文件用于確認用戶所做的交易，并為他們分配司機和估計的載客時間。它從Kafka服務(wù)器讀取打車細節(jié)，并在確認打車后將其再次寫入到Kafka服務(wù)器。

現(xiàn)在，創(chuàng)建兩個名為email.py和analysis .py的Python文件，分別用于向客戶發(fā)送電子郵件以確認打車和執(zhí)行一些分析。創(chuàng)建這些文件只是為了表明甚至多個消費者都可以同時從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é)中，我們將同時運行所有服務(wù)并檢查性能。

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

在四個單獨的命令提示符中逐一運行每個文件。

$ python transaction.py

$ python email.py

$ python analysis.py

$ python ride.py

圖6

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

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

七、結(jié)語

我們在本文中了解了Apache Kafka工作原理及實際實現(xiàn)該架構(gòu)的出租車預(yù)訂應(yīng)用程序用例。使用Kafka設(shè)計一條可擴展的管道需要認真計劃和實施。您可以增加代理和分區(qū)的數(shù)量，提高這些應(yīng)用程序的可擴展性。每個分區(qū)都獨立處理，這樣負載可以在它們之間予以分配。此外，您還可以通過設(shè)置緩存大小、緩沖區(qū)大小或線程數(shù)量來優(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