在网络中使用IO流进行数据收发

在Java当中，所有对外设的操作都通过IO流来实现，不管是从磁盘中读取或写入文件，或者是从网络环境中接收或发送数据。IO流的基类有两个InputSstream和OutputStream，它们实现IO最基本的、无数据缓冲的、按节节流进行读写的操作功能。但是在实际的处理当中，为了数据读写的方便或提高读写的效率，往往会用到它们的子类，比如带缓冲区的类BufferedInputStream、BufferedOutputStream，DataInputStream、DataOutputStream等等。

现在，我要来模拟一个常见的网络模型：客户端/服务器模式，来了解网络IO流的具体操作。首先我定义数据包的格式分别为消息头和消息体，其中消息头由三部分组成：消息包的总长度、命令字、序列号，各为一个整形，消息体为一个Byte数组，由若干字节组成，长度由消息包的总长度减去12个消息头的长度得来，消息体当中有一部分是消息内容，这个具体的组成不再详述，消息包的定义不是我们关注的重点，我们着重来看一下数据发送、接收的实现及效率如何。

首先来写一个服务端的程序来接收数据，写一个客户端程序来发送数据，总共发送1万条数据进行测试，服务器端代码如下：

importjava.io.IOException;
importjava.net.ServerSocket;
importjava.net.Socket;

importcom.gftech.cmpp.bean.CmppBody;
importcom.gftech.cmpp.bean.CmppDeliver;
importcom.gftech.cmpp.bean.CmppHead;
importcom.gftech.cmpp.bean.CmppPack;
importcom.gftech.cmpp.bean.ICmppCmdID;
importcom.gftech.smp.SmpDeliverPack;

/***//**
*测试读取速度：使用Buffer和不使用Buffer的情况
*
*@authorsinboy
*@since2007.3.30
*
*/
publicclassServerDemo...{

publicstaticvoidmain(String[]args)...{
intlistenPort=2000;
try...{
Socketclient=null;
ServerSocketss=newServerSocket(listenPort);
System.out.println("侦听"+listenPort+"端口，等待客户端的连接...");
while(true)...{
client=ss.accept();
System.out.println("接收到通信平台或客服系统的连接"+client.toString());

read(client);
}
}catch(IOExceptione)...{
e.printStackTrace();
}
}

publicstaticvoidread(Socketsock)...{
if(sock!=null)
try...{
while(true)...{
CmppPackcp=CmppCommu.receive(sock);
if(cp!=null)...{
CmppHeadhead=cp.getHead();
CmppBodybody=cp.getBody();
if(head!=null)...{
switch(head.getCmdID())...{
caseICmppCmdID.CMPP_DELIVER:
CmppDelivercd=newCmppDeliver(body.getBody());
if(cd!=null)...{
SmpDeliverPackpack=newSmpDeliverPack(cd);
Stringcontent="cmppdeliver:"+pack.getSrcAddr()+""+pack.getDestAddr()+""+pack.getContent()+""
+pack.getLinkID();
System.out.println(content);

}
}
}
}
}
}catch(IOExceptione)...{
e.printStackTrace();
}
}

}

publicvoidsend(byte[]b)throwsIOException...{
if(sock==null)
thrownewIOException();
if(b!=null)...{
try...{
BufferedOutputStreamos=newBufferedOutputStream(sock.getOutputStream());
if(b!=null)...{
os.write(b);
os.flush();
}
}catch(IOExceptione)...{
throwe;
}
}
}

客户端代码如下：

publicclassClientDemo...{

publicstaticvoidmain(String[]args)...{
try...{
Socketclient=newSocket("192.168.10.8",2000);
write(client);
}catch(UnknownHostExceptione)...{
e.printStackTrace();
}catch(IOExceptione)...{
e.printStackTrace();
}

}

publicstaticvoidwrite(Socketclient)...{
try...{
ArrayList<CmppPack>cpList=newArrayList<CmppPack>();
for(inti=0;i<10000;i++)...{
SmpDeliverPackpack=newSmpDeliverPack("13612345678","01234",""+(i+1));
CmppDelivercd=pack.toCmppDeliver();
if(cd!=null)...{
CmppPackcp=newCmppPack(newCmppHead(12+cd.getBody().length,ICmppCmdID.CMPP_DELIVER,100),cd);
cpList.add(cp);
}
}

for(inti=0;i<cpList.size();i++)...{
CmppCommu.send(client,cpList.get(i));
System.out.println(i);
}
Thread.sleep(10000);
}catch(IOExceptione)...{
e.printStackTrace();
}
catch(InterruptedExceptione)...{

}
}
}

