Java数据报之失序和丢包

　　习惯了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";
　　  }