基于Storm的流媒體實時傳輸系統①

翁小松,張 征

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

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

20世紀的90年代出現了流處理的概念,最早應用于數據庫技術中,而分布式流處理系統由原有的分布式系統發展而來,S4、Twitter Storm、Spark Streaming 等技術的發展克服了傳統流處理技術在數據傳輸和資源分配中的不足之處,由此分布式流處理技術取代了集中式流處理技術[2].Storm 技術開源于2011年,其有著非常優異的實時性、容錯性、魯棒性、可擴展性等特點,被廣泛應用于金融、交通、電子等服務行業和實時數據計算、實時推薦系統等.

為了將Storm 在流處理中的優異性能應用到流媒體的傳輸中,本文將分析流媒體的視頻數據在實時傳輸中的難點和關鍵技術,之后在Linux 上完成Storm框架的搭建,設計基于Storm 平臺的分布式計算系統和任務拓撲用于流媒體視頻數據的實時傳輸,并部署Zookeeper 為框架提供高效可靠的分布式協調服務,最后,框架通過了大規模流媒體數據的傳輸測試,為框架在實際生產生活中的應用提供參考.

1 需求分析和架構設計

本文的流媒體視頻數據傳輸系統采用大數據流式計算框架Storm 對視頻源數據進行采集、切分、壓縮、推流,最后存儲到流媒體服務器中.本章是針對系統的需求分析,講述流媒體傳輸中的難點和關鍵技術,設計整體系統架構(如圖1所示)和各個部分的實現方法.

1.1 流媒體大數據實時傳輸的需求分析

流媒體視頻數據的傳輸依賴于流媒體技術,該技術與常規的視頻媒體技術之間最大的不同之處在于其可以使流媒體實現邊下載邊播放,兩者同時進行的實時工作模式,是一種被廣泛應用于視頻直播、遠程教育、網絡電臺等的新技術.

實現流媒體視頻數據的傳輸主要有以下幾個難點和需要用到的關鍵技術:

(1)流媒體傳輸的實現需要專用的服務器、播放器和合適的傳輸協議,TCP 協議由于其過多的網絡開銷不適用于流媒體技術,所以采用HTTP/TCP 協議來傳輸系統的控制信息,用RTP/UDP 協議來傳輸實時的視頻數據[3].為了能夠把服務器的輸出重定向到客戶機的目的地址,需要使用上述兩種協議來與服務器建立聯系.

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

(3)流媒體視頻的傳輸需要使用緩存技術.由于網絡是動態波動的,在視頻數據分段后每個分組最終所采用的路由是不同的,導致其到達客戶端所使用的的時間也不同,這時就需要使用緩存技術來保證分組后的數據的時序性,使得輸出做到連續性.

(4)在Storm 框架內傳遞的數據格式是結構化的,不能直接處理非結構化的視頻數據格式.本文實現一個特定的序列化封裝器,在Storm 平臺上對流媒體視頻源數據經采集、切塊、分組后,用于對切分后的視頻片段進行對象的序列化和在視頻推流階段的反序列化,使得視頻數據能在Storm 平臺上高效可靠的傳輸.

圖1 系統總體架構

1.2 系統整體架構設計

本文所搭建的流媒體視頻數據傳輸系統主要分為兩個部分:用于數據處理和傳輸的Storm 平臺、用于數據存儲和點播的流媒體服務器.

在Storm 的核心代碼任務拓撲Topology 的主方法入口main 函數中配置了流媒體的視頻源地址及推流地址,在將任務拓撲提交到Storm 集群環境中運行后,由第一部分Storm 平臺負責流媒體視頻數據的接收、分片、壓縮和推流等工作,完成系統的核心處理功能部分.在Storm 的第一個Spout 模塊進行源視頻的接收和分片,將處理之后的數據發送到之后的Bolt 模塊進行視頻數據的壓縮和推送任務,每個任務分別由一個Bolt 負責,減少任務之間的耦合度.Storm 平臺上的各個模塊之間的協同工作是由其獨特的拓撲結構保證的,本文采用Storm 的默認調度器來進行任務的資源分配和負載均衡,提高系統的數據傳輸效率.

