基于UDP的可靠性高速传输（大文件）的实现（一）

TCP 协议和 UDP 协议是 TCP / I P协议栈中的两个最主要的传输层协议. 其中 TCP 提供可靠的、 有序的、 端到端的数据传输服务, 而 UDP则提供的是不可靠的、 不保证有序到达的、 端到端的数据传输服务. TCP 协议一般应用在数据传输量大或可靠性要求高的场合 (例如: 文件传输 ); UDP协议则一般是应用在数据传输量不大且可靠性要求不高的场合.
然而, 在实践中却会经常碰到这样的情况, 需要传输的数据不连续或数据量不确定, 但可靠性却要求很高. 例如这个项目: 一银行终端管理软件, 网内终端数量最少时 2 台, 最多时可达 150 台, 传输的数据具
有实时性, 最少时只需传输几个字节 ( 例如: 远程关机、 重启等简单指令 ) , 最多可达几 KB ( 例如: 远程设置
终端属性 ), 要求数据在传输过程中安全可靠. 在该系统中到底该采用哪种协议作为传输层协议呢? 采用
TCP 协议则会因为开辟的侦听线程过多而直接导致服务器的效率下降, 特别是该系统中的终端机根本就
没有多余的资源可以浪费; 如果采用 UDP协议, 虽然可以解决资源浪费问题, 但该协议是一种不可靠的没有差错处理的协议, 而该系统必须保证数据传输过程中的安全可靠. 因此针对这种情况应该放弃采用 TCP协议作为传输层协议的方案, 而是采用 UDP协议作为传输协议通过增加差错处理、 拥塞控制等措施来提高数据传输的安全性与可靠性.

以上文字摘自--周锦才的可靠 UDP 协议的设计思路与实现方法论文。

对于UDP的可靠性传输已经研究了几天，今天实现了大文件的可靠性传输的初步，在接下来的几天里，会把它慢慢实现。

好的，下面入正题。

我们要先定义可靠性UDP的数据报，这相当于一种协议，只不过是由我们自己定义的。

先说说整体思路：

对于大文件的UDP传输，在发送端，首先要发送一个开始发送数据包，在接受端接受到这个开始发送数据包以后，要发送一个开始发送确认数据包，当发送端在一定时间内（这个时间根据网络情况自己调整）收到了确认数据包以后，开始将一个大文件分成若干个组，每个组包含n个数据包，每个数据包长m。然后根据上面发送开始发送数据包相似的原理，进行发送（这一部分很复杂，具体实现我也还在研究，在后文中会详细解析，敬请关注）。

当发送端发送完到最后一个数据包时，向接收端发送一个发送完成包，在收到接收端发来的反馈后，结束发送。

在以下代码中，实现了开始数据包的收发实现。

发送端

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.Random;

import javax.xml.crypto.Data;


public class UdpSend {
	private int myPort;// 我方端口
	private int otherPort;// 对方端口
	private String otherIP;// 对方IP
	private DatagramSocket dataSocket;
	private final int GROUP_DATA = 256;// 一个组拥有的数据包个数
	private final int DATA_LEN = 1024;// 一个数据包的子节长度
	private final int GROUP_LEN = GROUP_DATA * DATA_LEN;// 一个组的字节长度
	/* 用于分组的数据 */
	private final int START_TIP[] = { 8, 0, 4 }; // 开始传输包(8) 标识(4)|次随机数(4)
	private final int STARTACK_TIP[] = { 8, 0, 4 }; // 开始确认包(8) 标识(4)|次随机数(4)
	private final int END_TIP[] = { 8, 0, 4 };// 结束传输包
	private final int ENDACK_TIP[] = { 8, 0, 4 };// 结束确认包
	// 分组标识
	private final String START_IDEN = "[]01";// 开始传输标识
	private final String STARTACK_IDEN = "[]02";// 开始确认标识
	private final String END_IDEN = "[]03";// 结束传输标识
	private final String ENDACK_IDEN = "[]04";// 结束确认标识

