基于Storm的流媒體實(shí)時(shí)傳輸系統(tǒng)①

翁小松,張 征

(華中科技大學(xué) 人工智能與自動(dòng)化學(xué)院,武漢 430074)

流媒體是指在網(wǎng)絡(luò)中使用流式傳輸技術(shù)進(jìn)行下載點(diǎn)播的連續(xù)時(shí)基媒體,采用邊下載邊播放的方式,緩解了網(wǎng)絡(luò)帶寬的壓力和節(jié)省了相對(duì)傳輸時(shí)間,因有著良好的時(shí)間效應(yīng)而被廣大用戶所采納.為了符合流媒體在數(shù)據(jù)傳輸過(guò)程中的穩(wěn)定性、時(shí)效性、可靠性等諸多要求,已有的流媒體傳輸系統(tǒng)分別采取了不同的技術(shù)和軟硬件手段,Fraz 和Malkani 通過(guò)部署高速專(zhuān)用嵌入式處理平臺(tái)(DSP),采用動(dòng)態(tài)RTP 數(shù)據(jù)打包技術(shù)提高實(shí)時(shí)視頻流系統(tǒng)的性能,滿足系統(tǒng)的高數(shù)據(jù)吞吐量,確保了更少的延遲和更好的流媒體質(zhì)量[1].還有基于各種傳輸協(xié)議如RTSP、MPEG-DASH 等,基于各種視頻壓縮技術(shù)如AVS、SVC 等,改良網(wǎng)絡(luò)帶寬自適應(yīng)和改良系統(tǒng)資源分配算法等的流媒體傳輸系統(tǒng)都在一定程度上實(shí)現(xiàn)了優(yōu)秀的視頻流傳輸性能.

20世紀(jì)的90年代出現(xiàn)了流處理的概念,最早應(yīng)用于數(shù)據(jù)庫(kù)技術(shù)中,而分布式流處理系統(tǒng)由原有的分布式系統(tǒng)發(fā)展而來(lái),S4、Twitter Storm、Spark Streaming 等技術(shù)的發(fā)展克服了傳統(tǒng)流處理技術(shù)在數(shù)據(jù)傳輸和資源分配中的不足之處,由此分布式流處理技術(shù)取代了集中式流處理技術(shù)[2].Storm 技術(shù)開(kāi)源于2011年,其有著非常優(yōu)異的實(shí)時(shí)性、容錯(cuò)性、魯棒性、可擴(kuò)展性等特點(diǎn),被廣泛應(yīng)用于金融、交通、電子等服務(wù)行業(yè)和實(shí)時(shí)數(shù)據(jù)計(jì)算、實(shí)時(shí)推薦系統(tǒng)等.

為了將Storm 在流處理中的優(yōu)異性能應(yīng)用到流媒體的傳輸中,本文將分析流媒體的視頻數(shù)據(jù)在實(shí)時(shí)傳輸中的難點(diǎn)和關(guān)鍵技術(shù),之后在Linux 上完成Storm框架的搭建,設(shè)計(jì)基于Storm 平臺(tái)的分布式計(jì)算系統(tǒng)和任務(wù)拓?fù)溆糜诹髅襟w視頻數(shù)據(jù)的實(shí)時(shí)傳輸,并部署Zookeeper 為框架提供高效可靠的分布式協(xié)調(diào)服務(wù),最后,框架通過(guò)了大規(guī)模流媒體數(shù)據(jù)的傳輸測(cè)試,為框架在實(shí)際生產(chǎn)生活中的應(yīng)用提供參考.

1 需求分析和架構(gòu)設(shè)計(jì)

本文的流媒體視頻數(shù)據(jù)傳輸系統(tǒng)采用大數(shù)據(jù)流式計(jì)算框架Storm 對(duì)視頻源數(shù)據(jù)進(jìn)行采集、切分、壓縮、推流,最后存儲(chǔ)到流媒體服務(wù)器中.本章是針對(duì)系統(tǒng)的需求分析,講述流媒體傳輸中的難點(diǎn)和關(guān)鍵技術(shù),設(shè)計(jì)整體系統(tǒng)架構(gòu)(如圖1所示)和各個(gè)部分的實(shí)現(xiàn)方法.

1.1 流媒體大數(shù)據(jù)實(shí)時(shí)傳輸?shù)男枨蠓治?/h3>

流媒體視頻數(shù)據(jù)的傳輸依賴(lài)于流媒體技術(shù),該技術(shù)與常規(guī)的視頻媒體技術(shù)之間最大的不同之處在于其可以使流媒體實(shí)現(xiàn)邊下載邊播放,兩者同時(shí)進(jìn)行的實(shí)時(shí)工作模式,是一種被廣泛應(yīng)用于視頻直播、遠(yuǎn)程教育、網(wǎng)絡(luò)電臺(tái)等的新技術(shù).

