郭金磊，張玉生，胡愛蘭
（華北計(jì)算機(jī)系統(tǒng)工程研究所，北京 100083）
摘要：隨著大數(shù)據(jù)技術(shù)的發(fā)展，多線程高并發(fā)等技術(shù)已經(jīng)越來越成為大數(shù)據(jù)處理中的關(guān)鍵技術(shù)。非阻塞式I/O(new I/O,NIO)技術(shù)作為一種分布式高并發(fā)技術(shù)被廣泛應(yīng)用，但對(duì)于大數(shù)據(jù)量的通信往往需要很多的時(shí)間才能完成。Google提出的Protocol Buffer序列化壓縮技術(shù)相對(duì)于傳統(tǒng)序列化效率高、時(shí)間短、使用簡(jiǎn)單。文章將傳統(tǒng)NIO技術(shù)與Protocol Buffer相結(jié)合，在分布式系統(tǒng)不同節(jié)點(diǎn)通信中，極大地降低了分布式系統(tǒng)的網(wǎng)絡(luò)負(fù)載，大大節(jié)省了數(shù)據(jù)傳輸時(shí)間。
關(guān)鍵詞：NIO（new I/O） ；Protocol Buffer ；分布式系統(tǒng)；序列化
0引言
隨著大數(shù)據(jù)技術(shù)的發(fā)展，多線程高并發(fā)等技術(shù)已經(jīng)越來越成為大數(shù)據(jù)處理中的關(guān)鍵技術(shù)，同一個(gè)節(jié)點(diǎn)中的不同線程和不同節(jié)點(diǎn)的線程間的通信越來越密切。Java NIO作為一種分布式數(shù)據(jù)傳輸技術(shù)在多線程高并發(fā)［1］的實(shí)際應(yīng)用中扮演著至關(guān)重要的角色。為減小網(wǎng)絡(luò)負(fù)載，加速分布式系統(tǒng)中網(wǎng)絡(luò)通信，迫切需要一種高效率壓縮序列化技術(shù)。
1研究現(xiàn)狀
Java NIO的核心是Channel、Buffer 和 Selector。NIO基于通道（Channel）和緩沖區(qū)（Buffer）進(jìn)行操作，通道先在選擇器注冊(cè)讀寫事件，讀數(shù)據(jù)時(shí)，當(dāng)選擇器發(fā)現(xiàn)該通道準(zhǔn)備讀完成，通道直接將數(shù)據(jù)從底層網(wǎng)卡隊(duì)列讀進(jìn)緩沖區(qū)。寫數(shù)據(jù)時(shí)，當(dāng)選擇器發(fā)現(xiàn)該通道準(zhǔn)備寫完成，通道將數(shù)據(jù)寫進(jìn)緩沖區(qū)。通道可以實(shí)現(xiàn)在緩沖區(qū)中對(duì)每個(gè)字節(jié)類似于指針對(duì)數(shù)據(jù)操作，可以來回移動(dòng)讀取數(shù)據(jù)。選擇器可以用一個(gè)單獨(dú)的線程同時(shí)監(jiān)聽管理多個(gè)通道。
傳統(tǒng)的NIO［2］都是使用Java自帶的序列化形式對(duì)傳輸數(shù)據(jù)和對(duì)象進(jìn)行序列化壓縮。這種情況下，數(shù)據(jù)壓縮率［3］較低，需要傳輸?shù)膶?duì)象數(shù)據(jù)流很大時(shí)，尤其在分布式系統(tǒng)中，容易造成網(wǎng)絡(luò)擁堵。本文在傳統(tǒng)NIO技術(shù)的基礎(chǔ)上結(jié)合Google Protocol Buffer技術(shù)實(shí)現(xiàn)了數(shù)據(jù)對(duì)象的高效序列化壓縮傳輸。
2Protocol Buffer優(yōu)點(diǎn)
Google Protocol Buffer(簡(jiǎn)稱Protobuf)是Google公司提出的混合語(yǔ)言數(shù)據(jù)標(biāo)準(zhǔn)，用于 RPC 系統(tǒng)和持續(xù)數(shù)據(jù)存儲(chǔ)系統(tǒng)。同時(shí)也可用于通信協(xié)議、數(shù)據(jù)存儲(chǔ)等領(lǐng)域的語(yǔ)言無關(guān)、平臺(tái)無關(guān)、可擴(kuò)展的序列化結(jié)構(gòu)數(shù)據(jù)格式。目前提供了C++、Java、Python三種語(yǔ)言的API。Protobuf 具有很多優(yōu)點(diǎn)：實(shí)現(xiàn)簡(jiǎn)單，壓縮速度快，傳輸速度快，存儲(chǔ)空間小。用Protobuf與Java自帶的序列化工具實(shí)現(xiàn)的對(duì)象壓縮相比，存儲(chǔ)空間大了一個(gè)數(shù)量級(jí)，時(shí)間上快了一個(gè)數(shù)量級(jí)，尤其是可以自動(dòng)生成遠(yuǎn)程過程調(diào)用協(xié)議（Remote Procedure Call Protocol, RPC）的數(shù)據(jù)結(jié)構(gòu),特別是service業(yè)務(wù)邏輯，是一種很好地實(shí)現(xiàn)RPC的自動(dòng)化工具。Protobuf 編譯器會(huì)將.proto文件編譯生成對(duì)應(yīng)的數(shù)據(jù)結(jié)構(gòu)以對(duì)Protobuf數(shù)據(jù)進(jìn)行序列化、反序列化操作。
以最簡(jiǎn)單的一個(gè)對(duì)象Person（僅有三個(gè)屬性：姓名、年齡和住址）為例，用Java自帶的序列化工具與Protobuf來對(duì)比。使用Java自帶的序列化工具，經(jīng)過壓縮后的數(shù)據(jù)是181 B，如圖1所示。
而當(dāng)采用Protobuf時(shí)，如圖2所示，占用空間僅有20 B，而且實(shí)現(xiàn)簡(jiǎn)單，壓縮速度快，傳輸速度快，反序列化也快。可以很好地實(shí)現(xiàn)分布式高并發(fā)式的數(shù)據(jù)傳輸，大大降低了網(wǎng)絡(luò)傳輸負(fù)載。
壓縮person對(duì)象時(shí)間和大小對(duì)比如表1所示。
3簡(jiǎn)單實(shí)例實(shí)現(xiàn)
本文根據(jù)Protobuf的優(yōu)點(diǎn)在NIO的基礎(chǔ)上實(shí)現(xiàn)了一個(gè)分布式的高并發(fā)、高傳輸效率的項(xiàng)目。系統(tǒng)采用多個(gè)一級(jí)引擎來處理原始日志數(shù)據(jù)，讀取后進(jìn)行分段，分段后采用Hash映射到多個(gè)二級(jí)引擎（可以任意臺(tái)Hash映射）中進(jìn)行數(shù)據(jù)融合,融合后的數(shù)據(jù)再匯總到一臺(tái)服務(wù)器上，客戶端可以通過遠(yuǎn)程Web訪問這個(gè)服務(wù)器上的數(shù)據(jù)。其中一級(jí)引擎與二級(jí)引擎之間的數(shù)據(jù)傳輸就是使用的NIO與Protobuf相結(jié)合的技術(shù)，如圖3所示。
圖3分布式NIO結(jié)構(gòu)示意圖客戶端使用Protobuf對(duì)數(shù)據(jù)序列化壓縮發(fā)送。
ListrpcList = new ArrayList ();//實(shí)例化發(fā)送數(shù)據(jù)
for(HTTPAPPHost hah : list){
RPCHah rpchah = RPCHah.newBuilder()
.setCellid(hah.getCellid()).setAppType(hah
.getAppType()).build();
rpcList.add(rpchah);//將原始list轉(zhuǎn)化為RPCList完成
RPCReq req = RPCReq.newBuilder()
.addAllHahs(rpcList).build();//序列化壓縮完成
if(e2info.getDataQueue().offer(req.toByteArray())){//調(diào)用網(wǎng)絡(luò)模塊，將數(shù)據(jù)發(fā)送到二級(jí)引擎
NIOClientRunner.sendData(e2info); }//發(fā)送數(shù)據(jù)
服務(wù)器端采用NIO接收數(shù)據(jù)并使用Protobuf反序列化及處理。
Selector selector=Selector.open();//開啟選擇器
ServerSocketChannel ssc=
ServerSocketChannel.open();
ssc.configureBlocking(false);//配置為非阻塞模式
ssc.register(selector, SelectionKey.OP_WRITE);
while(isRunning){
selector.select(1);//阻塞延時(shí)1ns
Set set=selector.selectedKeys();
Iterator<SelectionKey> it=set.iterator();
while(it.hasNext()){
SelectionKey skey=it.next();
if(skey.isReadable()){//選擇讀數(shù)據(jù)通道
SocketChannel sct = skey.channel();
ByteBuffer tempBuf=
ByteBuffer.allocate(1);
String dataStr="";
while(!dataStr.endsWith("\\r\\n")){
sct.read(tempBuf);
dataStr +=new String (tempBuf.array());
tempBuf.clear();}//防止粘包
byte［］ data= dataStr.array();
recoverData2List.handlerData(engine1Info, data); }
下面服務(wù)器端把data數(shù)據(jù)反序列化。
List<RPCHah> pcList=request.getHahsList();
for(RPCHah rpchah : rpcList){
HTTPAPPHost hah = new HTTPAPPHost();
hah.setCellid(rpchah.getCellid());
hah.setAppType(rpchah.getAppType());
Global.getDataQueue().put(hah);//將反序列化的對(duì)象存儲(chǔ)到dataQueue中，反序列化完成
}
表2是一級(jí)引擎向二級(jí)引擎發(fā)送17 980條實(shí)例HTTPAPPHost對(duì)象數(shù)據(jù)與Java自帶序列化的數(shù)據(jù)傳輸這些數(shù)據(jù)量的效率對(duì)比。
本文在傳統(tǒng)NIO的基礎(chǔ)上結(jié)合了Proto Buffer，使得壓縮后的數(shù)據(jù)量大致是原來的1/9，壓縮時(shí)間上大致是原來表2實(shí)際環(huán)境序列化rpcList對(duì)象
時(shí)間和大小對(duì)比ProtobufSerializable序列化時(shí)間/ns7466 687反序列化時(shí)間/ns95141 083數(shù)據(jù)大小/B2 084 93918 707 213的1/8，反序列化時(shí)間大致是原來的1/40，極大地提高了傳輸?shù)男剩档土司W(wǎng)絡(luò)負(fù)載［4］。
4結(jié)論
本文在傳統(tǒng)NIO的基礎(chǔ)上應(yīng)用Protobuf后，能夠使得分布式高并發(fā)下性能極大提升，網(wǎng)絡(luò)負(fù)載大大減小，優(yōu)化性能明顯，尤其在以Map Reduce［5］為核心技術(shù)的大數(shù)據(jù)處理應(yīng)用中性能更為突出。
參考文獻(xiàn)
［1］ GOETZ B,PEIERLS T，BIOCH J，等.Java并發(fā)編程實(shí)戰(zhàn)［M］.童云蘭，譯.北京:機(jī)械工業(yè)出版社，2012.
［2］ 李林鋒.Netty權(quán)威指南［M］.北京:電子工業(yè)出版社，2014.
［3］ 程超,楊風(fēng)召.基于Java非阻塞I/O開發(fā)高性能網(wǎng)絡(luò)應(yīng)用程序［J］.電子工程師，2006，32(10):7173.
［4］ 徐忠勝,沈蘇彬.一種云計(jì)算資源的多目標(biāo)優(yōu)化的調(diào)度方法［J］.微型機(jī)與應(yīng)用, 2015, 34(13):1720.
［5］ 元二菊,郭進(jìn)偉,皮建勇，等.基于MapReduce的序列規(guī)則在推薦系統(tǒng)中的研究［J］.微型機(jī)與應(yīng)用,2014,33(6):6870,73.