1. 啟動Kafka Server
bin/kafka-server-start.sh config/server.properties &
2. 創建一個新topic
bin/kafka-topics.sh --create --zookeeper xxxx --replication-factor 1 --partitions 1 --topic video
3. 安裝相關依賴
sudo pip-3.6 install kafka-python opencv-contrib-python imutils
4. 創建一個 Kafka Producer，並發送到kafka消息隊列
# kafkaProducer
def publish_video(server, topic):
# start producer
producer = KafkaProducer(bootstrap_servers=server)
vs = VideoStream(src=0).start()
time.sleep(2.0)
print("publishing video...")
while True:
frame = vs.read()
frame = imutils.resize(frame, width=400)
frame = detection(frame, 'pretrained.prototxt.txt', 'pretrained.caffemodel')
# send to kafka topic
producer.send(topic, frame.tobytes())
vs.stop()
這里使用opencv 采集本地攝像頭視頻，讀取每一幀數據，做圖像識別處理，並轉化為bytes數據發送到 kafka topic。
這里選擇了在Producer端直接對圖像做處理，然后直接發送到流，Consumer端只做圖像顯示。
其中detection方法為以一個已經訓練好的深度學習模型，用於做圖像識別以及描繪邊框。網上類似模型很多，這里不多做贅述。這里對每幀做了一個resize是由於原視頻采集的每幀數據較大，超過了kafka里默認的一個item大小，所以需要裁剪每幀，以減少傳輸數據量。
5. 驗證是否可以接收到流數據：
bin/kafka-console-consumer.sh --bootstrap-server xxxx:port --topic video
6. 創建一個Kafka Consumer，用於獲取流數據
from imutils.video import VideoStream
from imutils.video import FPS
import imutils
import numpy as np
import time
import cv2
from kafka import KafkaConsumer
import sys
def showCam(server, topic):
consumer = KafkaConsumer(
topic,
bootstrap_servers=[server])
fps = FPS().start()
for msg in consumer:
decoded = np.frombuffer(msg.value, np.uint8)
decoded = decoded.reshape(225, 400, 3)
cv2.imshow("Cam", decoded)
key = cv2.waitKey(1) & 0xFF
if key == ord("q"):
break
fps.update()
fps.stop()
cv2.destroyAllWindows()
這里使用np.frombuffer() 方法將每一幀的bytes數據轉為一維numpy數組，由於采集的幀數據為3維numpy數組，所以需要對此數組做reshape，以還原為原數據格式，最后顯示在屏幕上。
7. 執行代碼：
首先啟動 Consumer：python3 kafkaCCam.py server:port topic
然后啟動 Producer：python3 KafkaPCam.py server:port topic
即可在Consumer端獲取到Producer送入到流里的實時視頻圖像：