實(shí)現(xiàn)流媒體視頻數(shù)據(jù)的傳輸主要有以下幾個(gè)難點(diǎn)和需要用到的關(guān)鍵技術(shù):

(1)流媒體傳輸?shù)膶?shí)現(xiàn)需要專(zhuān)用的服務(wù)器、播放器和合適的傳輸協(xié)議,TCP 協(xié)議由于其過(guò)多的網(wǎng)絡(luò)開(kāi)銷(xiāo)不適用于流媒體技術(shù),所以采用HTTP/TCP 協(xié)議來(lái)傳輸系統(tǒng)的控制信息,用RTP/UDP 協(xié)議來(lái)傳輸實(shí)時(shí)的視頻數(shù)據(jù)[3].為了能夠把服務(wù)器的輸出重定向到客戶機(jī)的目的地址,需要使用上述兩種協(xié)議來(lái)與服務(wù)器建立聯(lián)系.

(2)進(jìn)行流媒體傳輸?shù)囊曨l文件需要使用到視頻壓縮技術(shù),將其轉(zhuǎn)換成特定的視頻格式,通常格式的視頻文件的容量太大,在進(jìn)行網(wǎng)絡(luò)傳輸時(shí)需要占用過(guò)多的資源和花費(fèi)更長(zhǎng)的時(shí)間,而進(jìn)行壓縮后可以有效減少數(shù)字視頻傳輸所需的帶寬.由于壓縮技術(shù)是以消除冗余數(shù)據(jù)為原則,會(huì)影響到圖像質(zhì)量,所以需要在處理效率、磁盤(pán)空間、視頻質(zhì)量和所需的系統(tǒng)成本之間進(jìn)行權(quán)衡.在進(jìn)行格式轉(zhuǎn)換時(shí)需要注意在文件中添加“流”信息以便于進(jìn)行后續(xù)的視頻的合理切分.

(3)流媒體視頻的傳輸需要使用緩存技術(shù).由于網(wǎng)絡(luò)是動(dòng)態(tài)波動(dòng)的,在視頻數(shù)據(jù)分段后每個(gè)分組最終所采用的路由是不同的,導(dǎo)致其到達(dá)客戶端所使用的的時(shí)間也不同,這時(shí)就需要使用緩存技術(shù)來(lái)保證分組后的數(shù)據(jù)的時(shí)序性,使得輸出做到連續(xù)性.

(4)在Storm 框架內(nèi)傳遞的數(shù)據(jù)格式是結(jié)構(gòu)化的,不能直接處理非結(jié)構(gòu)化的視頻數(shù)據(jù)格式.本文實(shí)現(xiàn)一個(gè)特定的序列化封裝器,在Storm 平臺(tái)上對(duì)流媒體視頻源數(shù)據(jù)經(jīng)采集、切塊、分組后,用于對(duì)切分后的視頻片段進(jìn)行對(duì)象的序列化和在視頻推流階段的反序列化,使得視頻數(shù)據(jù)能在Storm 平臺(tái)上高效可靠的傳輸.

圖1 系統(tǒng)總體架構(gòu)

1.2 系統(tǒng)整體架構(gòu)設(shè)計(jì)

本文所搭建的流媒體視頻數(shù)據(jù)傳輸系統(tǒng)主要分為兩個(gè)部分:用于數(shù)據(jù)處理和傳輸?shù)腟torm 平臺(tái)、用于數(shù)據(jù)存儲(chǔ)和點(diǎn)播的流媒體服務(wù)器.

在Storm 的核心代碼任務(wù)拓?fù)銽opology 的主方法入口main 函數(shù)中配置了流媒體的視頻源地址及推流地址,在將任務(wù)拓?fù)涮峤坏絊torm 集群環(huán)境中運(yùn)行后,由第一部分Storm 平臺(tái)負(fù)責(zé)流媒體視頻數(shù)據(jù)的接收、分片、壓縮和推流等工作,完成系統(tǒng)的核心處理功能部分.在Storm 的第一個(gè)Spout 模塊進(jìn)行源視頻的接收和分片,將處理之后的數(shù)據(jù)發(fā)送到之后的Bolt 模塊進(jìn)行視頻數(shù)據(jù)的壓縮和推送任務(wù),每個(gè)任務(wù)分別由一個(gè)Bolt 負(fù)責(zé),減少任務(wù)之間的耦合度.Storm 平臺(tái)上的各個(gè)模塊之間的協(xié)同工作是由其獨(dú)特的拓?fù)浣Y(jié)構(gòu)保證的,本文采用Storm 的默認(rèn)調(diào)度器來(lái)進(jìn)行任務(wù)的資源分配和負(fù)載均衡,提高系統(tǒng)的數(shù)據(jù)傳輸效率.