流媒體服務器用于流媒體視頻數據在經Storm 平臺推流后的緩存和提供客戶端播放器點播.系統采用開源的Nginx 輕量級流媒體服務器,采用RTMP 協議進行視頻數據的傳輸,提供視頻流的拉取和點播服務,同時保證了高并發性和穩定性的要求.同時在視頻服務器上集成了FFmpeg 多媒體視頻處理工具用于視頻信息的解析、推流等.

流媒體視頻數據緩存到流媒體服務器上后,可以通過客戶端的播放器(VLC)進行網絡串流,調用流媒體服務器的視頻存儲端口地址來進行拉流和點播.

2 系統搭建

以Storm 框架作為流媒體視頻數據傳輸的基礎并提供低延遲和高可靠性保證.本章將圍繞Storm 框架在Linux 服務器上的搭建流程展開.介紹Storm 框架搭建所需的包括硬件環境、軟件環境和框架的后臺環境配置,部署Zookeeper 提供高可用的協調解決方案,在框架搭建好后啟動Storm 和提交任務拓撲.

2.1 Storm 部署與框架搭建

本文搭建的Storm 框架的硬件環境配置如下:

華為彈性云服務器上搭建Ubuntu64 位系統,并安裝可視化界面用于Storm 集群信息的查看需要,主節點雙核單處理器,4 GB 內存,40 GB 硬盤,副節點單核單處理器,2 GB 內存,20 GB 硬盤.

Storm 所需的軟件開發環境如下:

操作系統: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 作為一個項目管理工具用于系統的項目代碼管理,包括依賴包源碼的下載編譯,程序的jar 打包,必要時還可以充當bug 調試工具.

在Storm 的所有Supervisor 節點中,需要部署Zookeeper 分布式應用程序協調服務作為協調者,在集群運行時發揮如下作用[4]:

(1)Nimbus 和Supervisor 之間是沒有直接的信息傳遞的,Nimbus 在接收Storm 集群的任務拓撲后,將任務信息寫入Zookeeper 中,提供給Supervisor 從節點讀取這些任務的狀態信息,從而分配資源;

(2)在Task 執行失敗或Supervisor 節點宕機時,Zookeeper 可以獲得失敗信息,使得Nimbus 主節點可以根據心跳信息來重啟失敗的Task 或Supervisor.

2.2 Storm 啟動和任務提交

Storm 的啟動依賴于JDK 環境,Zookeeper 的部署是必要的,便于監控整個Storm 集群的狀態信息.Zookeeper 的部署需要使用Maven 進行編譯,本文所采用的的Zookeeper3.4.14 版本是編譯好的版本,所以省略這一步驟.在Storm 框架及其所需的一系列軟件開發環境搭建完畢后,可以啟動Storm 服務,需要先運行Zookeeper 服務,首先進入Zookeeper 的安裝目錄下,執行如下命令:bin/zkServer.sh start,啟動服務后可以運行bin/zkServer.sh status 命令查看Zookeeper 的運行狀態.Zookeeper 成功運行后,進入Storm 的bin 目錄下,運行下列命令完成Storm 框架的啟動:

./storm nimbus ＆

./storm ui ＆

./storm supervisor ＆

./storm logviewer ＆

./storm drpc ＆

在這些命令中,Nimbus 主節點的啟動應優先于Supervisor 副節點,防止集群信息報錯,UI 和Logviewer服務的啟動順序沒有要求,DRPC 用于分布式遠程調用Storm 集群的計算資源,而省略連接集群中的具體節點的過程,該項服務和Logviewer 都是可選項,只有在具體任務需要時才會發揮作用.

打開瀏覽器,進入localhost:8080 查看WebUI 界面(如圖2所示),驗證Storm 是否啟動成功(所有Storm 的進程必須在后臺運行,否則會占用終端控制臺).Storm 框架的各個組件的運行情況.

圖2 Storm 的WebUI 圖

在WebUI 圖中,第一欄顯示的是Storm 的版本信息、Supervisor 啟動個數、已使用和未使用的端口數量及端口總數、程序中所定義的Executors (線程數)和Tasks (任務數).WebUI 圖中還包括Nimbus 主節點所在的服務器信息,集群上所運行的Topology (任務拓撲)的相關信息、Supervisor 從節點的服務器信息以及集群的配置信息等,每一類信息都提供了查看更深層次細化信息的接口按鈕.

