Java数据报之失序和丢包

欢迎进入Java社区论坛，与200万技术人员互动交流 >>进入

　　习惯了TCP编程，认为UDP可以包办这些问题是错误的。一个UDP应用程序要承担可靠性方面的全部工作，包括报文的丢失、重复、时延、乱序以及连接失效等问题。

　　通常我们在可靠性好，传输时延小的局域网上开发测试，一些问题不容易暴露，但在大型互联网上却会出现错误。

　　UDP协议把递送的可靠性责任推到了上层即应用层，下面简单编写了几个类来专门处理两个问题：乱序和丢包。

　　四个类：DataPacket 类，PacketHeader类，PacketBody类 ，DataEntry类，位于同一个文件DataPacket .java中。

　　DataPacket 类相当于一个门面模式，提供给外部使用，通信数据也在这个类中处理。

package com.skysoft.pcks;

import java.io.*;import java.net.*;import java.util.*;

public class DataPacket { InputStream is; OutputStream os; PacketHeader header; PacketBody body; ArrayList al; public static final int DataSwapSize = 64532;

/** * 在接收数据报使用 */ public DataPacket() { header = new PacketHeader(); body = new PacketBody(); al = new ArrayList(); } /** * 在发送数据报时使用,它调用报文分割操作. * @param file String 硬盘文件 */ public DataPacket(String file) { this(); try { is = new FileInputStream(file); header.CalcHeaderInfo(is.available()); this.madeBody(); is.close(); //this.Gereratedata(); } catch (FileNotFoundException ex) { ex.printStackTrace(); } catch (IOException ex1) { ex1.printStackTrace(); } } /** * 在发送数据报时使用,它调用报文分割操作. * @param url URL url地址 */ public DataPacket(URL url) { this(); try { //is = url.openStream(); URLConnection conn=url.openConnection(); is=conn.getInputStream(); int total=conn.getContentLength(); header.CalcHeaderInfo(total); this.madeBody(); //System.out.println(total+”:”+total); is.close(); } catch (IOException ex) { ex.printStackTrace(); } } /** * 为发送构造分组,使用PackageHeader处理了报头格式,并为分组编序号. */ private void madeBody() { al.clear(); byte[] buffer; DataEntry de; for (int i = 0; i < header.fragmentcounter; i++) { try { ByteArrayOutputStream bos = new ByteArrayOutputStream(); //is.skip(i * body.BODY_BUFFER_SIZE); header.ArrageSort(i); de = new DataEntry(PacketBody.BODY_BUFFER_SIZE); de.setSn(i); de.setStreamsize(header.getStreamsize()); de.setFragmentcounter(header.getFragmentcounter()); if (header.isWTailFragment(i)) { buffer = new byte[header.getMinfragment()]; is.read(buffer, 0, buffer.length); header.setActByteSize(header.getMinfragment()); de.setActByteSize(header.getMinfragment()); } else { buffer = new byte[body.BODY_BUFFER_SIZE]; is.read(buffer, 0, buffer.length); } //System.out.println(“length——-“+i+” “+body.getBody().length+” “+header.getMinfragment()); body.setBody(buffer); //System.out.println(“length:” + i + ” ” + header.toString()); bos.write(header.getByte(), 0, header.HEADER_BUFFER_SIZE); bos.write(body.getBody(), 0, body.getBody().length); de.setBytes(bos.toByteArray()); al.add(de); } catch (IOException ex) { ex.printStackTrace(); } } } /** * 为发送构造分组,没有考虑报头格式,也没有为分组编序号. */ private void madeBody1() { al.clear(); for (int i = 0; i < header.fragmentcounter; i++) { try { if (header.isWTailFragment(i)) is.read(body.getBody(), i * body.BODY_BUFFER_SIZE, header.getMinfragment()); else is.read(body.getBody(), i * body.BODY_BUFFER_SIZE, body.BODY_BUFFER_SIZE); ByteArrayOutputStream bos = new ByteArrayOutputStream(); bos.write(header.getByte(), 0, header.HEADER_BUFFER_SIZE); bos.write(body.getBody(), header.HEADER_BUFFER_SIZE, body.getBody().length); al.add(bos); } catch (IOException ex) { ex.printStackTrace(); } } } /** * 在接收到报文后,对此报文执行组装,并处理报文丢失和乱序情况. * @param b1 byte[] */ public void Add(byte[] b1) { byte[] buffer = (byte[]) b1.clone(); handlerText(buffer); DataEntry de = new DataEntry(buffer, header.getActByteSize()); de.setSn(header.getSn()); de.setStreamsize(header.getStreamsize()); de.setFragmentcounter(header.getFragmentcounter()); al.add(de); } private void handlerText(byte[] buffer) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); baos.write(buffer, 0, header.HEADER_BUFFER_SIZE); byte[] b=new byte[header.HEADER_BUFFER_SIZE]; System.arraycopy(buffer,0,b,0,b.length); ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray()); InputStreamReader isr = new InputStreamReader(bais); BufferedReader br = new BufferedReader(isr); try { header = new PacketHeader(br.readLine()); } catch (Exception ex) { ex.printStackTrace(); } } private String calFileSize(int size) { return size / 1024 + “K”; }

public ArrayList getDataPackets() { return al; }/*** 是否接收完毕,通过序号是否等于最大段数来判断,这也许有问题,比如,正好是最后一个段丢失了,这样* 这个包整个就丢失了.* @return*/ public boolean isFull() { return this.header.getSn() == this.header.getFragmentcounter() – 1 ? true : false; }/*** 判断是否只有一个段.* @return*/ public boolean isZero() { return this.header.getSn() == 0 ? true : false; }/*** 该函数执行报文组装,不考虑丢失的报文.* @return*/ private ByteArrayOutputStream fetchDataPackets() { ByteArrayOutputStream bos = new ByteArrayOutputStream(); byte[] buffer = null; DataEntry de; for (int i = 0; i < al.size(); i++) { try { de = this.getSnData(i); buffer = de.getByte(); if (header.getStreamsize() == de.getStreamsize()) { bos.write(de.getByte(), header.HEADER_BUFFER_SIZE, de.getActByteSize()); System.out.println(de.toString() + ” — fetchDataPackets”); } } catch (Exception ex) { ex.printStackTrace(); } } return bos; }

/** * 该函数执行报文组装,对于丢失的报文,写入空报文. * @return ByteArrayOutputStream */ private ByteArrayOutputStream fetchDataPackets_sn() { ByteArrayOutputStream bos = new ByteArrayOutputStream(); byte[] buffer; DataEntry de; for (int i = 0; i < header.getFragmentcounter(); i++) { try { de = this.getSnData(i); if (de == null) { de = seachDeData(i); } buffer = de.getByte(); //System.out.println(de.getSn() + “:” + i); //handlerText(buffer); //bos.write(buffer, header.HEADER_BUFFER_SIZE, // buffer.length – header.HEADER_BUFFER_SIZE); if (header.getStreamsize() == de.getStreamsize()) { bos.write(de.getByte(), header.HEADER_BUFFER_SIZE, de.getActByteSize()); //System.out.println(de.toString()); } } catch (Exception ex) { ex.printStackTrace(); } } return bos; }

[1][2]

婚姻犹如一艘雕刻的船，看你怎样去欣赏它，又怎样驾驭它。