流媒體服務(wù)器用于流媒體視頻數(shù)據(jù)在經(jīng)Storm 平臺(tái)推流后的緩存和提供客戶端播放器點(diǎn)播.系統(tǒng)采用開(kāi)源的Nginx 輕量級(jí)流媒體服務(wù)器,采用RTMP 協(xié)議進(jìn)行視頻數(shù)據(jù)的傳輸,提供視頻流的拉取和點(diǎn)播服務(wù),同時(shí)保證了高并發(fā)性和穩(wěn)定性的要求.同時(shí)在視頻服務(wù)器上集成了FFmpeg 多媒體視頻處理工具用于視頻信息的解析、推流等.

流媒體視頻數(shù)據(jù)緩存到流媒體服務(wù)器上后,可以通過(guò)客戶端的播放器(VLC)進(jìn)行網(wǎng)絡(luò)串流,調(diào)用流媒體服務(wù)器的視頻存儲(chǔ)端口地址來(lái)進(jìn)行拉流和點(diǎn)播.

2 系統(tǒng)搭建

以Storm 框架作為流媒體視頻數(shù)據(jù)傳輸?shù)幕A(chǔ)并提供低延遲和高可靠性保證.本章將圍繞Storm 框架在Linux 服務(wù)器上的搭建流程展開(kāi).介紹Storm 框架搭建所需的包括硬件環(huán)境、軟件環(huán)境和框架的后臺(tái)環(huán)境配置,部署Zookeeper 提供高可用的協(xié)調(diào)解決方案,在框架搭建好后啟動(dòng)Storm 和提交任務(wù)拓?fù)?

2.1 Storm 部署與框架搭建

本文搭建的Storm 框架的硬件環(huán)境配置如下:

華為彈性云服務(wù)器上搭建Ubuntu64 位系統(tǒng),并安裝可視化界面用于Storm 集群信息的查看需要,主節(jié)點(diǎn)雙核單處理器,4 GB 內(nèi)存,40 GB 硬盤(pán),副節(jié)點(diǎn)單核單處理器,2 GB 內(nèi)存,20 GB 硬盤(pán).

Storm 所需的軟件開(kāi)發(fā)環(huán)境如下:

操作系統(tǒng):Ubuntu 7.4.0-1Ubuntu1-18.04.1

JDK 版本:JDK1.8.0_231

Storm 版本:Storm2.0.0

Zookeeper 版本:Zookeeper3.4.14

Python 版本:Python2.7.2

Zeromq 版本:Zeromq4.2.2

Jzmq 版本:JJzmq2.1.0-SNAPSHOT

Maven 版本:Apache-maven-2.5.5

Maven 作為一個(gè)項(xiàng)目管理工具用于系統(tǒng)的項(xiàng)目代碼管理,包括依賴(lài)包源碼的下載編譯,程序的jar 打包,必要時(shí)還可以充當(dāng)bug 調(diào)試工具.

在Storm 的所有Supervisor 節(jié)點(diǎn)中,需要部署Zookeeper 分布式應(yīng)用程序協(xié)調(diào)服務(wù)作為協(xié)調(diào)者,在集群運(yùn)行時(shí)發(fā)揮如下作用[4]:

(1)Nimbus 和Supervisor 之間是沒(méi)有直接的信息傳遞的,Nimbus 在接收Storm 集群的任務(wù)拓?fù)浜?將任務(wù)信息寫(xiě)入Zookeeper 中,提供給Supervisor 從節(jié)點(diǎn)讀取這些任務(wù)的狀態(tài)信息,從而分配資源;

(2)在Task 執(zhí)行失敗或Supervisor 節(jié)點(diǎn)宕機(jī)時(shí),Zookeeper 可以獲得失敗信息,使得Nimbus 主節(jié)點(diǎn)可以根據(jù)心跳信息來(lái)重啟失敗的Task 或Supervisor.

2.2 Storm 啟動(dòng)和任務(wù)提交

Storm 的啟動(dòng)依賴(lài)于JDK 環(huán)境,Zookeeper 的部署是必要的,便于監(jiān)控整個(gè)Storm 集群的狀態(tài)信息.Zookeeper 的部署需要使用Maven 進(jìn)行編譯,本文所采用的的Zookeeper3.4.14 版本是編譯好的版本,所以省略這一步驟.在Storm 框架及其所需的一系列軟件開(kāi)發(fā)環(huán)境搭建完畢后,可以啟動(dòng)Storm 服務(wù),需要先運(yùn)行Zookeeper 服務(wù),首先進(jìn)入Zookeeper 的安裝目錄下,執(zhí)行如下命令:bin/zkServer.sh start,啟動(dòng)服務(wù)后可以運(yùn)行bin/zkServer.sh status 命令查看Zookeeper 的運(yùn)行狀態(tài).Zookeeper 成功運(yùn)行后,進(jìn)入Storm 的bin 目錄下,運(yùn)行下列命令完成Storm 框架的啟動(dòng):

