建站
咨詢
售后
建站咨詢
新聞中心
?Apache Kafka是一種基于發(fā)布者-訂閱者模型的分布式消息傳遞系統(tǒng)。它由Apache軟件基金會開發(fā)，用Java和Scala編寫。Kafka的初衷是為了克服傳統(tǒng)消息傳遞系統(tǒng)的分布和可擴展性面臨的問題。它可以處理和存儲大量數(shù)據(jù)，具有延遲低、吞吐量高的優(yōu)點。因此，它適用于構建實時數(shù)據(jù)處理應用程序和流媒體服務。它目前已開源，被Netflix、沃爾瑪和領英等許多組織使用。
創(chuàng)新互聯(lián)專注于平南企業(yè)網(wǎng)站建設,成都響應式網(wǎng)站建設,購物商城網(wǎng)站建設。平南網(wǎng)站建設公司,為平南等地區(qū)提供建站服務。全流程定制網(wǎng)站建設，專業(yè)設計，全程項目跟蹤，創(chuàng)新互聯(lián)專業(yè)和態(tài)度為您提供的服務
消息傳遞系統(tǒng)使多個應用程序可以相互收發(fā)數(shù)據(jù)，不用擔心數(shù)據(jù)的傳輸和共享。點到點和發(fā)布者-訂閱者是兩種廣泛使用的消息傳遞系統(tǒng)。在點到點模型中，發(fā)送方將數(shù)據(jù)推送到隊列，接收方從隊列中彈出數(shù)據(jù)，就像遵循先進先出（FIFO）原則的標準隊列系統(tǒng)一樣。此外，一旦讀取數(shù)據(jù)，數(shù)據(jù)就被刪除，并且每次只允許一個接收方。接收方讀取消息時不存在時間依賴關系。
圖1. 點對點消息系統(tǒng)
在發(fā)布者-訂閱者模型中，發(fā)送方稱為發(fā)布者，接收方稱為訂閱者。在這種模型中，多個發(fā)送方和接收方可以同時讀取或寫入數(shù)據(jù)。但是它有時間依賴關系。消費者必須在一定的時間之前消費消息，因為此后消息被刪除，即使它沒有被讀取。這個時間限制可能是一天、一周或一個月，視用戶的配置而定。
圖2. 發(fā)布者-訂閱者消息系統(tǒng)
一、Kafka的架構
Kafka架構由幾個關鍵組件組成：
1. 主題
2. 分區(qū)
3. 代理
4. 生產(chǎn)者
5. 消費者
6. Kafka集群
7. Zookeeper
圖3. Kafka的架構
不妨簡單了解一下每個組件。
Kafka將消息存儲在不同的主題中。主題是一個組，含有特定類別的消息。它類似數(shù)據(jù)庫中的表。主題由名稱作為唯一標識符。不能創(chuàng)建名稱相同的兩個主題。
主題進一步劃分為分區(qū)。這些分區(qū)的每個記錄都與一個名為Offset的唯一標識符相關聯(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í)行類似任務的消費者將組成一個組，名為消費者組。
還有其他系統(tǒng)，比如代理（Broker）和Zookeeper，它們在Kafka服務器的后臺運行。代理是維護和保存已發(fā)布消息記錄的軟件。它還負責使用offset以正確的順序將正確的消息傳遞給正確的使用者。相互之間進行集體通信的代理集可以稱為Kafka集群。代理可以動態(tài)添加到Kafka集群中或從集群中動態(tài)刪除，系統(tǒng)不會遇到任何停機。Kafka集群中的其中一個代理名為控制器。它負責管理集群內(nèi)的狀態(tài)和副本，并執(zhí)行管理任務。
另一方面，Zookeeper負責維護Kafka集群的健康狀態(tài)，并與該集群的每個代理進行協(xié)調(diào)。它以鍵值對的形式維護每個集群的元數(shù)據(jù)。
本教程主要介紹實際實現(xiàn)Apache Kafka的例子。
二、出租車預訂應用程序：實際用例
以優(yōu)步之類的出租車預訂服務這一用例為例。這個應用程序使用Apache Kafka通過各種服務（比如事務、電子郵件、分析等）發(fā)送和接收消息。
圖4出租車應用程序架構圖
架構由幾個服務組成。Rides服務接收來自客戶的打車請求，并將打車詳細信息寫入到Kafka消息系統(tǒng)上。
然后，Transaction服務讀取這些訂單詳細信息，確認訂單和支付狀態(tài)。在確認這趟打車之后，該Transaction服務將再次在消息系統(tǒng)中寫入確認的打車信息，并添加一些額外的詳細信息。最后，電子郵件或數(shù)據(jù)分析等其他服務讀取已確認的打車細節(jié)，并向客戶發(fā)送確認郵件，并對其進行一些分析。
我們可以以非常高的吞吐量和極小的延遲實時執(zhí)行所有這些進程。此外，由于Apache Kafka能夠橫向擴展，我們可以擴展這個應用程序以處理數(shù)百萬用戶。
三、上述用例的實際實現(xiàn)
本節(jié)包含在我們的應用程序中實現(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:驅動器解壓已下載的zip文件，并將文件夾重命名為/apache-kafka。
3）父目錄包含兩個子目錄：/bin和/config，分別含有Zookeeper和Kafka服務器的可執(zhí)行文件和配置文件。
2.配置步驟
首先，我們需要為Kafka和Zookeeper服務器創(chuàng)建日志目錄。這些目錄將存儲這些集群的所有元數(shù)據(jù)以及主題和分區(qū)的消息。
注意：默認情況下，這些日志目錄創(chuàng)建在/tmp目錄中，這是一個易變目錄：當系統(tǒng)關閉或重啟時，該目錄中的所有數(shù)據(jù)都會消失。我們需要為日志目錄設置永久路徑來解決這個問題。不妨看看怎么做。
導航到apache-kafka >> config，打開server.properties文件。在這里您可以配置Kafka的許多屬性，比如日志目錄路徑、日志保留時間和分區(qū)數(shù)量等。
在server.properties文件中，我們必須將日志目錄文件的路徑從臨時/tmp目錄改為永久目錄。日志目錄含有Kafka Server中的生成或寫入的數(shù)據(jù)。若要更改路徑，將log.dirs變量由/tmp/kafka-logs改為c:/apache-kafka/kafka-logs。這將使您的日志永久存儲。
log.dirs = c: / apache-kafka / kafka-logs
Zookeeper服務器還包含一些日志文件，用于存儲Kafka服務器的元數(shù)據(jù)。若要更改路徑，重復上面的步驟，即打開zookeeper.properties文件，并按如下方式替換路徑。
dataDir = c: / apache-kafka / zookeeper-logs
該Zookeeper服務器將充當Kafka服務器的資源管理器。
四、運行Kafka和Zookeeper服務器
若要運行Zookeeper服務器，在父目錄中打開一個新的cmd提示符，并運行以下命令。
$ .\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties
圖5
保持Zookeeper實例運行。
若要運行Kafka服務器，打開一個單獨的cmd提示符，并執(zhí)行以下代碼：
$ .\bin\windows\kafka-server-start.bat .\config\server.properties
保持Kafka和Zookeeper服務器運行；在下一節(jié)中，我們將創(chuàng)建生產(chǎn)者和消費者函數(shù)，它們用于讀取數(shù)據(jù)并將數(shù)據(jù)寫入到Kafka服務器。
五、創(chuàng)建生產(chǎn)者和消費者函數(shù)
為了創(chuàng)建生產(chǎn)者和消費者函數(shù)，我們將以前面討論的電子商務應用程序為例?！坝唵巍狈諏⒊洚斏a(chǎn)者，將訂單細節(jié)寫入到Kafka服務器，而電子郵件和分析服務將充當消費者，從服務器讀取該數(shù)據(jù)。交易服務將充當消費者和生產(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)
解釋：
首先，我們導入了所有必要的庫，包括Kafka。然后，定義主題名稱和各項目的列表。記住，主題是一個含有類似類型消息的組。在本例中，該主題將包含所有訂單。
然后，我們創(chuàng)建一個KafkaProducer函數(shù)的實例，并將其連接到在localhost:9092上運行的Kafka服務器。如果您的Kafka服務器在不同的地址和端口上運行，那么您必須在那里提及服務器的IP和端口號。
之后，我們將生成一些JSON格式的訂單，并根據(jù)定義的話題名稱將它們寫入到Kafka服務器。睡眠函數(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服務器讀取打車細節(jié)，并在確認打車后將其再次寫入到Kafka服務器。
現(xiàn)在，創(chuàng)建兩個名為email.py和analysis .py的Python文件，分別用于向客戶發(fā)送電子郵件以確認打車和執(zhí)行一些分析。創(chuàng)建這些文件只是為了表明甚至多個消費者都可以同時從Kafka服務器讀取數(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)在，我們已完成了應用程序。在下一節(jié)中，我們將同時運行所有服務并檢查性能。
六、測試應用程序
在四個單獨的命令提示符中逐一運行每個文件。
$ python transaction.py
$ python email.py
$ python analysis.py
$ python ride.py
圖6
當打車詳細信息被推送到服務器時，您可以同時接收來自所有文件的輸出。您還可以通過刪除rides.py文件中的延遲函數(shù)來提高處理速度。'rides.py'文件將數(shù)據(jù)推送到Kafka服務器，另外三個文件同時從Kafka服務器讀取數(shù)據(jù)，并發(fā)揮相應的作用。
但愿您對Apache Kafka以及如何實現(xiàn)它已有了基本的了解。
七、結語
我們在本文中了解了Apache Kafka工作原理及實際實現(xiàn)該架構的出租車預訂應用程序用例。使用Kafka設計一條可擴展的管道需要認真計劃和實施。您可以增加代理和分區(qū)的數(shù)量，提高這些應用程序的可擴展性。每個分區(qū)都獨立處理，這樣負載可以在它們之間予以分配。此外，您還可以通過設置緩存大小、緩沖區(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
網(wǎng)站題目：如何用ApacheKafka搭建可擴展的數(shù)據(jù)架構？
網(wǎng)頁網(wǎng)址：http://www.5511xx.com/article/codiipg.html
