郭金磊，張玉生，胡愛(ài)蘭

　?。ㄈA北計(jì)算機(jī)系統(tǒng)工程研究所，北京 100083）

　　摘要：隨著大數(shù)據(jù)技術(shù)的發(fā)展，多線程高并發(fā)等技術(shù)已經(jīng)越來(lái)越成為大數(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)越來(lái)越成為大數(shù)據(jù)處理中的關(guān)鍵技術(shù)，同一個(gè)節(jié)點(diǎn)中的不同線程和不同節(jié)點(diǎn)的線程間的通信越來(lái)越密切。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è)讀寫(xiě)事件，讀數(shù)據(jù)時(shí)，當(dāng)選擇器發(fā)現(xiàn)該通道準(zhǔn)備讀完成，通道直接將數(shù)據(jù)從底層網(wǎng)卡隊(duì)列讀進(jìn)緩沖區(qū)。寫(xiě)數(shù)據(jù)時(shí)，當(dāng)選擇器發(fā)現(xiàn)該通道準(zhǔn)備寫(xiě)完成，通道將數(shù)據(jù)寫(xiě)進(jìn)緩沖區(qū)。通道可以實(shí)現(xiàn)在緩沖區(qū)中對(duì)每個(gè)字節(jié)類似于指針對(duì)數(shù)據(jù)操作，可以來(lái)回移動(dòng)讀取數(shù)據(jù)。選擇器可以用一個(gè)單獨(dú)的線程同時(shí)監(jiān)聽(tīng)管理多個(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ǔ)言無(wú)關(guān)、平臺(tái)無(wú)關(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)程過(guò)程調(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來(lái)對(duì)比。使用Java自帶的序列化工具，經(jīng)過(guò)壓縮后的數(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í)引擎來(lái)處理原始日志數(shù)據(jù)，讀取后進(jìn)行分段，分段后采用Hash映射到多個(gè)二級(jí)引擎（可以任意臺(tái)Hash映射）中進(jìn)行數(shù)據(jù)融合,融合后的數(shù)據(jù)再匯總到一臺(tái)服務(wù)器上，客戶端可以通過(guò)遠(yuǎn)程Web訪問(wèn)這個(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();//開(kāi)啟選擇器

　　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ù)量大致是原來(lái)的1/9，壓縮時(shí)間上大致是原來(lái)表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í)間大致是原來(lái)的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.

　?。?］ 李林鋒.Netty權(quán)威指南［M］.北京:電子工業(yè)出版社，2014.

　　［3］ 程超,楊風(fēng)召.基于Java非阻塞I/O開(kāi)發(fā)高性能網(wǎng)絡(luò)應(yīng)用程序［J］.電子工程師，2006，32(10):7173.

　?。?］ 徐忠勝,沈蘇彬.一種云計(jì)算資源的多目標(biāo)優(yōu)化的調(diào)度方法［J］.微型機(jī)與應(yīng)用, 2015, 34(13):1720.

　?。?］ 元二菊,郭進(jìn)偉,皮建勇，等.基于MapReduce的序列規(guī)則在推薦系統(tǒng)中的研究［J］.微型機(jī)與應(yīng)用,2014,33(6):6870,73.