./storm nimbus ＆

./storm ui ＆

./storm supervisor ＆

./storm logviewer ＆

./storm drpc ＆

在這些命令中,Nimbus 主節(jié)點(diǎn)的啟動(dòng)應(yīng)優(yōu)先于Supervisor 副節(jié)點(diǎn),防止集群信息報(bào)錯(cuò),UI 和Logviewer服務(wù)的啟動(dòng)順序沒(méi)有要求,DRPC 用于分布式遠(yuǎn)程調(diào)用Storm 集群的計(jì)算資源,而省略連接集群中的具體節(jié)點(diǎn)的過(guò)程,該項(xiàng)服務(wù)和Logviewer 都是可選項(xiàng),只有在具體任務(wù)需要時(shí)才會(huì)發(fā)揮作用.

打開(kāi)瀏覽器,進(jìn)入localhost:8080 查看WebUI 界面(如圖2所示),驗(yàn)證Storm 是否啟動(dòng)成功(所有Storm 的進(jìn)程必須在后臺(tái)運(yùn)行,否則會(huì)占用終端控制臺(tái)).Storm 框架的各個(gè)組件的運(yùn)行情況.

圖2 Storm 的WebUI 圖

在WebUI 圖中,第一欄顯示的是Storm 的版本信息、Supervisor 啟動(dòng)個(gè)數(shù)、已使用和未使用的端口數(shù)量及端口總數(shù)、程序中所定義的Executors (線程數(shù))和Tasks (任務(wù)數(shù)).WebUI 圖中還包括Nimbus 主節(jié)點(diǎn)所在的服務(wù)器信息,集群上所運(yùn)行的Topology (任務(wù)拓?fù)?的相關(guān)信息、Supervisor 從節(jié)點(diǎn)的服務(wù)器信息以及集群的配置信息等,每一類(lèi)信息都提供了查看更深層次細(xì)化信息的接口按鈕.

Storm 任務(wù)提交和執(zhí)行過(guò)程示意圖如圖3所示.

Storm 的整體工作流程可以簡(jiǎn)化為以下步驟:客戶端新建Topology,在其中定義Spout 和Bolt 的初始并發(fā)度,即初始的Executors 個(gè)數(shù),并定義各個(gè)組件之間的流分組策略,之后Client 提交Topology 到Nimbus中;Nimbus 分配任務(wù),根據(jù)Topology 定義中給定的參數(shù),下載對(duì)應(yīng)的依賴(lài)包的源代碼數(shù)據(jù),并將分配好的任務(wù)提交到Zookeeper 上;子節(jié)點(diǎn)Supervisor 會(huì)通過(guò)定期查詢(xún)Zookeeper 中的信息,分配具體的Worker 以及Executors 執(zhí)行具體的Tasks[5].

圖3 Storm 任務(wù)提交和執(zhí)行過(guò)程示意圖

Storm 的Topology 拓?fù)淙蝿?wù)可以通過(guò)Maven 編譯打包成jar,相關(guān)的配置信息都集中在pom.xml 文件中,包括各個(gè)依賴(lài)包資源及其版本號(hào)等,在pom.xml 文件所在目錄下運(yùn)行mvn package assembly:single 命令可以在編譯測(cè)試后創(chuàng)建Target 目錄并生成一個(gè)xxxjar-with-dependencies.jar 文件,這個(gè)文件中包含了Storm集群環(huán)境運(yùn)行所需的依賴(lài)資源和工程源代碼[6].在打包完成后,運(yùn)行如下命令:

storm jar /home/song/storm/xxx.jar cn.storm.topology.WordCountTopology

其中,/home/song/storm/xxx.jar 是程序?qū)崿F(xiàn)代碼經(jīng)Maven 命令編譯打包成的包含依賴(lài)包資源在內(nèi)的jar 文件及其所在的相對(duì)于終端位置所在的文件路徑,cn.storm.topology.WordCountTopology 是程序的主方法入口,如果測(cè)試程序要在集群上運(yùn)行,需要在命令后面追加任務(wù)的名字,否則會(huì)在本地模擬模式下運(yùn)行.

2.3 流媒體服務(wù)器搭建

