郭金磊，张玉生，胡爱兰

　　（华北计算机系统工程研究所，北京 100083）

摘要：随着大数据技术的发展，多线程高并发等技术已经越来越成为大数据处理中的关键技术。非阻塞式I/O(new I/O,NIO)技术作为一种分布式高并发技术被广泛应用，但对于大数据量的通信往往需要很多的时间才能完成。Google提出的Protocol Buffer序列化压缩技术相对于传统序列化效率高、时间短、使用简单。文章将传统NIO技术与Protocol Buffer相结合，在分布式系统不同节点通信中，极大地降低了分布式系统的网络负载，大大节省了数据传输时间。

　关键词：NIO（new I/O）；Protocol Buffer ；分布式系统；序列化

0引言

　　随着大数据技术的发展，多线程高并发等技术已经越来越成为大数据处理中的关键技术，同一个节点中的不同线程和不同节点的线程间的通信越来越密切。Java NIO作为一种分布式数据传输技术在多线程高并发［1］的实际应用中扮演着至关重要的角色。为减小网络负载，加速分布式系统中网络通信，迫切需要一种高效率压缩序列化技术。

1研究现状

　　Java NIO的核心是Channel、Buffer 和 Selector。NIO基于通道（Channel）和缓冲区（Buffer）进行操作，通道先在选择器注册读写事件，读数据时，当选择器发现该通道准备读完成，通道直接将数据从底层网卡队列读进缓冲区。写数据时，当选择器发现该通道准备写完成，通道将数据写进缓冲区。通道可以实现在缓冲区中对每个字节类似于指针对数据操作，可以来回移动读取数据。选择器可以用一个单独的线程同时监听管理多个通道。

　　传统的NIO［2］都是使用Java自带的序列化形式对传输数据和对象进行序列化压缩。这种情况下，数据压缩率［3］较低，需要传输的对象数据流很大时，尤其在分布式系统中，容易造成网络拥堵。本文在传统NIO技术的基础上结合Google Protocol Buffer技术实现了数据对象的高效序列化压缩传输。

2Protocol Buffer优点

　　Google Protocol Buffer(简称Protobuf)是Google公司提出的混合语言数据标准，用于 RPC 系统和持续数据存储系统。同时也可用于通信协议、数据存储等领域的语言无关、平台无关、可扩展的序列化结构数据格式。目前提供了C++、Java、Python三种语言的API。Protobuf 具有很多优点：实现简单，压缩速度快，传输速度快，存储空间小。用Protobuf与Java自带的序列化工具实现的对象压缩相比，存储空间大了一个数量级，时间上快了一个数量级，尤其是可以自动生成远程过程调用协议（Remote Procedure Call Protocol, RPC）的数据结构,特别是service业务逻辑，是一种很好地实现RPC的自动化工具。Protobuf 编译器会将.proto文件编译生成对应的数据结构以对Protobuf数据进行序列化、反序列化操作。

　　以最简单的一个对象Person（仅有三个属性：姓名、年龄和住址）为例，用Java自带的序列化工具与Protobuf来对比。使用Java自带的序列化工具，经过压缩后的数据是181 B，如图1所示。　　

　　而当采用Protobuf时，如图2所示，占用空间仅有20 B，而且实现简单，压缩速度快，传输速度快，反序列化也快。可以很好地实现分布式高并发式的数据传输，大大降低了网络传输负载。

　　压缩person对象时间和大小对比如表1所示。

3简单实例实现

　　本文根据Protobuf的优点在NIO的基础上实现了一个分布式的高并发、高传输效率的项目。系统采用多个一级引擎来处理原始日志数据，读取后进行分段，分段后采用Hash映射到多个二级引擎（可以任意台Hash映射）中进行数据融合,融合后的数据再汇总到一台服务器上，客户端可以通过远程Web访问这个服务器上的数据。其中一级引擎与二级引擎之间的数据传输就是使用的NIO与Protobuf相结合的技术，如图3所示。

　　图3分布式NIO结构示意图客户端使用Protobuf对数据序列化压缩发送。

　　ListrpcList = new ArrayList ();//实例化发送数据

　　for(HTTPAPPHost hah : list){

　　RPCHah rpchah = RPCHah.newBuilder()

　　.setCellid(hah.getCellid()).setAppType(hah

　　.getAppType()).build();

　　rpcList.add(rpchah);//将原始list转化为RPCList完成

　　RPCReq req = RPCReq.newBuilder()

　　.addAllHahs(rpcList).build();//序列化压缩完成

　　if(e2info.getDataQueue().offer(req.toByteArray())){//调用网络模块，将数据发送到二级引擎

　　NIOClientRunner.sendData(e2info); }//发送数据

　　服务器端采用NIO接收数据并使用Protobuf反序列化及处理。

　　Selector selector=Selector.open();//开启选择器

　　ServerSocketChannel ssc=

　　ServerSocketChannel.open();

　　ssc.configureBlocking(false);//配置为非阻塞模式

　　ssc.register(selector, SelectionKey.OP_WRITE);

　　while(isRunning){

　　selector.select(1);//阻塞延时1ns

　　Set set=selector.selectedKeys();

　　Iterator it=set.iterator();

　　while(it.hasNext()){

　　SelectionKey skey=it.next();

　　if(skey.isReadable()){//选择读数据通道

　　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); }

　　下面服务器端把data数据反序列化。

　　List pcList=request.getHahsList();

　　for(RPCHah rpchah : rpcList){

　　HTTPAPPHost hah = new HTTPAPPHost();

　　hah.setCellid(rpchah.getCellid());

　　hah.setAppType(rpchah.getAppType());

　　Global.getDataQueue().put(hah);//将反序列化的对象存储到dataQueue中，反序列化完成

　　}

　　表2是一级引擎向二级引擎发送17 980条实例HTTPAPPHost对象数据与Java自带序列化的数据传输这些数据量的效率对比。

　　本文在传统NIO的基础上结合了Proto Buffer，使得压缩后的数据量大致是原来的1/9，压缩时间上大致是原来表2实际环境序列化rpcList对象

　　时间和大小对比ProtobufSerializable序列化时间/ns7466 687反序列化时间/ns95141 083数据大小/B2 084 93918 707 213的1/8，反序列化时间大致是原来的1/40，极大地提高了传输的效率，降低了网络负载［4］。

4结论

　　本文在传统NIO的基础上应用Protobuf后，能够使得分布式高并发下性能极大提升，网络负载大大减小，优化性能明显，尤其在以Map Reduce［5］为核心技术的大数据处理应用中性能更为突出。

　　参考文献

　　［1］ GOETZ B,PEIERLS T，BIOCH J，等.Java并发编程实战［M］.童云兰，译.北京:机械工业出版社，2012.

　　［2］ 李林锋.Netty权威指南［M］.北京:电子工业出版社，2014.

　　［3］ 程超,杨风召.基于Java非阻塞I/O开发高性能网络应用程序［J］.电子工程师，2006，32(10):7173.

　　［4］ 徐忠胜,沈苏彬.一种云计算资源的多目标优化的调度方法［J］.微型机与应用, 2015, 34(13):1720.

　　［5］ 元二菊,郭进伟,皮建勇，等.基于MapReduce的序列规则在推荐系统中的研究［J］.微型机与应用,2014,33(6):6870,73.