用到的发送和接入方法如下：

publicclassCmppCommu...{

publicstaticvoidsend(Socketsock,CmppPackpack)throwsIOException...{
if(sock!=null&&pack!=null)...{
try...{
byte[]b=pack.getBytes();
BufferedOutputStreamos=newBufferedOutputStream(sock.getOutputStream());
if(b!=null)...{
os.write(b);
os.flush();
}
}catch(IOExceptione)...{
throwe;
}
}

}
publicstaticCmppPackreceive(Socketsock)throwsIOException...{
finalintMAX_LEN=10000;
CmppPackpack=null;

if(sock==null)
thrownewIOException();
try...{
BufferedInputStreambis=newBufferedInputStream(sock.getInputStream());
DataInputStreamin=newDataInputStream(bis);

intlen=in.readInt();//读取消息头
if(len>=12&&len<MAX_LEN)...{
intcmd=in.readInt();
intseq=in.readInt();

intbodyLen=len-12;
byte[]msg=newbyte[bodyLen];
in.read(msg);//读取消息体

CmppHeadhead=newCmppHead(len,cmd,seq);
CmppBodybody=newCmppBody(msg);
pack=newCmppPack(head,body);
}
}catch(SocketTimeoutExceptione)...{
//logger.warn("timeout");
}catch(IOExceptione)...{
throwe;
}

returnpack;
}
}

使用Eclipse TPTP对程序的执行进行监控，结果如下：

<shapetype id="_x0000_t75" stroked="f" filled="f" path="m@4@5l@4@11@9@11@9@5xe" o:preferrelative="t" o:spt="75" coordsize="21600,21600"><a href="http://p.blog.csdn.net/images/p_blog_csdn_net/sinboy/293422/o_1.JPG"><img alt="" src="http://p.blog.csdn.net/images/p_blog_csdn_net/sinboy/293422/o_1.JPG"></a></shapetype>

<shapetype stroked="f" filled="f" path="m@4@5l@4@11@9@11@9@5xe" o:preferrelative="t" o:spt="75" coordsize="21600,21600"><span style="FONT-SIZE: 10.5pt; FONT-FAMILY: 宋体; mso-ascii-font-family: 'Times New Roman'; mso-hansi-font-family: 'Times New Roman'; mso-bidi-font-family: 'Times New Roman'; mso-ansi-language: EN-US; mso-fareast-language: ZH-CN; mso-bidi-language: AR-SA; mso-bidi-font-size: 12.0pt; mso-font-kerning: 1.0pt">如上图所示，服务器端程序在接收数据时总共花了</span><span lang="EN-US" style="FONT-SIZE: 10.5pt; FONT-FAMILY: 'Times New Roman'; mso-ansi-language: EN-US; mso-fareast-language: ZH-CN; mso-bidi-language: AR-SA; mso-fareast-font-family: 宋体; mso-bidi-font-size: 12.0pt; mso-font-kerning: 1.0pt">21</span><span style="FONT-SIZE: 10.5pt; FONT-FAMILY: 宋体; mso-ascii-font-family: 'Times New Roman'; mso-hansi-font-family: 'Times New Roman'; mso-bidi-font-family: 'Times New Roman'; mso-ansi-language: EN-US; mso-fareast-language: ZH-CN; mso-bidi-language: AR-SA; mso-bidi-font-size: 12.0pt; mso-font-kerning: 1.0pt">秒多。因为从理论上讲把输入流用</span><span lang="EN-US" style="FONT-SIZE: 10.5pt; FONT-FAMILY: 'Times New Roman'; mso-ansi-language: EN-US; mso-fareast-language: ZH-CN; mso-bidi-language: AR-SA; mso-fareast-font-family: 宋体; mso-bidi-font-size: 12.0pt; mso-font-kerning: 1.0pt">Buffer</span><span style="FONT-SIZE: 10.5pt; FONT-FAMILY: 宋体; mso-ascii-font-family: 'Times New Roman'; mso-hansi-font-family: 'Times New Roman'; mso-bidi-font-family: 'Times New Roman'; mso-ansi-language: EN-US; mso-fareast-language: ZH-CN; mso-bidi-language: AR-SA; mso-bidi-font-size: 12.0pt; mso-font-kerning: 1.0pt">包装一下接收的速度会更快一些，下面我们对它进行验证，把接收程序略做改动，增加一句</span><span lang="EN-US" style="FONT-SIZE: 10.5pt; FONT-FAMILY: 'Times New Roman'; mso-ansi-language: EN-US; mso-fareast-language: ZH-CN; mso-bidi-language: AR-SA; mso-fareast-font-family: 宋体; mso-bidi-font-size: 12.0pt; mso-font-kerning: 1.0pt">Buffer</span><span style="FONT-SIZE: 10.5pt; FONT-FAMILY: 宋体; mso-ascii-font-family: 'Times New Roman'; mso-hansi-font-family: 'Times New Roman'; mso-bidi-font-family: 'Times New Roman'; mso-ansi-language: EN-US; mso-fareast-language: ZH-CN; mso-bidi-language: AR-SA; mso-bidi-font-size: 12.0pt; mso-font-kerning: 1.0pt">包装：</span></shapetype>