在流媒體視頻流數(shù)據(jù)經(jīng)過(guò)Storm 平臺(tái)的處理到達(dá)推送步驟后,需要流媒體服務(wù)器進(jìn)行視頻流的接收和分發(fā).本文搭建的RTMP 流媒體服務(wù)器是基于Nginx開(kāi)源項(xiàng)目的輕量級(jí)流媒體服務(wù)器,作為流媒體視頻數(shù)據(jù)的存儲(chǔ)服務(wù)器,同時(shí)提供第三方客戶端使用播放器進(jìn)行視頻的網(wǎng)絡(luò)串流點(diǎn)播[7].

Nginx 的搭建流程如下:

(1)安裝GCC 和相關(guān)C++工具;

(2)安裝依賴(lài)庫(kù)libpcre3,libpcre3-dev;

(3)安裝libssl-dev 和OpenSSL 工具;

(4)解壓后的Nginx 重新編譯和安裝.

如果Nginx 編譯成功,在/etc/nginx 目錄下修改配置文件nginx.conf,添加RTMP 的推流端口live,然后在/usr/local/nginx/sbin 目錄下啟動(dòng)Nginx 的主程序,啟動(dòng)后可以在瀏覽器中打開(kāi)localhost 的特定端口查看啟動(dòng)情況以及服務(wù)器的相關(guān)信息.

FFmpeg 是一個(gè)音視頻軟編解碼和RTMP 流發(fā)送接收的完整解決方案,本系統(tǒng)的視頻數(shù)據(jù)采用H.264進(jìn)行編解碼,使用FFmpeg 命令執(zhí)行視頻流的拉流轉(zhuǎn)推任務(wù),完成對(duì)視頻數(shù)據(jù)的相關(guān)處理操作.

3 系統(tǒng)測(cè)試

在Storm 框架搭建完畢后,編寫(xiě)Topology 拓?fù)淙蝿?wù)用于視頻數(shù)據(jù)的接收、分片、壓縮、推流,并提交Storm 集群執(zhí)行,進(jìn)行流媒體大數(shù)據(jù)的傳輸測(cè)試,使其滿足流媒體對(duì)于數(shù)據(jù)傳輸?shù)膶?shí)時(shí)性、高容錯(cuò)性、數(shù)據(jù)完整性的要求.

3.1 拓?fù)淙蝿?wù)設(shè)計(jì)

針對(duì)在Storm 平臺(tái)上傳輸?shù)牧髅襟w視頻流需要選擇一個(gè)合理的切塊分組方案,能夠方便后續(xù)的連續(xù)化處理和壓縮推流等操作.視頻切分的依據(jù)是視頻編碼時(shí)的關(guān)鍵幀信息,流媒體視頻數(shù)據(jù)中主要包括3 種類(lèi)型的編碼幀信息:I 幀(關(guān)鍵幀)、B 幀和P 幀[8].

系統(tǒng)采用GOP (2 個(gè)關(guān)鍵幀之間的間隔)作為視頻流數(shù)據(jù)切分的基本單位,通過(guò)對(duì)前一個(gè)GOP 的冗余片段進(jìn)行切分,將切分后的視頻片段順序發(fā)送到Storm中,以此形成連續(xù)化的視頻組數(shù)據(jù)流[9].

序列化是流媒體視頻數(shù)據(jù)在Storm 平臺(tái)上傳輸?shù)年P(guān)鍵步驟,它將非結(jié)構(gòu)化的視頻格式對(duì)象轉(zhuǎn)換為結(jié)構(gòu)化的數(shù)據(jù)類(lèi)型,使得視頻數(shù)據(jù)可以在網(wǎng)絡(luò)中進(jìn)行傳遞,之后在系統(tǒng)的推流階段對(duì)數(shù)據(jù)進(jìn)行反序列化,將視頻數(shù)據(jù)的類(lèi)型還原,完成數(shù)據(jù)的傳輸過(guò)程.在Storm 框架中,集成了Kyro 序列化技術(shù),Kyro 序列化處理后的數(shù)據(jù)占用更少的內(nèi)存,比通用的Java 序列化的效率更高,耗時(shí)更少[10].本文在Kryo 序列化的基礎(chǔ)上編寫(xiě)了更加適用于系統(tǒng)的新序列化器,便于視頻數(shù)據(jù)在Storm 平臺(tái)上的傳遞.

傳統(tǒng)的視頻格式如rmvb、mkv 等占用的容量太大,不利于視頻數(shù)據(jù)在網(wǎng)絡(luò)中的傳輸,所以需要在傳輸前進(jìn)行格式的壓縮.對(duì)于流媒體視頻數(shù)據(jù)的壓縮來(lái)講,影響視頻最終性能的因素有很多,主要是壓縮效率和速率調(diào)節(jié)方面,本文在對(duì)視頻數(shù)據(jù)進(jìn)行壓縮階段采用H.264 視頻編碼算法,擁有更節(jié)約的碼率、更高質(zhì)量的視頻畫(huà)面和更強(qiáng)的網(wǎng)絡(luò)適應(yīng)性,并提供了差錯(cuò)恢復(fù)能力.視頻流切分流程圖如圖4所示.