Storm 任務提交和執行過程示意圖如圖3所示.

Storm 的整體工作流程可以簡化為以下步驟:客戶端新建Topology,在其中定義Spout 和Bolt 的初始并發度,即初始的Executors 個數,并定義各個組件之間的流分組策略,之后Client 提交Topology 到Nimbus中;Nimbus 分配任務,根據Topology 定義中給定的參數,下載對應的依賴包的源代碼數據,并將分配好的任務提交到Zookeeper 上;子節點Supervisor 會通過定期查詢Zookeeper 中的信息,分配具體的Worker 以及Executors 執行具體的Tasks[5].

圖3 Storm 任務提交和執行過程示意圖

Storm 的Topology 拓撲任務可以通過Maven 編譯打包成jar,相關的配置信息都集中在pom.xml 文件中,包括各個依賴包資源及其版本號等,在pom.xml 文件所在目錄下運行mvn package assembly:single 命令可以在編譯測試后創建Target 目錄并生成一個xxxjar-with-dependencies.jar 文件,這個文件中包含了Storm集群環境運行所需的依賴資源和工程源代碼[6].在打包完成后,運行如下命令:

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

其中,/home/song/storm/xxx.jar 是程序實現代碼經Maven 命令編譯打包成的包含依賴包資源在內的jar 文件及其所在的相對于終端位置所在的文件路徑,cn.storm.topology.WordCountTopology 是程序的主方法入口,如果測試程序要在集群上運行,需要在命令后面追加任務的名字,否則會在本地模擬模式下運行.

2.3 流媒體服務器搭建

在流媒體視頻流數據經過Storm 平臺的處理到達推送步驟后,需要流媒體服務器進行視頻流的接收和分發.本文搭建的RTMP 流媒體服務器是基于Nginx開源項目的輕量級流媒體服務器,作為流媒體視頻數據的存儲服務器,同時提供第三方客戶端使用播放器進行視頻的網絡串流點播[7].

Nginx 的搭建流程如下:

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

(2)安裝依賴庫libpcre3,libpcre3-dev;

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

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

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

FFmpeg 是一個音視頻軟編解碼和RTMP 流發送接收的完整解決方案,本系統的視頻數據采用H.264進行編解碼,使用FFmpeg 命令執行視頻流的拉流轉推任務,完成對視頻數據的相關處理操作.

3 系統測試

在Storm 框架搭建完畢后,編寫Topology 拓撲任務用于視頻數據的接收、分片、壓縮、推流,并提交Storm 集群執行,進行流媒體大數據的傳輸測試,使其滿足流媒體對于數據傳輸的實時性、高容錯性、數據完整性的要求.

3.1 拓撲任務設計

針對在Storm 平臺上傳輸的流媒體視頻流需要選擇一個合理的切塊分組方案,能夠方便后續的連續化處理和壓縮推流等操作.視頻切分的依據是視頻編碼時的關鍵幀信息,流媒體視頻數據中主要包括3 種類型的編碼幀信息:I 幀(關鍵幀)、B 幀和P 幀[8].

系統采用GOP (2 個關鍵幀之間的間隔)作為視頻流數據切分的基本單位,通過對前一個GOP 的冗余片段進行切分,將切分后的視頻片段順序發送到Storm中,以此形成連續化的視頻組數據流[9].

序列化是流媒體視頻數據在Storm 平臺上傳輸的關鍵步驟,它將非結構化的視頻格式對象轉換為結構化的數據類型,使得視頻數據可以在網絡中進行傳遞,之后在系統的推流階段對數據進行反序列化,將視頻數據的類型還原,完成數據的傳輸過程.在Storm 框架中,集成了Kyro 序列化技術,Kyro 序列化處理后的數據占用更少的內存,比通用的Java 序列化的效率更高,耗時更少[10].本文在Kryo 序列化的基礎上編寫了更加適用于系統的新序列化器,便于視頻數據在Storm 平臺上的傳遞.