BufferedInputStreambis=newBufferedInputStream(sock.getInputStream());
DataInputStreamin=newDataInputStream(bis);

再次运行，发现一个很奇怪的问题，接收时有数据丢失的情况，正常情况下应该是从1到10000：

cmppdeliver:1361234567801234116240905710000010001
cmppdeliver:1361234567801234216241005710000010001
cmppdeliver:136123456780123426116241005710000010001
cmppdeliver:136123456780123426216241005710000010001
cmppdeliver:136123456780123426316241005710000010001
cmppdeliver:136123456780123426416241005710000010001
......

为何用Buffer进行包装之后会有数据丢失而没有包装时接收正常呢？我们仔细研究一下接收的源代码发现，在每次对数据进行接收时，都会在一个Socket上重新建一个输入流，因为BufferedInputStream在创建时会自动建立一个缓冲区用于整块数据的读取，系统默认为8192,也就是说当你读第一个包的时候，系统本身其实已经读取的8192个字节的数据，而你的包大小可能是只有100个字节，在你循环过来再次读第二个数据包时，等于又重新创建了一个输入流，把前面那个丢掉了，而它原先已经预先读取的数据也随之丢失。而不用Buffer包装的输入流因为每次都是从实际的IO中读取数据所以不存在数据丢失的情况。

明白了这一点，那我们可以把创建输入流放到每次读取的循环之外，修改读取方法如下：

publicstaticCmppPackreceive(DataInputStreamin)throwsIOException...{
finalintMAX_LEN=10000;
CmppPackpack=null;
if(in==null)
thrownewIOException();
try...{
intlen=in.readInt();//读取消息头
if(len>=12&&len<MAX_LEN)...{
intcmd=in.readInt();
intseq=in.readInt();

intbodyLen=len-12;
byte[]msg=newbyte[bodyLen];
in.read(msg);//读取消息体

CmppHeadhead=newCmppHead(len,cmd,seq);
CmppBodybody=newCmppBody(msg);
pack=newCmppPack(head,body);
}
}catch(SocketTimeoutExceptione)...{
//logger.warn("timeout");
}catch(IOExceptione)...{
throwe;
}

returnpack;
}

服务器端的代码也作相应调整,把创建输入流的过程放在循环之外，保证每次读取数据包都是使用同一个输入流：

BufferedInputStreambis=newBufferedInputStream(sock.getInputStream());
DataInputStreamin=newDataInputStream(bis);
while(true)...{
CmppPackcp=CmppCommu.receive(in);
...

再次进行测试，结果如下：

从上图可以看出来，效率提高了很多，这里提高的原因有两点：一个使用了Buffer进行输入流的包装，二是每次读取数据包时创建新的输入流的过程放到了循环之外，减少的资源的开销。但纯粹的使用Buffer进行包装，效果究竟能提高多少呢？我们再次使用非Buffer的读取方式，但把创建输入流的过程放在循环之外，接收方法不变，服务端代码如下：

DataInputStreamin=newDataInputStream(sock.getInputStream());
while(true)...{
CmppPackcp=CmppCommu.receive(in);
...

从上图可以看出，非Buffer包装的输入流，时间增加了约3秒钟，效率下降了50%左右。

再发客户端的程序做一个改进，把创建输出流这一步也提到循环之外：

BufferedOutputStreamout=newBufferedOutputStream(client.getOutputStream());
for(inti=0;i<cpList.size();i++)...{
CmppCommu.send(out,cpList.get(i));
System.out.println(i);
}

提高虽然不明显，但也有提高，结果如下:

从上述的测试过程中可以明白两点：

1、使用Buffer进行输入输出流的包装可以减少程序对IO的实际操作次数，提高效率

2、尽量把不变的东西提到循环之外，象创建输入流这一步，极大地影响了程序的执行效率