圖4 視頻流切分流程圖[9]

WorkFrames 封裝類(lèi)型用于對(duì)Storm 接收的視頻源經(jīng)過(guò)切分后的數(shù)據(jù)片段實(shí)例化.Storm 平臺(tái)在讀取了流媒體視頻流分段數(shù)據(jù)后,無(wú)法直接使其在系統(tǒng)中進(jìn)行信息的傳遞,自定義的WorkFrames 數(shù)據(jù)結(jié)構(gòu)和序列化器WorkFramesSerializer 如圖5所示,用于對(duì)視頻數(shù)據(jù)進(jìn)行分組轉(zhuǎn)換和序列化操作,使其能進(jìn)行實(shí)時(shí)流傳遞,其中的sequenceId 表示Spout 采集到的視頻數(shù)據(jù)片段的序號(hào),而streamId 表示在Storm 平臺(tái)上傳遞的視頻流編號(hào),getTuple 方法用于獲取Bolt 中的視頻數(shù)據(jù),metadata 中的Object 對(duì)象存儲(chǔ)了實(shí)際序列化后的流媒體視頻數(shù)據(jù).WorkFramesSerializer 序列化器中的write 方法序列化視頻數(shù)據(jù),而read 方法則是反序列化方法,fromTuple 方法和toTuple 方法用于write 序列化和read 序列化的抽象封裝,傳遞Storm 中的數(shù)據(jù)單元.

圖5 對(duì)象封裝類(lèi)和序列化器封裝類(lèi)

在系統(tǒng)的實(shí)時(shí)傳輸作業(yè)中,Spout 數(shù)據(jù)采集器模塊負(fù)責(zé)讀取RTMP 協(xié)議的視頻流,并對(duì)視頻數(shù)據(jù)文件進(jìn)行切分,序列化切分后的視頻片段,使其能在Storm 平臺(tái)上進(jìn)行數(shù)據(jù)傳遞,然后將數(shù)據(jù)發(fā)送給拓?fù)渲械南乱粋€(gè)組件.Bolt 組件在接收到Spout 的數(shù)據(jù)信息后對(duì)其進(jìn)行并行壓縮操作,最后將反序列化后的視頻數(shù)據(jù)推流到流媒體服務(wù)器中(由2 個(gè)Bolt 分別處理,減少邏輯之間的耦合),提供播放器的拉流點(diǎn)播.

新建VideoChannelTopology 作為整個(gè)拓?fù)涞闹黝?lèi),在主方法入口main 函數(shù)中添加流媒體視頻源地址信息、Storm 集群配置信息、Spout 和Bolt 各組件模塊的創(chuàng)建信息和數(shù)據(jù)分組模式等.具體拓?fù)鋱D如圖6.

圖6 視頻實(shí)時(shí)傳輸拓?fù)鋱D

GatherSpout 是整個(gè)拓?fù)淙蝿?wù)的數(shù)據(jù)采集模塊,負(fù)責(zé)對(duì)流媒體視頻流數(shù)據(jù)的采集,并完成對(duì)視頻文件的切塊和分組操作,之后調(diào)用基于kyro 的自定義序列化器WorkFramesSerializer 對(duì)封裝后的WorkFrames 視頻對(duì)象進(jìn)行序列化,將序列化后的數(shù)據(jù)單元tuple 并發(fā)的發(fā)送到下游的多個(gè)CompressBolt 中.

系統(tǒng)的Storm 平臺(tái)上一共運(yùn)行2 種類(lèi)型的Bolt 組件:CompressBolt 和PushingBolt,上游的CompressBolt采用并發(fā)方式運(yùn)行,接收從采集器Spout 組件傳來(lái)的數(shù)據(jù),調(diào)用自定義序列化器的反序列化方法得到具體的視頻數(shù)據(jù),然后對(duì)視頻分段數(shù)據(jù)采用H.264 編碼算法進(jìn)行壓縮操作,完成后賦值到新的對(duì)象output 中,進(jìn)行二次序列化成value 類(lèi)型,調(diào)用tuple 的emit 方法傳遞數(shù)據(jù)到一個(gè)下游PushingBolt 中,同時(shí)進(jìn)行ack 應(yīng)答,保證數(shù)據(jù)的可靠性.下游的PushingBolt 則負(fù)責(zé)進(jìn)行處理后的視頻數(shù)據(jù)的推送,使其保存到流媒體服務(wù)器上,由于要保證視頻片段分組順序的正確性,采用滑動(dòng)窗口模式來(lái)避免在傳輸過(guò)程中的延遲問(wèn)題,對(duì)視頻片段數(shù)據(jù)采用緩存處理,保證推送視頻服務(wù)的穩(wěn)定性和連續(xù)性[11].

