基于rsync同步算法的文件同步系统JAVA实现（四）——分析并发送不匹配数据块

作用：

利用不完整文件的校验块所构成的哈希表顺序查询完整文件的校验块，从而得出匹配情况，并返回不匹配数据块及其相应编号，当然还有一些控制信息。由于该过程比较复杂，所以最好设计一个数据报，便于客户端分析和重组文件。

数据报格式：

|IndexFlag(1字节)	|Index(4字节)	|Length(1字节)	|Buffer(可变，<=127)
|区别数据报种类	|记录校验块编号	|Buffer中所含的数据大小	|未匹配数据

基本思想：

按顺序读取完整文件中的数据块并得到校验块，按照建立的哈希表进行查询，若匹配（滚动和md4均匹配），则读取下一个数据块，若不匹配，则读取下一个字节（类似滑动窗口），并将先前头一个字节加入数据报中，再进行匹配分析。

注意点：

1. 数据报发送时机：数据报buffer字段满时，从不匹配到匹配时。

2. 发送数据报之前写入Length字段。

3. 结尾处理后，最好发送匹配的编号，对客户端重组有利。

4. 一些IO异常处理。

5. 发送时并不直接发送到socket的输出流中，而是先写入本地文件，因为某些缓冲和接受顺序原因。

6. 此部分较为复杂，可能有许多不合理的地方，需要不断完善。

源码：

public void startMatch(String filetotal, InputStream inStreamCheckSum,
OutputStream outStreamFIleBack, OutputStream outStreamMatchIndexs)
throws IOException {
BufferedInputStream inStreamFileTotal = new BufferedInputStream(
new FileInputStream(filetotal));
// BufferedOutputStream outStreamFIleBack = new BufferedOutputStream(
// outStreamBack);
// DataOutputStream outStreamMatchIndexs = new DataOutputStream(
// new BufferedOutputStream(outStreamMatchOrder));
DataInputStream bf = new DataInputStream(inStreamFileTotal);
ServerCheckSumHashCreater testCreateHash = new ServerCreateHash();
testCreateHash.buildHashTable(inStreamCheckSum);
byte[] bufferWindow = new byte[ServerCheckSumCreater.BLOCKSIZE];

int amount = 0;// 缓冲区实际数量
boolean newBufferSend = false;
boolean lastmatchFlag = true;
int bufferDataIndex = 0;
long lastMatchIndex = 0;
byte bufferSend[] = new byte[DATATOTALSIZE];
int matchindexs[] = new int[inStreamFileTotal.available()
/ ServerCheckSumCreater.BLOCKSIZE + 1];
int matchnum = 0;
try {
// 开始读取文件循环
while ((amount = bf.read(bufferWindow)) != -1) {
SumChecking.checkSum_Adler32(bufferWindow,
ServerCheckSumCreater.BLOCKSIZE);
int matchflag = testCreateHash.checkMatch(bufferWindow,
ServerCheckSumCreater.BLOCKSIZE);
if (matchflag != -1) {
lastMatchIndex++;
matchindexs[matchnum++] = matchflag;
}

while (matchflag == -1) {
if (lastmatchFlag == true) {
lastmatchFlag = false;
// writeIndexFlagTrue
bufferSend[INDEXFLAGOFF] = DatagramType.UNMATCHDATA_INDEX_VALID;
bufferSend[1] = (byte) (lastMatchIndex & 0xff);
bufferSend[2] = (byte) ((lastMatchIndex & 0xff00) >> 8);
bufferSend[3] = (byte) ((lastMatchIndex & 0xff0000) >> 16);
bufferSend[4] = (byte) ((lastMatchIndex & 0xff000000) >> 24);
} else if (newBufferSend) {
// writeIndexFlagFalse
bufferSend[INDEXFLAGOFF] = DatagramType.UNMATCHDATA_INDEX_INVALID;
newBufferSend = false;
}

// readNextByte
byte nextbyte = bf.readByte();
// addUnMatchByte
bufferSend[DATAOFF + bufferDataIndex] = bufferWindow[0];
bufferDataIndex++;
if (bufferDataIndex == DATAACTUALSIZE) {

// writeDataLength
bufferSend[LENGTHOFF] = (byte) bufferDataIndex;
// sendBuffer
outStreamFIleBack.write(bufferSend);
bufferDataIndex = 0;
newBufferSend = true;
}
System.arraycopy(bufferWindow, 1, bufferWindow, 0,
amount - 1);
bufferWindow[amount - 1] = nextbyte;
// computeCheckSum
matchflag = testCreateHash.checkMatch(bufferWindow,
ServerCheckSumCreater.BLOCKSIZE);

if (matchflag != -1) {
lastMatchIndex++;
matchindexs[matchnum++] = matchflag;
}

}// 未匹配内循环结束

if (!lastmatchFlag && !newBufferSend) {
// writeDataLength
bufferSend[LENGTHOFF] = (byte) (bufferDataIndex);
outStreamFIleBack.write(bufferSend);
bufferDataIndex = 0;
// getLastMatchIndex
}
newBufferSend = false;
lastmatchFlag = true;
}

} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();

if (bufferDataIndex > 0) {
bufferSend[LENGTHOFF] = (byte) (bufferDataIndex);
outStreamFIleBack.write(bufferSend);
bufferDataIndex = 0;
bufferSend[INDEXFLAGOFF] = DatagramType.UNMATCHDATA_INDEX_INVALID;
}

for (int i = 0; i < amount; i++) {
bufferSend[DATAOFF + bufferDataIndex] = bufferWindow[i];
bufferDataIndex++;
if (bufferDataIndex == DATAACTUALSIZE) {
// writeDataLength
bufferSend[LENGTHOFF] = (byte) bufferDataIndex;
// sendBuffer
outStreamFIleBack.write(bufferSend);
bufferDataIndex = 0;
}
}
if (bufferDataIndex > 0) {
bufferSend[LENGTHOFF] = (byte) (bufferDataIndex);
outStreamFIleBack.write(bufferSend);
bufferSend[INDEXFLAGOFF] = DatagramType.UNMATCHDATA_INDEX_INVALID;
bufferDataIndex = 0;
}
} finally {
// wirteMatchIndexs
bufferSend[INDEXFLAGOFF] = DatagramType.UNMATCHDATA_END;
outStreamFIleBack.write(bufferSend);

outStreamMatchIndexs.write(DataConvertUtils.int2bytes(matchnum));
for (int i = 0; i < matchnum; i++) {
outStreamMatchIndexs.write(DataConvertUtils
.int2bytes(matchindexs[i]));
}
out.println("matchnum:" + matchnum);
out.println("matchindexs:");
out.println(Arrays.toString(matchindexs));

//			//判断全部匹配和全部不匹配的特殊情况
//			if (matchnum == 0) {
//				bufferSend[INDEXFLAGOFF] = DatagramType.ZEROMATCH;
//			} else if (unmatchnum == 0) {
//				bufferSend[INDEXFLAGOFF] = DatagramType.ALLMATCH;
//			}

// try {
// inStreamFileTotal.close();
// outStreamFIleBack.close();
// outStreamMatchIndexs.close();
// } catch (IOException e) {
// e.printStackTrace();
// }
}
}