傳統的視頻格式如rmvb、mkv 等占用的容量太大,不利于視頻數據在網絡中的傳輸,所以需要在傳輸前進行格式的壓縮.對于流媒體視頻數據的壓縮來講,影響視頻最終性能的因素有很多,主要是壓縮效率和速率調節方面,本文在對視頻數據進行壓縮階段采用H.264 視頻編碼算法,擁有更節約的碼率、更高質量的視頻畫面和更強的網絡適應性,并提供了差錯恢復能力.視頻流切分流程圖如圖4所示.

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

WorkFrames 封裝類型用于對Storm 接收的視頻源經過切分后的數據片段實例化.Storm 平臺在讀取了流媒體視頻流分段數據后,無法直接使其在系統中進行信息的傳遞,自定義的WorkFrames 數據結構和序列化器WorkFramesSerializer 如圖5所示,用于對視頻數據進行分組轉換和序列化操作,使其能進行實時流傳遞,其中的sequenceId 表示Spout 采集到的視頻數據片段的序號,而streamId 表示在Storm 平臺上傳遞的視頻流編號,getTuple 方法用于獲取Bolt 中的視頻數據,metadata 中的Object 對象存儲了實際序列化后的流媒體視頻數據.WorkFramesSerializer 序列化器中的write 方法序列化視頻數據,而read 方法則是反序列化方法,fromTuple 方法和toTuple 方法用于write 序列化和read 序列化的抽象封裝,傳遞Storm 中的數據單元.

圖5 對象封裝類和序列化器封裝類

在系統的實時傳輸作業中,Spout 數據采集器模塊負責讀取RTMP 協議的視頻流,并對視頻數據文件進行切分,序列化切分后的視頻片段,使其能在Storm 平臺上進行數據傳遞,然后將數據發送給拓撲中的下一個組件.Bolt 組件在接收到Spout 的數據信息后對其進行并行壓縮操作,最后將反序列化后的視頻數據推流到流媒體服務器中(由2 個Bolt 分別處理,減少邏輯之間的耦合),提供播放器的拉流點播.

新建VideoChannelTopology 作為整個拓撲的主類,在主方法入口main 函數中添加流媒體視頻源地址信息、Storm 集群配置信息、Spout 和Bolt 各組件模塊的創建信息和數據分組模式等.具體拓撲圖如圖6.

圖6 視頻實時傳輸拓撲圖

GatherSpout 是整個拓撲任務的數據采集模塊,負責對流媒體視頻流數據的采集,并完成對視頻文件的切塊和分組操作,之后調用基于kyro 的自定義序列化器WorkFramesSerializer 對封裝后的WorkFrames 視頻對象進行序列化,將序列化后的數據單元tuple 并發的發送到下游的多個CompressBolt 中.

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

流媒體服務器上存儲的視頻文件可以通過流媒體播放器(VLC 等)進行網絡串流,采用RTMP 實時消息傳輸協議,通過配置流媒體服務器的地址IP 和對應的端口號打開遠程鏈接,拉取視頻數據并播放.

3.2 性能測試

在項目的程序代碼編寫完畢后,通過Maven 命令打包成jar 在測試通過后提交到服務器的Storm 集群執行.Storm 的Topology 拓撲運行細節如圖7所示.

圖7 拓撲運行圖

通過在系統集群的UI 中進行數據量的定量統計和實時性分析,主要集中在整體拓撲性能和集群中分組件傳輸效率,表1為截取10 min、30 min、1 h 的數據分析圖,整理后得出該集群在傳輸效應方面的數據表.

表1 拓撲統計信息表

在拓撲信息統計表中,emitted 表示窗口發射出去的元組總數,即輸出收集器上調用emit 方法的次數,transferred 則表示基于時間窗口所有發送至任務的元組數量,元組發送到流上可能沒有組件立即訂閱讀取,這會使得其數量少于發射的數量.

在Spout 的信息統計表(表2)中,emitted 表示該組件上已發射元組數量,transferred 表示發送至其他任務上的元組數量,acked 表示應答的元組數,在任務設計時Spout 作為消息收集器不作消息應答,所以該項數值始終為0,failed 表示失敗的元組數量.