流媒體服務(wù)器上存儲(chǔ)的視頻文件可以通過(guò)流媒體播放器(VLC 等)進(jìn)行網(wǎng)絡(luò)串流,采用RTMP 實(shí)時(shí)消息傳輸協(xié)議,通過(guò)配置流媒體服務(wù)器的地址IP 和對(duì)應(yīng)的端口號(hào)打開(kāi)遠(yuǎn)程鏈接,拉取視頻數(shù)據(jù)并播放.

3.2 性能測(cè)試

在項(xiàng)目的程序代碼編寫(xiě)完畢后,通過(guò)Maven 命令打包成jar 在測(cè)試通過(guò)后提交到服務(wù)器的Storm 集群執(zhí)行.Storm 的Topology 拓?fù)溥\(yùn)行細(xì)節(jié)如圖7所示.

圖7 拓?fù)溥\(yùn)行圖

通過(guò)在系統(tǒng)集群的UI 中進(jìn)行數(shù)據(jù)量的定量統(tǒng)計(jì)和實(shí)時(shí)性分析,主要集中在整體拓?fù)湫阅芎图褐蟹纸M件傳輸效率,表1為截取10 min、30 min、1 h 的數(shù)據(jù)分析圖,整理后得出該集群在傳輸效應(yīng)方面的數(shù)據(jù)表.

表1 拓?fù)浣y(tǒng)計(jì)信息表

在拓?fù)湫畔⒔y(tǒng)計(jì)表中,emitted 表示窗口發(fā)射出去的元組總數(shù),即輸出收集器上調(diào)用emit 方法的次數(shù),transferred 則表示基于時(shí)間窗口所有發(fā)送至任務(wù)的元組數(shù)量,元組發(fā)送到流上可能沒(méi)有組件立即訂閱讀取,這會(huì)使得其數(shù)量少于發(fā)射的數(shù)量.

在Spout 的信息統(tǒng)計(jì)表(表2)中,emitted 表示該組件上已發(fā)射元組數(shù)量,transferred 表示發(fā)送至其他任務(wù)上的元組數(shù)量,acked 表示應(yīng)答的元組數(shù),在任務(wù)設(shè)計(jì)時(shí)Spout 作為消息收集器不作消息應(yīng)答,所以該項(xiàng)數(shù)值始終為0,failed 表示失敗的元組數(shù)量.

表2 gatherSpout 統(tǒng)計(jì)信息表

在Bolt 的信息統(tǒng)計(jì)表(表3、表4)中,emitted 和transferred 分別表示該Bolt 上發(fā)射的元組數(shù)量和實(shí)際傳輸元組數(shù)量,由于pushingBolt 作為拓?fù)渲械淖詈笠粋€(gè)節(jié)點(diǎn),不需要再將消息發(fā)送給下一個(gè)組件,所以其transferred 的值始終為0,executed 表示該Bolt 上處理的元組數(shù)量,acked 表示應(yīng)答的元組數(shù)量,兩者數(shù)量相等表示消息的可靠性高,failed 表示失敗的元組數(shù)量.

表3 CompressBolt 統(tǒng)計(jì)信息表

表4 PushingBolt 統(tǒng)計(jì)信息表

通過(guò)折線圖(圖8～圖10)可以看出,系統(tǒng)在運(yùn)行過(guò)程中的數(shù)據(jù)處理能力相對(duì)平穩(wěn),在3 個(gè)時(shí)間點(diǎn)的信息處理量基本符合等比增長(zhǎng)規(guī)律,單位時(shí)間的系統(tǒng)性能沒(méi)有遭遇瓶頸.在集群的數(shù)據(jù)吞吐量方面,在綜合考慮帶寬因素影響的前提下,基本滿足視頻數(shù)據(jù)的傳輸需要.在實(shí)時(shí)性和可靠性方面,通過(guò)幾個(gè)折線圖的橫向?qū)Ρ?消息在流中傳輸時(shí),經(jīng)過(guò)處理后由acked 進(jìn)行應(yīng)答,兩者比例接近100%,說(shuō)明消息的每次處理都收到了系統(tǒng)的反饋,沒(méi)有出現(xiàn)失敗的消息元組.