	/**
	 * 构造方法设置端口 Ip
	 * 
	 * @param myPort
	 * @param otherPort
	 * @param otherIP
	 */
	public UdpSend(int myPort, int otherPort, String otherIP) {
		this.myPort = myPort;
		this.otherPort = otherPort;
		this.otherIP = otherIP;
		try {
			dataSocket = new DatagramSocket(myPort);
			dataSocket.setReceiveBufferSize(GROUP_LEN);
		} catch (SocketException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 将数据拆成若干个组 ,每个组有GROUP_DATA(256)个数据包，每个数据包为DATA_LEN(1024)个字节长
	 * 
	 * @param totBuf
	 * @param totLen
	 */
	public void sendAll(byte[] totBuf, int totLen) {
		int groupNum = totLen / GROUP_LEN + (totLen % GROUP_LEN == 0 ? 0 : 1);// 分组个数
		byte tempBuf[] = new byte[GROUP_LEN];
		int numRandom = (new Random()).nextInt();
		// 发送开始传输包
		System.out.println("开始发送开始传输包");
		sendStart(numRandom);
		System.out.println("开始传输包发送完毕");
		// 拆分 
		/*
		for (int i = 0; i < groupNum; i++) {
			int groupLen;// 当前分组长度
			if (i == groupNum - 1) {
				groupLen = totLen % GROUP_LEN;
			} else {
				groupLen = GROUP_LEN;
			}
			for (int j = 0; j < groupLen; j++) {
				tempBuf[j] = totBuf[i * GROUP_LEN + j];
			}
			// 发送一个分组
			sendGroup(tempBuf, numRandom);
		}
		sendEnd(numRandom);
		*/
	}

	/**
	 * 发送开始传输包
	 * 
	 * @param numRandom
	 */
	void sendStart(int numRandom) {
		DatagramPacket sendPacket;
		DatagramPacket recePacket;
		byte[] sendBuf = new byte[START_TIP[0]];
		byte[] receBuf = new byte[STARTACK_TIP[0]];

		// 填充开始传输包
		fillStartData(sendBuf, numRandom);

		try {
			sendPacket = new DatagramPacket(sendBuf, sendBuf.length,
					InetAddress.getByName(otherIP), otherPort);
			recePacket = new DatagramPacket(receBuf, receBuf.length);
			// 设置接受超时时间
			dataSocket.setSoTimeout(1);
			boolean canSend = true;
			while (true) {
				try {
					if (canSend) {
						dataSocket.send(sendPacket);
					}
					dataSocket.receive(recePacket);
					// 只要收到了数据包 就不再发送开始数据包
					canSend = false;
					// 判断收到的数据包是否是开始确认包
					// 判断标识
					String checkStr = new String(receBuf, STARTACK_TIP[1],
							STARTACK_TIP[2]);
					if (!checkStr.equals(STARTACK_IDEN)) {
						continue;
					}
					// 判断随机数
					int checkNum = ByteChange.byteToInt(receBuf,
							STARTACK_TIP[2]);
					if (checkNum != numRandom) {
						continue;
					}
					// 能执行到这里说明接受到的是开始确认包 退出循环
					break;
				} catch (Exception e) {
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	

	/**
	 * 填充开始传输包
	 * 
	 * @param buf
	 *            开始传输包
	 * @param numRandom
	 *            次随机数
	 */
	void fillStartData(byte[] buf, int numRandom) {
		// 开始传输包组成 标识 次随机数
		// 填充标识
		ByteChange.stringToByte(START_IDEN, buf, START_TIP[1]);
		// 填充随机数
		ByteChange.intToByte(numRandom, buf, START_TIP[2]);
	}



	public static void main(String[] args) {
		UdpSend udpSend = new UdpSend(9090, 9191, "127.0.0.1");
		byte totBuf[] = "hello world!".getBytes();
		udpSend.sendAll(totBuf, totBuf.length);

	}

}

 

接受端：

 

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.UnknownHostException;



public class UdpRece {
	private int myPort;// 我方端口
	private int otherPort;// 对方端口
	private String otherIP;// 对方IP
	private DatagramSocket dataSocket;
	private final int GROUP_DATA = 256;// 一个组拥有的数据包个数
	private final int DATA_LEN = 1024;// 一个数据包的子节长度
	private final int GROUP_LEN = GROUP_DATA * DATA_LEN;// 一个组的字节长度
	/* 用于分组的数据 */
	private final int START_TIP[] = { 8, 0, 4 }; // 开始传输包(8) 标识(4)|次随机数(4)
	private final int STARTACK_TIP[] = { 8, 0, 4 }; // 开始确认包(8) 标识(4)|次随机数(4)
	private final int END_TIP[] = { 8, 0, 4 };// 结束传输包
	private final int ENDACK_TIP[] = { 8, 0, 4 };//结束确认包
	// 分组标识
	private final String START_IDEN = "[]01";// 开始传输标识
	private final String STARTACK_IDEN = "[]02";// 开始确认标识
	private final String END_IDEN = "[]03";// 结束传输标识
	private final String ENDACK_IDEN = "[]04";// 结束确认标识

	/**
	 * 构造方法设置端口 Ip
	 * 
	 * @param myPort
	 * @param otherPort
	 * @param otherIP
	 */
	public UdpRece(int myPort, int otherPort, String otherIP) {
		this.myPort = myPort;
		this.otherPort = otherPort;
		this.otherIP = otherIP;
		try {
			dataSocket = new DatagramSocket(myPort);
			dataSocket.setReceiveBufferSize(GROUP_LEN);
		} catch (SocketException e) {
			e.printStackTrace();
		}
	}

	void receAll() {
		int numRandom = receStart();
		
	}

	/**
	 * 用于收取开始传输包,收取成功时会自动发送开始确认包
	 * 
	 * @return 收到的次随机数
	 */
	int receStart() {
		DatagramPacket recePacket;
		byte[] receBuf = new byte[START_TIP[0]];// 用于接受开始传输包
		try {
			recePacket = new DatagramPacket(receBuf, receBuf.length);
			// 接收超时时间为无穷大
			dataSocket.setSoTimeout(0);
			while (true) {
				dataSocket.receive(recePacket);
				// 检查标识
				String checkStr = new String(receBuf, START_TIP[1],
						START_TIP[2]);
				if (!checkStr.equals(START_IDEN)) {
					continue;
				}
				System.out.println("收到开始传输包");
				int numRandom = ByteChange.byteToInt(receBuf, START_TIP[2]);
				sendStartAck(numRandom);
				return numRandom;
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return 0;
	}

	
	/**
	 * 用于发送开始确认包
	 * 
	 * @param numRandom
	 */
	void sendStartAck(int numRandom) {
		DatagramPacket sendPacket;
		byte[] sendBuf = new byte[STARTACK_TIP[0]];// 用于发送开始确认包
		try {
			sendPacket = new DatagramPacket(sendBuf, sendBuf.length,
					InetAddress.getByName(otherIP), otherPort);
			fillStartAckData(sendBuf, numRandom);
			dataSocket.send(sendPacket);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 填充开始确认包
	 * 
	 * @param buf
	 *            开始确认包
	 * @param numRandom
	 *            次随机数
	 */
	void fillStartAckData(byte[] buf, int numRandom) {
		// 开始传输包组成 标识 次随机数
		// 填充标识
		ByteChange.stringToByte(STARTACK_IDEN, buf, STARTACK_TIP[1]);
		// 填充随机数
		ByteChange.intToByte(numRandom, buf, STARTACK_TIP[2]);
	}
	

	public static void main(String[] args) {
		UdpRece udpRece = new UdpRece(9191, 9090, "127.0.0.1");
		udpRece.receAll();
	}
}

 

数据转换：

public class ByteChange {
	
	public static void stringToByte(String s, byte[] buf, int start) {
		byte tByteArr[] = s.getBytes();
		for (int i = 0; i < tByteArr.length; i++) {
			buf[start + i] = tByteArr[i];
		}
	}
	
	public static void intToByte(int num,byte[] buf ,int start)
	{
		for (int i = start; i < start +4;i++){
	            buf[i]=(byte)num;
	            num>>>=8;
	        }
	}
	 //从byte数组中提取int,使用小端模式
    public static int byteToInt(byte b[],int start){
        int num=0;
        for(int i=3+start;i>=start;i--){
            num=num<<8;
            num+=(0xff&(int)b[i]);
        }
        return num;
    }
}