表2 gatherSpout 統計信息表

在Bolt 的信息統計表(表3、表4)中,emitted 和transferred 分別表示該Bolt 上發射的元組數量和實際傳輸元組數量,由于pushingBolt 作為拓撲中的最后一個節點,不需要再將消息發送給下一個組件,所以其transferred 的值始終為0,executed 表示該Bolt 上處理的元組數量,acked 表示應答的元組數量,兩者數量相等表示消息的可靠性高,failed 表示失敗的元組數量.

表3 CompressBolt 統計信息表

表4 PushingBolt 統計信息表

通過折線圖(圖8～圖10)可以看出,系統在運行過程中的數據處理能力相對平穩,在3 個時間點的信息處理量基本符合等比增長規律,單位時間的系統性能沒有遭遇瓶頸.在集群的數據吞吐量方面,在綜合考慮帶寬因素影響的前提下,基本滿足視頻數據的傳輸需要.在實時性和可靠性方面,通過幾個折線圖的橫向對比,消息在流中傳輸時,經過處理后由acked 進行應答,兩者比例接近100%,說明消息的每次處理都收到了系統的反饋,沒有出現失敗的消息元組.

本文還進行了系統的數據傳輸量測試,針對的是Worker 并發數的變化所引起的對視頻處理性能的影響.具體的測試方法是在Storm 平臺的視頻壓縮處理模塊進行多路視頻流的并發傳輸任務,測試時在拓撲任務的主方法中配置組件信息,采集器Spout 設置為8 個,由8 個Acker 來保障數據傳輸的可靠性,避免消息應答失敗導致的數據錯誤,每個消息的大小限定為100 KB,設置24 個CompressBolt 用于視頻流數據的壓縮任務.在集群上設置了多個10 s 時間段的消息記錄器,具體任務測試時設置不同數量的Workers 任務并發度,統計集群在每10 s 所處理的Tuple 數量,繪制出不同Workers 并發數下的Tuple 處理量的折線圖(圖11),以此直觀的表現系統的集群性能變化情況.

圖8 拓撲信息統計折線圖

圖9 Spout 信息統計折線圖

圖10 Bolt 信息統計折線圖

對于視頻作業任務而言,每增加一個Worker 數,會增加同時間段內系統的數據處理數和系統平臺的線程處理量,如Netty 收發線程和心跳線程,同時會使得原來在線程間內存通信的組件變成網絡通信,降低了系統的吞吐量[12].Workers 進程數量的增加不會造成系統傳輸效率的無限增長,在Workers 過大時反而會因信息阻塞造成系統性能的下降.經測試發現,在8 個Workers 并發度的影響下,系統的傳輸效率達到了最大化.影響系統傳輸性能的因素還有很多:服務器的硬件條件,Storm 集群的任務調度算法,拓撲任務的消息復雜度等,其對系統的影響并不都是線性和正相關的,這有待于后續更深入的研究.

圖11 不同Workers 數對系統性能的影響

4 結語

流媒體的實時可靠傳輸是當今網絡技術實踐中一個重要的研究內容,被廣泛應用于各種視頻資源網站的點播和視頻直播中.Storm 作為一個免費開源的分布式實時計算框架,設計用于在容錯和水平可擴展方法中處理大量數據,利用Storm 可以很容易做到可靠地處理無限的數據流,很好地滿足了大數據在容量不斷膨脹擴大,又對實時性有著高要求的現狀[13].本文分析了流媒體視頻數據傳輸的原理和需要解決的問題難點及所需的關鍵性技術,設計并實現了基于Storm 的流媒體視頻數據傳輸系統.之后在云服務器的Linux 系統中搭建了Storm 集群,并部署了Zookeeper 分布式協調服務.在Storm 框架搭建成功后,編寫任務拓撲實現視頻的接收、分片、壓縮和推流,搭建流媒體服務器用于流媒體視頻數據經Storm 平臺推流后的存儲和客戶端播放器的拉取點播,之后進行了流媒體視頻大數據的傳輸測試,保證了Storm 框架能完成大數據的實時可靠、高并發量的傳輸.