本文還進(jìn)行了系統(tǒng)的數(shù)據(jù)傳輸量測(cè)試,針對(duì)的是Worker 并發(fā)數(shù)的變化所引起的對(duì)視頻處理性能的影響.具體的測(cè)試方法是在Storm 平臺(tái)的視頻壓縮處理模塊進(jìn)行多路視頻流的并發(fā)傳輸任務(wù),測(cè)試時(shí)在拓?fù)淙蝿?wù)的主方法中配置組件信息,采集器Spout 設(shè)置為8 個(gè),由8 個(gè)Acker 來(lái)保障數(shù)據(jù)傳輸?shù)目煽啃?避免消息應(yīng)答失敗導(dǎo)致的數(shù)據(jù)錯(cuò)誤,每個(gè)消息的大小限定為100 KB,設(shè)置24 個(gè)CompressBolt 用于視頻流數(shù)據(jù)的壓縮任務(wù).在集群上設(shè)置了多個(gè)10 s 時(shí)間段的消息記錄器,具體任務(wù)測(cè)試時(shí)設(shè)置不同數(shù)量的Workers 任務(wù)并發(fā)度,統(tǒng)計(jì)集群在每10 s 所處理的Tuple 數(shù)量,繪制出不同Workers 并發(fā)數(shù)下的Tuple 處理量的折線圖(圖11),以此直觀的表現(xiàn)系統(tǒng)的集群性能變化情況.

圖8 拓?fù)湫畔⒔y(tǒng)計(jì)折線圖

圖9 Spout 信息統(tǒng)計(jì)折線圖

圖10 Bolt 信息統(tǒng)計(jì)折線圖

對(duì)于視頻作業(yè)任務(wù)而言,每增加一個(gè)Worker 數(shù),會(huì)增加同時(shí)間段內(nèi)系統(tǒng)的數(shù)據(jù)處理數(shù)和系統(tǒng)平臺(tái)的線程處理量,如Netty 收發(fā)線程和心跳線程,同時(shí)會(huì)使得原來(lái)在線程間內(nèi)存通信的組件變成網(wǎng)絡(luò)通信,降低了系統(tǒng)的吞吐量[12].Workers 進(jìn)程數(shù)量的增加不會(huì)造成系統(tǒng)傳輸效率的無(wú)限增長(zhǎng),在Workers 過(guò)大時(shí)反而會(huì)因信息阻塞造成系統(tǒng)性能的下降.經(jīng)測(cè)試發(fā)現(xiàn),在8 個(gè)Workers 并發(fā)度的影響下,系統(tǒng)的傳輸效率達(dá)到了最大化.影響系統(tǒng)傳輸性能的因素還有很多:服務(wù)器的硬件條件,Storm 集群的任務(wù)調(diào)度算法,拓?fù)淙蝿?wù)的消息復(fù)雜度等,其對(duì)系統(tǒng)的影響并不都是線性和正相關(guān)的,這有待于后續(xù)更深入的研究.

圖11 不同Workers 數(shù)對(duì)系統(tǒng)性能的影響

4 結(jié)語(yǔ)

流媒體的實(shí)時(shí)可靠傳輸是當(dāng)今網(wǎng)絡(luò)技術(shù)實(shí)踐中一個(gè)重要的研究?jī)?nèi)容,被廣泛應(yīng)用于各種視頻資源網(wǎng)站的點(diǎn)播和視頻直播中.Storm 作為一個(gè)免費(fèi)開(kāi)源的分布式實(shí)時(shí)計(jì)算框架,設(shè)計(jì)用于在容錯(cuò)和水平可擴(kuò)展方法中處理大量數(shù)據(jù),利用Storm 可以很容易做到可靠地處理無(wú)限的數(shù)據(jù)流,很好地滿足了大數(shù)據(jù)在容量不斷膨脹擴(kuò)大,又對(duì)實(shí)時(shí)性有著高要求的現(xiàn)狀[13].本文分析了流媒體視頻數(shù)據(jù)傳輸?shù)脑砗托枰鉀Q的問(wèn)題難點(diǎn)及所需的關(guān)鍵性技術(shù),設(shè)計(jì)并實(shí)現(xiàn)了基于Storm 的流媒體視頻數(shù)據(jù)傳輸系統(tǒng).之后在云服務(wù)器的Linux 系統(tǒng)中搭建了Storm 集群,并部署了Zookeeper 分布式協(xié)調(diào)服務(wù).在Storm 框架搭建成功后,編寫(xiě)任務(wù)拓?fù)鋵?shí)現(xiàn)視頻的接收、分片、壓縮和推流,搭建流媒體服務(wù)器用于流媒體視頻數(shù)據(jù)經(jīng)Storm 平臺(tái)推流后的存儲(chǔ)和客戶端播放器的拉取點(diǎn)播,之后進(jìn)行了流媒體視頻大數(shù)據(jù)的傳輸測(cè)試,保證了Storm 框架能完成大數(shù)據(jù)的實(shí)時(shí)可靠、高并發(fā)量的傳輸.