多文件云傳輸系統框架

文章目錄

• 多文件云傳輸系統框架
• • 1. 需求分析
  • 2.資源的表示
  • • 2.1文件片段化處理
    • • 2.1.1文件片段頭----- FileSectionHandle類
      • 2.1.2int與byte類型之間的轉換----- TypeUtil類
      • 2.1.3 文件片段-----FileSection類
    • 2.2 資源基本信息-----ResourceBaseInfo類
    • 2.3 資源-----Resource類
    • 2.4 掃描本地資源 ----- Scanner類
  • 3.系統結構及功能詳解
  • • 3.1 注冊中心
    • 3.2 資源發送者
    • 3.3 資源請求者
  • 4.分配策略
  • • 4.1 資源分配策略
    • • 4.11 IResourceDistribution類
      • 4.12 ResourceDistributionStrategy
    • 4.2節點分配策略
    • • 4.2.1 INetNodeStrategy
      • 4.2.2 隨機節點分配-----NetNodeStrategy類
      • 4.2.3 按節點的發送次數進行分配-----NetNodeSelectStrategy 類
  • 5 文件指針池-----RandAccessFilePool類
  • 6.斷點續傳的基礎------UnReceiveSection類
  • 7.注冊中心代碼
  • • 7.1 注冊中心啟動類------RegisterCenter類
    • 7.2 注冊中心RPC接口-----INodeAction
    • 7.3 注冊中心RPC接口實現類 ----- NodeAction類
    • 7.4 ResourceName類
    • 7.5 ResourceNode類
    • 7.6 關于注冊中心的偵聽者模式-----ISpeaker、IListener
  • 8.資源發送者
  • 9 資源接受者

1. 需求分析

我們希望能實現如下功能：

當一個客戶端請求資源時，會從注冊中心得到擁有該資源的所有網絡節點，該請求者會選出當前壓力最小的K個發送者，并對他們請求這個資源的不同部分，最終請求者將會得到他請求的資源。

我們最終希望能實現斷點續傳。斷點續傳有兩種情況：
情況1：比如發送端像請求端發送資源時，它還沒發送完就下線了，最終接收端將接收不到完整的資源，我們希望接收端能清楚自己的哪些文件沒有收到，并重新像其他擁有該資源的在線節點進行進行請求。
情況2： 接收端接收到一半，電腦關機了，我們希望他重新開機后，能從斷點處重新下載，而不是重頭下載。

2.資源的表示

我們把多個文件或者單個文件稱作資源，比如QQ和微信就是兩種資源。我們需要用一個類去描述該資源。

2.1文件片段化處理

文件傳輸需要用到網絡，所以不可能將一個很大的文件一口氣發過去，所以我們將一個文件片段話。
文件片段由兩部分組成：文件片段頭和該文件片段的內容。

2.1.1文件片段頭----- FileSectionHandle類

/*** 文件片段頭* 功能：* 1. 文件片段頭的功能是為了描述一個文件片段屬于哪一個文件，在此文件中的偏移量是多少，以及該片段內容的長度。* 2. 由于網絡間的傳輸是以byte為單位的，所以我們需要提供將文件片段頭變換為byte[]類型的方法，* 當然還需提供反變換的方法.* @author 田宜凡**/ public class FileSectionHandle {private int fileHandle; //文件片段所屬文件的文件句柄，最終會通過這個文件句柄去找到該文件的相對路徑，以及文件的大小。總體來說，文件句柄映射著一個文件。//為什么不直接把文件的路徑直接替換掉fileHandle，因為你這是文件片段頭，使用這些信息沒用，而且文件路徑的長度是不確定的。private int offset; //該片段的偏移量private int len;//還片段的長度public FileSectionHandle() {}//三參構造函數public FileSectionHandle(int fileHandle, int offset, int len) {this.fileHandle = fileHandle;this.offset = offset;this.len = len;}//將三個int類型的成員變為byte類型并且放到同一個byte數組中public byte[] tobytes() {byte[] result = new byte[12];byte[] fileHandleBytes = TypeUtil.intToBytes(fileHandle);byte[] offsetBytes = TypeUtil.intToBytes(offset);byte[] lenBytes = TypeUtil.intToBytes(len);setBytes(result, 0, fileHandleBytes);setBytes(result, 0 + 4, offsetBytes);setBytes(result, 0 + 8, lenBytes);return result;}//將byte[]變換為真正的成員public FileSectionHandle(byte[] value) {byte[] fileHandleBytes = getByte(value, 0, 4); byte[] offsetBytes = getByte(value, 0 + 4, 8); byte[] lenBytes = getByte(value, 0 + 8, 12);fileHandle = TypeUtil.bytesToInt(fileHandleBytes);offset = TypeUtil.bytesToInt(offsetBytes);len = TypeUtil.bytesToInt(lenBytes);}void setBytes(byte[] resouce, int start, byte[] target) {int length = target.length;int end = start + length;for (int i = start ; i < end ; i++) {resouce[i] = target[i % length];}}byte[] getByte(byte[] resource, int start, int end) {int length = end - start;byte[] result = new byte[length];for (int i = 0 ; i < length ; i++) {result[i] = resource[i + start];}return result;}public int getFileHandle() {return fileHandle;}public void setFileHandle(int fileHandle) {this.fileHandle = fileHandle;}public int getOffset() {return offset;}public void setOffset(int offset) {this.offset = offset;}public int getLen() {return len;}public void setLen(int len) {this.len = len;}@Overridepublic String toString() {return "fileHandle=" + fileHandle + ", offset=" + offset + ", len=" + len + "\n";} }

2.1.2int與byte類型之間的轉換----- TypeUtil類

/*** 功能；用于int類型與byte類型的轉換(核心思想： 位運算)* @author ty**/ public class TypeUtil {public TypeUtil() {}/*** * @param 一個int的數據* @return 將int數據轉換為byte[]類型*/public static byte[] intToBytes (int value) {byte[] result = new byte[4];for (int i = 0 ; i < 4 ; i++) {//如果將int強轉為byte保留的是低八位result[i] = (byte) (value >> (8 * i));}return result;}/*** * @param 一個長度為4的byte[]數據* @return 將byte[]數據轉換為int類型。*/public static int bytesToInt(byte[] value) {int length = value.length;int result = 0;for (int i = 0 ; i < length ; i++) {result |= ((((int)value[i]) & 0xFF) << (8 * i));}return result;} }

2.1.3 文件片段-----FileSection類

/*** 用于表示一個文件片段，每一個文件片段都能通過文件句柄對應一個文件基本信息* @author 田宜凡**/ public class FileSection {//文件片段頭private FileSectionHandle fileSectionHandle;//本片段的字節內容private byte[] value;public FileSection() {fileSectionHandle = new FileSectionHandle();}public FileSection(int fileHandle, int offset, int len) {this.fileSectionHandle = new FileSectionHandle();fileSectionHandle.setFileHandle(fileHandle);fileSectionHandle.setOffset(offset);fileSectionHandle.setLen(len);}public FileSectionHandle getFileSectionHandle() {return fileSectionHandle;}public void setFileSectionHandle(FileSectionHandle fileSectionHandle) {this.fileSectionHandle = fileSectionHandle;}public byte[] getValue() {return value;}public void setValue(byte[] value) {this.value = value;}public void setFileHandle(int fileHandle) {fileSectionHandle.setFileHandle(fileHandle);}public int getFileHandle() {return fileSectionHandle.getFileHandle();}public void setLen(int len) {fileSectionHandle.setLen(len);}public int getLen() {return fileSectionHandle.getLen();}public void setOffset(int offset) {fileSectionHandle.setOffset(offset);}public int getOffSet() {return fileSectionHandle.getOffset();}@Overridepublic String toString() {return fileSectionHandle.toString();}}

2.2 資源基本信息-----ResourceBaseInfo類

/*** 文件基本信息，對應著一個文件，文件片段會通過文件句柄，找到對應的文件基本信息* @author ty**/ public class ResourceBaseInfo {private int fileHandle;//文件句柄private String relativePath;//該文件的相對路徑private long size;//該文件的大小public int getFileHandle() {return fileHandle;}public void setFileHandle(int fileHandle) {this.fileHandle = fileHandle;}public String getRelativePath() {return relativePath;}public void setRelativePath(String relativePath) {this.relativePath = relativePath;}public long getSize() {return size;}public void setSize(long size) {this.size = size;}@Overridepublic String toString() {return "fileHandle=" + fileHandle + ", relativePath=" + relativePath + ", size=" + size + "\n";} }

2.3 資源-----Resource類

package com.mec.ManyFile.resource;import java.util.List; /*** 資源，一個資源可能是單文件也可能是多文件* @author ty**/ public class Resource {private String AppName;//資源名稱private String absolutePath;//資源絕對根路徑private String version;//資源的版本private List<FileSection> FileSectionList;//文件片段列表private List<ResourceBaseInfo> baseInfoList;//資源基本信息列表，請求時這個列表為空。public Resource() { }public void setAbsolutePath(String absolutePath) {this.absolutePath = absolutePath;}public String getAppName() {return AppName;}public void setAppName(String appName) {AppName = appName;}public String getVersion() {return version;}public void setVersion(String version) {this.version = version;}public List<FileSection> getFileSectionList() {return FileSectionList;}public void setFileSectionList(List<FileSection> fileSectionList) {FileSectionList = fileSectionList;}public List<ResourceBaseInfo> getBaseInfoList() {return baseInfoList;}public void setBaseInfoList(List<ResourceBaseInfo> baseInfoList) {this.baseInfoList = baseInfoList;}public String getAbsolutePath() {return absolutePath;}//找到一個文件片段對對應的文件（資源基本信息）public ResourceBaseInfo getResourceBaseInfo(FileSection section) {int fileHandle = section.getFileHandle();return getSameFileHandle(fileHandle);}//這個方法用于找到文件句柄相同的資源基本信息private ResourceBaseInfo getSameFileHandle(int filehandle) {for (ResourceBaseInfo rbi : baseInfoList) {int temp = rbi.getFileHandle();if (temp == filehandle) {return rbi;}}return null;}@Overridepublic String toString() {StringBuffer sb = new StringBuffer();sb.append(AppName + "\n" + absolutePath + "\n"+ version+ "\n");if (FileSectionList != null) {for (FileSection filesection : FileSectionList) {sb.append(filesection.toString());} }if (baseInfoList != null) {for (ResourceBaseInfo rbi : baseInfoList) {sb.append(rbi.toString());}}return sb.toString();} }

2.4 掃描本地資源 ----- Scanner類

/*** 1.遞歸掃描本地文件* 2.自動獲取對應文件的大小、相對路徑* 3.自動添加文件句柄* @author ty**/ public class Scanner {public Scanner() {}/*** * @param appPath 資源所對應的根目錄* @return*/public List<ResourceBaseInfo> ScannerAppPath(String appPath) {File file = new File(appPath);List<ResourceBaseInfo> rbiList = new ArrayList<>();explore(appPath, file, rbiList, 0);return rbiList;}int explore(String appPath, File file, List<ResourceBaseInfo> rbiList, int fileHandle) {File[] files = file.listFiles();for (File f : files) {if (f.isFile()) {fileHandle = creatResourceBaseInfo(appPath, f, rbiList, ++fileHandle);}if (f.isDirectory()) {fileHandle = explore(appPath, f,rbiList, fileHandle);}}return fileHandle;}int creatResourceBaseInfo(String appPath, File file, List<ResourceBaseInfo> rbiList, int fileHandle) {ResourceBaseInfo resourceBaseInfo = new ResourceBaseInfo();resourceBaseInfo.setFileHandle(fileHandle);resourceBaseInfo.setRelativePath(file.getAbsolutePath().replace(appPath + "\\", ""));resourceBaseInfo.setSize(file.length());rbiList.add(resourceBaseInfo);return fileHandle;} }

3.系統結構及功能詳解

3.1 注冊中心

功能分析：

每當有客戶端上線時，客戶端都要以RPC方式連接注冊測中心，以資源名稱#資源版本號為鍵，比如 QQ#1，以NetNode類為值，向注冊中心注冊自己擁有的所有資源。

當一個發送者下線時，需要將自己的節點注銷掉。正常下線還好辦，但是異常下線我們就無法注銷了，這顯然是不合理的，當請求者RPC連接發送者時，如果連不上，會出現異常處理，這時再進行注銷操作。

資源請求者可以從注冊中心得到一個資源所對應網絡節點列表。

注冊中心可以動態更新每一個網絡節點（NetNode）的發送次數。

除了注冊資源名稱還要注冊每個資源的資源基本信息，比如該資源有多少個文件，每個文件有多大，以及文件的相對路徑。這些都寫在一個類里面。當然與之對應的也要有注銷。得到資源基本信息的操作。

綜上所述：注冊中心主要有三個核心功能：注冊、注銷、得到列表。
NetNode類：
此類有三個成員

//網絡節點 public class NetNode {private int port;//網絡節點的端口號private String ip;//網絡節點的IP地址private int sendingTime;該節點已經發送了多少次資源public NetNode(int port, String ip, int sendingTime) {super();this.port = port;this.ip = ip;this.sendingTime = sendingTime;}public NetNode() {}public int getPort() {return port;}public void setPort(int port) {this.port = port;}public String getIp() {return ip;}public void setIp(String ip) {this.ip = ip;}public int getSendingTime() {return sendingTime;}public void setSendingCount(int sendingTime) {this.sendingTime = sendingTime;}public void increase() {sendingTime++;}public void crease() {sendingTime--;}public String toString() {return "NetNode [port=" + port + ", ip=" + ip + ", sendingTime=" + sendingTime + "]";}@Overridepublic int hashCode() {final int prime = 31;int result = 1;result = prime * result + ((ip == null) ? 0 : ip.hashCode());result = prime * result + port;return result;}@Overridepublic boolean equals(Object obj) {if (this == obj)return true;if (obj == null)return false;if (getClass() != obj.getClass())return false;NetNode other = (NetNode) obj;if (ip == null) {if (other.ip != null)return false;} else if (!ip.equals(other.ip))return false;if (port != other.port)return false;return true;}}

3.2 資源發送者

功能分析：

資源請求者會開啟資源資源接收服務器，和RPC客戶端。通過RPC的方式將自己資源接收服務器的ip地址和端口號以及要請求的資源基本信息發過去，讓資源發送者連接資源請求者，并發送請求的資源。

3.3 資源請求者

功能分析：

為了實現斷點續傳，需要對接收的文件片端進行記錄

每接收到一個文件片段。要根據接收端自己設定的絕對根路徑，結合資源基本信息的中的相對路徑把文件片段，寫入到磁盤上。

4.分配策略

4.1 資源分配策略

4.11 IResourceDistribution類

/*** 不論是默認資源分配策略還是自定義資源分配策略都有應有* 1.默認文件片段大小* 2.最大文件片段大小* 3.分配單文件資源* 4.分配多文件資源* @author ty**/ public interface IResourceDistribution {long DEFAULT_SIZE = 1 << 14;long MAX_SIZE = 1 << 15;public List<List<FileSection>> divideResourceBaseInfos(ResourceBaseInfo rbi, List<NetNode> nodeList);public List<List<FileSection>> divideResourceBaseInfo(List<ResourceBaseInfo> rbis, List<NetNode> nodeList); }

4.12 ResourceDistributionStrategy

/** 資源分配策略：* 我們收到了資源的信息列表，叢中我們可以知道* 1.每一個文件的句柄* 2.相對路徑* 3.以及該文件的大小* 4.我們根據文件的大小進行分片，訂一個默認的大小* 5.如果該文件的大小小于默認大小，不用進行分片，* 6.如果大于默認大小就要進行分片* 這個默認大小最終我們希望實現可配置。* */ public class ResourceDistributionStrategy implements IResourceDistribution{private long bufferSize = DEFAULT_SIZE;public ResourceDistributionStrategy() {}public void setBufferSize(long bufferSize) {if (bufferSize < 0 || bufferSize > MAX_SIZE) {return;}this.bufferSize = bufferSize;}//分配單文件資源public List<List<FileSection>> divideResourceBaseInfos(ResourceBaseInfo rbi, List<NetNode> nodeList) {List<ResourceBaseInfo> rbis = new ArrayList<ResourceBaseInfo>();rbis.add(rbi);List<List<FileSection>> result = divideResourceBaseInfo(rbis, nodeList);return result;}/*** 功能：<br>* 1.根據發送端列表得知發送端的個數* 2.遍歷每個文件信息，對文件的大小進行分解* 3.最終得到和發送端個數一致的文件片段堆* @param rbis 資源的所有文件列表* @param nodeList 發送端列表* @return 得到根據發送者的數量分配的文件片段列表*/public List<List<FileSection>> divideResourceBaseInfo(List<ResourceBaseInfo> rbis, List<NetNode> nodeList) {int sendCount = nodeList.size();int index = 0;List<List<FileSection>> result = new ArrayList<List<FileSection>>();for (int i = 0 ; i < sendCount ; i++) {List<FileSection> temp = new ArrayList<FileSection>();result.add(temp);}for (ResourceBaseInfo rbi : rbis) {long size = rbi.getSize();int fileHandle = rbi.getFileHandle();if (size < bufferSize) {FileSection fileSection = new FileSection(fileHandle, 0 ,(int)size);List<FileSection> secList = result.get(index);index = (index + 1) % sendCount;secList.add(fileSection);} else {long restSize = size;int offset = 0;int len;while (restSize != 0) {len = (int) (restSize > bufferSize ? bufferSize : restSize);FileSection fileSection = new FileSection(fileHandle, offset ,(int)len);offset += len;restSize -= len;List<FileSection> secList = result.get(index);index = (index + 1) % sendCount;secList.add(fileSection);}}}return result;}}

4.2節點分配策略

關于節點分配我有兩種想法，一種是隨機分配，一種是根據每個節點已經發送的次數進行分配 INetNodeStrategy

4.2.1 INetNodeStrategy

/*** 接點分配策略接口* 1.設置默認發送次數* 2.設置最大發送次數* 3.選擇網絡節點* @author ty**/ public interface INetNodeStrategy {int DEFAULT_SENDER_COUNT = 3;int MAX_SENDER_COUNT = 20;List<NetNode> SelectNetNdoe(List<NetNode> netNodeLists); }

4.2.2 隨機節點分配-----NetNodeStrategy類

/** 這里的節點分配采用的是隨機的辦法 1.根據需要發送端的個數，將整個接點列表分為多份 2.從每一份中隨機挑選一個節點 @author ty* */ public class NetNodeStrategy implements INetNodeStrategy{private static int maxSendCount = DEFAULT_SENDER_COUNT;public NetNodeStrategy() {}public List<NetNode> SelectNetNdoe(List<NetNode> netNodeLists) {int sendCount = netNodeLists.size();if (sendCount <= maxSendCount) {return netNodeLists;} else {return getSendNodeList(netNodeLists);}}private List<NetNode> getSendNodeList(List<NetNode> netNodeLists) {List<NetNode> netList = new ArrayList<NetNode>();int sendCount = netNodeLists.size();int oneGroupCount = sendCount / maxSendCount;int restCount = sendCount % maxSendCount;Random rand = new Random();for (int i = 0 ; i < maxSendCount ; i++) {int temp = i == (maxSendCount - 1) ? rand.nextInt(oneGroupCount + restCount): rand.nextInt(oneGroupCount);int index = temp + i * oneGroupCount;netList.add(netNodeLists.get(index));}return netList;} }

4.2.3 按節點的發送次數進行分配-----NetNodeSelectStrategy 類

該分配策略的核心問題：從多個節點中找到發送次數的幾個
一般我們會進行升序排序，然后選出最小的是幾個，一般排序的時間復雜度為O(n^2),我采用的方法將時間復雜度控制到最大O(3n);
算法圖解：

/** 這里的節點分配采用的找出發送次數最少的幾個節點1.遍歷節點列表2.找出發送次數最少的是三個節點* */ public class NetNodeSelectStrategy implements INetNodeStrategy{private int maxSenderCount = DEFAULT_SENDER_COUNT;public NetNodeSelectStrategy() {}public void setMaxSenderCount(int maxSenderCount) {this.maxSenderCount = maxSenderCount < MAX_SENDER_COUNT ? maxSenderCount : MAX_SENDER_COUNT;}@Overridepublic List<NetNode> SelectNetNdoe(List<NetNode> netNodeLists) {int sendCount = netNodeLists.size();if (sendCount <= maxSenderCount) {return netNodeLists;} else {return getMinSendNodeList(netNodeLists);}}//具體算法請看算法圖解private List<NetNode> getMinSendNodeList(List<NetNode> netNodeLists) {List<NetNode> result = new ArrayList<NetNode>();NetNode maxNode = netNodeLists.get(0);for (int i = 1 ; i < netNodeLists.size() ; i++) {NetNode temp = netNodeLists.get(i);if (temp.getSendingTime() > maxNode.getSendingTime()) {maxNode = temp;}}int maxsendingTime = maxNode.getSendingTime();int[] sendCount = new int[maxsendingTime + 1];for(NetNode node : netNodeLists) {sendCount[node.getSendingTime()]++;}int maxSendCount = maxSenderCount;System.out.println("maxSendCount" + maxSendCount);for (int i = 0 ; i < sendCount.length ; i++) {if (maxSendCount <= 0) {sendCount[i] = 0;continue;}maxSendCount -= sendCount[i]; System.out.println("s " +maxSendCount);if(maxSendCount < 0) {maxSendCount += sendCount[i];sendCount[i] = maxSendCount;}}for (NetNode netNode : netNodeLists) {int time = netNode.getSendingTime();if (sendCount[time] == 0) {continue;}sendCount[time]--;result.add(netNode);}return result;} }

5 文件指針池-----RandAccessFilePool類

每一次給一個文件中的指定位置，寫一個片段，都需要RandAccessFile對象，該對象用完后需要關閉,但是這個文件整體沒有接受完的時候。
就會存在RandAccessFile對象不停的創建以及關閉的問題。這樣很費時，所以以一個文件的路徑為鍵，以文件指針為值將它緩存起來，只有當一個文件全部接收完時，我們在關閉它。

public class RandAccessFilePool {private Map<String, RandomAccessFile> rafPool;RandAccessFilePool() {rafPool = new ConcurrentHashMap<>();}RandomAccessFile getRaf(String filePath) {RandomAccessFile raf = rafPool.get(filePath);if (raf == null) {try {// TODO 根據filePath，創建相關目錄raf = new RandomAccessFile(filePath, "rw");rafPool.put(filePath, raf);} catch (FileNotFoundException e) {e.printStackTrace();}}return raf;}void close(String filePath) {RandomAccessFile raf = rafPool.remove(filePath);raf.close();} }

6.斷點續傳的基礎------UnReceiveSection類

我們能實現斷點續傳的基礎是UnReceiveSection類，而該類的基礎是FileSection類，因為該類的對象保存了該文件片段在文件中的偏移量和長度。
我們使用一個FileSection的List來保存未接收的文件片段

/*** 這一個類對應一個文件，用來記錄我們的文件有沒有接受完畢* @author ty**/ public class UnReceiveSection {private int fileHandle//文件句柄private List<FileSection> unReceiveList;//未接收文件片段列表//初始化時，給unReceiveList加入一個文件片段，偏移量為0，片段長度為文件長度public UnReceiveSection(int fileHandle, int fileSize) {unReceiveList = new LinkedList<FileSection>();this.fileHandle = fileHandle;FileSection fileSection = new FileSection();fileSection.setFileHandle(fileHandle);fileSection.setLen(fileSize);fileSection.setOffset(0);unReceiveList.add(fileSection);}public int getFileHandle() {return fileHandle;}public void setFileHandle(int fileHandle) {this.fileHandle = fileHandle;}//給unReceiveList加入一個文件片段，具體思想請看上圖public void addFileSection(FileSection fileSection) {int fileHandle = fileSection.getFileHandle();if (fileHandle != this.fileHandle) {return;}int targetOffset = fileSection.getOffSet();int targetLen = fileSection.getLen();int targetIndex = targetLen + targetOffset;FileSection section = getDivideSection(targetOffset, targetLen);if (section == null) {return;}int srcOffset = section.getOffSet();int srcLen = section.getLen();int srcIndex = srcOffset + srcLen;if (targetOffset == srcOffset&& targetIndex == srcIndex) {return;}int leftOffset = srcIndex;int leftLen = targetOffset - srcOffset;int rightOffset = targetOffset + targetOffset;int rightLen = srcOffset + srcLen - rightOffset;if(leftLen != 0) {unReceiveList.add(new FileSection(fileHandle, leftOffset, leftLen));}if(rightLen != 0) {unReceiveList.add(new FileSection(fileHandle, rightOffset, rightLen));}}private FileSection getDivideSection(int offset, int len) {int targetIndex = offset + len;FileSection fileSection = null;for (FileSection section : unReceiveList) {int tmpOffset = section.getOffSet();int tmpLen = section.getLen();int srcIndex = tmpOffset + tmpLen;if (offset >= tmpOffset && targetIndex <= srcIndex) {fileSection = section;}}unReceiveList.remove(fileSection);return fileSection;}//unReceiveList為空代表列表該文件接收完成public boolean isFinish() {if (unReceiveList.size() == 0) {return true;}return false;}public List<FileSection> getUnReceiveFileSection() {return unReceiveList;} }

7.注冊中心代碼

當初有想過資源發送端與注冊中心之間進行長連接，因為異常掉線后，注冊中心可以及時注銷掉該節點，預防資源求者得到已經下線的節點列表，但是不論是資源請求端還是資源發送端。對于App服務器來說都為客戶端，所以這個數量很大，如果與注冊中心長連接的話，注冊中心的壓力很大，為了緩解這種壓力，我們采用短連接（RPC）。
短連接解決資源發送端異常掉線問題

短連接就是所謂的一回合，無狀態連接，所以注冊中心無法得知異常掉線

雖然可以在資源請求端RPC資源發送端時的以異常里進行注銷的操作，然后采用其他可用的節點。這樣就會產生一個問題：請求得到節點列表中摻雜了很多用不了的節點。

所以我們打算比如每半天對注冊中心進行一次心跳，讓它短連接所有的資源發送者，成功什么都不做，連接失敗了就注銷該節點，這樣可以定期銷毀不能使用的網絡節點。但是我嘗試了很多方式，都行不通。起初我本來想使用每個資源發送者的RPC服務器。我可以讓注冊中心連接資源發送者的RPC服務器進行一次短連接，如果連接失敗，注冊中心就可以認定資源發送者掉線了，聽起來很美好，但是判斷連接失敗的時間太長了。長到系統根本無法接受，所以這也是我這個系統留下的最大的遺憾。

7.1 注冊中心啟動類------RegisterCenter類

/*** 注冊中心功能：* 1.開啟RPC服務器服務器* 2.正常關閉RPC服務器* 3.提供默認端口號* @author ty**/ public class RegisterCenter implements ISpeaker{private RMIServer rmiServer;private int rmiPort;private static final int RMIDEFAULT_PORT = 54199; private List<IListener> listenrList;public RegisterCenter() {this(RMIDEFAULT_PORT);}public RegisterCenter(int rmiPort) {this.rmiPort = rmiPort;}public void setListenrList(List<IListener> listenrList) {this.listenrList = listenrList;}public void setRmiPort(int rmiPort) {this.rmiPort = rmiPort;}public void startup() {reportMessage("正在開啟注冊中心.....");rmiServer = new RMIServer(rmiPort);reportMessage("注冊中心開啟成功.....");rmiServer.startup();reportMessage("短連接服務器開始偵聽客戶端");rmiServer.registory("com.mec.ManyFile.RegistCenter");}public void shutdown() {rmiServer.close();reportMessage("短連接服務器正常關閉...");}@Overridepublic void addListener(IListener iListtener) {if (listenrList == null) {listenrList = new ArrayList<>(); }if (listenrList.contains(iListtener)) {return; }listenrList.add(iListtener);}@Overridepublic void removeListener(IListener iListtener) {if (!listenrList.contains(iListtener)) {return; }listenrList.remove(iListtener);}public void reportMessage(String message) {if (listenrList == null || listenrList.size() == 0) {return;}for (IListener listen : listenrList) {listen.dealMessage(message);}}

7.2 注冊中心RPC接口-----INodeAction

/*** 此接口更包含了注冊中心所擁有的功能，為了RPC的調用* @author ty**/ public interface INodeAction {void logoutNode(NetNode node);//注銷一個節點void registerNode(ResourceName service, NetNode node);//注冊一個節點List<NetNode> getNodeList(ResourceName service);//得到一個資源的節點列表void logout(ResourceName res);//注銷一個資源的信息void register(ResourceName service, Resource res);//注冊一個資源的信息Resource getResource(ResourceName service);//得到資源信息void inCreaseSendCount(NetNode node);//增加一個節點的發送次數void CreaseSendCount(NetNode node);//減少一個節點的發送次數 } }

7.3 注冊中心RPC接口實現類 ----- NodeAction類

我認為此類中的注銷節點，增加發送次數，減少發送次數這三個操作寫的時間復雜度都很高，但是我們希望這里能快速執行完，所有我覺得這里處理不是很好，希望以后會有更好的辦法。

/*** 注冊中心RPC實現類* @author ty**/ @Interfaces(interfacees = {INodeAction.class}) public class NodeAction implements INodeAction{/*** relationMap * 鍵為 資源名#版本號字符串 * 值為 該資源的資源信息以及擁有該資源的網絡節點*/private static Map<String, ResourceNode> relationMap = new ConcurrentHashMap<String, ResourceNode>();//遍歷每一個鍵所對應的節點列表，并在每一個節點列表中找到要注銷的NetNode，然后刪除//這種方法我感覺時間復雜度很高，不是很滿意，希望以后可以改進@Overridepublic void logoutNode(NetNode node) {Set<String> keyset = relationMap.keySet();Iterator<String> set = keyset.iterator();while(set.hasNext()) {String key = set.next();ResourceNode resNode = relationMap.get(key);List<NetNode> nodeList = resNode.getNetNodes();NetNode temp = null;for (NetNode one : nodeList) {if (one.equals(node)) {temp = one;break;} }nodeList.remove(temp);}}/*** 注冊一個節點* 根據鍵值在relationMap中找到有沒有對應的值，沒有的話初始化一個值放進去，再把node放進去*/@Overridepublic void registerNode(ResourceName service, NetNode node) {String key = service.toString();ResourceNode resNode = relationMap.get(key);if(resNode == null) {resNode = new ResourceNode();relationMap.put(key, resNode);}List<NetNode> NodeList = resNode.getNetNodes();if (NodeList.contains(node)) {return;}NodeList.add(node);}/*** 得到節點列表* 1.首先得判斷你尋求的節點信息列表存不存在* 2.存在的話返回，不存在返回null*/@Overridepublic List<NetNode> getNodeList(ResourceName service) {ResourceNode node = relationMap.get(service.toString());if (node == null) {return null;}return node.getNetNodes();}/*** 注銷資源信息，這是由APP服務器做的事情* 如果資源信息都了被服務器刪除了，關于這個資源的節點列表也要刪除*因為資源已經被APP服務器拋棄了*/@Overridepublic void logout(ResourceName service) {String key = service.toString();if(relationMap.get(key) == null) {return;}relationMap.remove(key);}/*** 注冊資源信息，由APP服務器進行* 先判斷在HashMap中鍵存不存在。沒有的話要先初始化*/@Overridepublic void register(ResourceName service, Resource res) {ResourceNode resNode = relationMap.get(service);if(resNode == null) {resNode = new ResourceNode();relationMap.put(service.toString(), resNode);}resNode.setRes(res);}/*** 得到資源信息* 從relationMap中根據傳進來的鍵得到資源信息* 如找找不到，返回null*/@Overridepublic Resource getResource(ResourceName service) {ResourceNode node = relationMap.get(service.toString());if (node == null) {return null;}return node.getRes();}/**增加發送次數，節點分配策略就是基于此數值的，所以每當一個發送端被分配出去，就我們就要通過RPC使這個節點的發送次數次數加一**/@Overridepublic void inCreaseSendCount(NetNode node) {Set<String> keyset = relationMap.keySet();Iterator<String> set = keyset.iterator();while(set.hasNext()) {String key = set.next();ResourceNode resNode = relationMap.get(key);List<NetNode> nodeList = resNode.getNetNodes();NetNode temp = null;for (NetNode one : nodeList) {if (one.equals(node)) {one.increase();break;} }}}/**減少發送次數，節點分配策略就是基于此數值的，每當一個節點發送完畢，既然讓個值減一**/@Overridepublic void CreaseSendCount(NetNode node) {Set<String> keyset = relationMap.keySet();Iterator<String> set = keyset.iterator();while(set.hasNext()) {String key = set.next();ResourceNode resNode = relationMap.get(key);List<NetNode> nodeList = resNode.getNetNodes();NetNode temp = null;for (NetNode one : nodeList) {if (one.equals(node)) {one.crease();break;} }}} }

7.4 ResourceName類

/*** 注冊中心的關系表中的鍵* 1.資源的名稱* 2.資源的版本* 最終根據toString()方法，以字符串的身份作為鍵 **/ public class ResourceName {String appName;String version;public ResourceName() {}public ResourceName(ResourceName resourceName) {this.appName = resourceName.getAppName();this.version = resourceName.getVersion();}public String getAppName() {return appName;}public void setAppName(String appName) {this.appName = appName;}public String getVersion() {return version;}public void setVersion(String version) {this.version = version;}@Overridepublic String toString() {return appName + "#" + version;}}

7.5 ResourceNode類

/*** 注冊中心關系表中的值，由兩部分組成* 1.資源信息* 2.節點信息列表* @author ty**/ public class ResourceNode{private Resource res;private List<NetNode> netNodes;public ResourceNode() {netNodes = new LinkedList<>();}public Resource getRes() {return res;}public void setRes(Resource res) {this.res = res;}public List<NetNode> getNetNodes() {return netNodes;}public void setNetNode(List<NetNode> netNode) {this.netNodes = netNode;} }

7.6 關于注冊中心的偵聽者模式-----ISpeaker、IListener

偵聽者機制的作用
比如服務器器開啟后你希望向界面上輸出一些東西，或者將一些信息寫入日志，但是當前狀態下并沒有界面，只有信息，如何把這個信息傳遞到未來才可能出現的界面上，偵聽者機制就可以很好的處理這個問題，首先偵聽者機制有兩個重要的接口ISpeaker、IListener

public interface ISpeaker {void addListener(IListener iListtener);void removeListener(IListener iListtener); } public interface IListener {void dealMessage(String message); }

注冊中心實現ISpeaker接口，未來的界面實現IListener接口

//此段代碼截取了RegisterCenter類的一部分內容 //這些內容就是實現偵聽者模式的全部，并不負載，但是要求你要對接口很熟悉 private List<IListener> listenrList;@Overridepublic void addListener(IListener iListtener) {if (listenrList == null) {listenrList = new ArrayList<>(); }if (listenrList.contains(iListtener)) {return; }listenrList.add(iListtener);}@Overridepublic void removeListener(IListener iListtener) {if (!listenrList.contains(iListtener)) {return; }listenrList.remove(iListtener);}public void reportMessage(String message) {if (listenrList == null || listenrList.size() == 0) {return;}for (IListener listen : listenrList) {listen.dealMessage(message);}}

在注冊中心使用時，你只需要調用reportMessage(String message)方法把你要傳遞出去的信息作為參數傳進去就行。具體把信息輸出到哪里，還得看IListenner的實現類怎么去寫dealMessage(String message)方法，最后再將IListener的實現類通過void addListener(IListener iListtener)方法提前加進去即可。

8.資源發送者

/*發送者：* 1.收到對方發來的資源請求，以及請求者的ip 和 port* 2.連接接收者服務器* 3.從本地中提取文件片段* 4.提取一個發送一個* */ public class Send {private Socket socket;RMIServer rmiServer;private String ip;private int port;private int RMIport;private DataOutputStream dos;private RafPool rafPool;public Send() { this("192.168.181.1",54188);}public Send(String ip, int port) {this.ip = ip;this.port = port;rafPool = new RafPool();}//初始化RMI服務器public void initRMIServer() {rmiServer = new RMIServer(RMIport);rmiServer.startup();rmiServer.registory("com.mec.ManyFile.send");}public String getIp() {return ip;}public void setIp(String ip) {this.ip = ip;}public int getPort() {return port;}public void setPort(int port) {this.port = port;}public int getRMIport() {return RMIport;}public void setRMIport(int rMIport) {RMIport = rMIport;}public void connectToServer() {try {socket = new Socket(ip, port);dos = new DataOutputStream(socket.getOutputStream()); } catch (UnknownHostException e) {e.printStackTrace();} catch (IOException e) {e.printStackTrace();} }void sendFileSection(FileSection fileSection) {FileSectionHandle fileHandle = fileSection.getFileSectionHandle();byte[] fileHandleByte = fileHandle.tobytes();byte[] value = fileSection.getValue();try {dos.write(fileHandleByte, 0, 12);dos.write(value,0, fileHandle.getLen());} catch (IOException e) {e.printStackTrace();}}//從本地中讀取這個文件片段public FileSection getFileSectionFromNative(FileSection section, String filePath) {RandomAccessFile raf = rafPool.get(filePath);int offset = section.getOffSet();int len = section.getLen();byte[] result = null;try {raf.seek(offset);result = new byte[len];raf.read(result);section.setValue(result);} catch (IOException e) {e.printStackTrace();}return section;}//這個參數resource擁有所有信息public void sendResource(Resource resource) {String absoluPath = resource.getAbsolutePath();List<FileSection> sectionList = resource.getFileSectionList();for (FileSection section : sectionList) {ResourceBaseInfo rbi = resource.getResourceBaseInfo(section);//通過這個rbi和section就可以得到這個文件片段的路徑String relaPath = rbi.getRelativePath();String filePath = absoluPath + "\\" + relaPath;FileSection resultSection = getFileSectionFromNative(section, filePath);sendFileSection(resultSection);}}public void close() {if (socket != null) {try {socket.close();} catch (IOException e) {} finally {socket = null;}}if (dos != null) {try {dos.close();} catch (IOException e) {} finally {dos = null;}}}public void RMIServerClose() {rmiServer.close();} } //資源請求接口 public interface IResquset {void send(NetNode receiver, Resource res); } /*** 連接資源接收者服務器* 發送資源* @author ty**/ public class Resquest implements IResquset{public Resquest() {}@Overridepublic void send(NetNode receiver, Resource res) {String ip = receiver.getIp();int port = receiver.getPort();Send send = new Send(ip, port);send.connectToServer();ResourceInfoPool infoPool = new ResourceInfoPool();ResourceName name = new ResourceName();name.setAppName(res.getAppName());name.setVersion(res.getVersion());Resource src = infoPool.gets(name);res.setBaseInfoList(src.getBaseInfoList());send.sendResource(res);} }

9 資源接受者

1.資源接收者控制類

package com.mec.ManyFile.receive;import java.io.IOException; import java.net.ServerSocket; import java.net.Socket; import java.util.ArrayList; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap;import com.mec.ManyFile.resource.FileSection; import com.mec.ManyFile.resource.Resource; import com.mec.ManyFile.resource.ResourceBaseInfo; import com.mec.ManyFile.resource.UnReceiveSection; import com.mec.rmi.core.IRMIAction; import com.mec.rmi.core.RMIAction; import com.mec.rmi.core.RMIClient; /*** 資源接收者服務器* 1.開啟等待資源發送者的連接* 2.每等待一個開啟一個線程去完成接受* @author ty**/ public class Receive implements Runnable{private ServerSocket receiveServer;private int port;private volatile boolean goon; private Resource resource; //保存每個文件的未接收片段private Map<Integer, UnReceiveSection> unReceiveMap; private IRMIAction iConnectError;public Receive() { this(54188);}public Receive(int port) {this.port = port;this.unReceiveMap = new ConcurrentHashMap<Integer, UnReceiveSection>();iConnectError = new RMIAction();}public void setPort(int port) {this.port = port;}public Resource getResource() {return resource;}public void setiConnectError(IRMIAction iConnectError) {this.iConnectError = iConnectError;}public void setResource(Resource resource) {this.resource = resource;}public void startUp() {if (port == 0) {return;}if (goon == true) {return;}try {receiveServer = new ServerSocket(port);goon = true;new Thread(this).start();initUnReceiveMap();} catch (IOException e) {e.printStackTrace();}}//初始化每一個文件對應的未接受片段列表private void initUnReceiveMap() {List<ResourceBaseInfo> resList = resource.getBaseInfoList();for (ResourceBaseInfo res : resList) {int fileSize = (int) res.getSize();int fileHandle = res.getFileHandle();UnReceiveSection unReceiveSection = new UnReceiveSection(fileHandle, fileSize);unReceiveMap.put(fileHandle, unReceiveSection);}}public void shutdown() {close();}public <T> T getProxy(String ip, int port, Class<?> clazz) {RMIClient rmiClient = new RMIClient();rmiClient.setIp(ip);rmiClient.setPort(port);rmiClient.setRmiAction(iConnectError);return rmiClient.getProxy(clazz);}private void close() {if(goon == false) {return;}try {receiveServer.close();} catch (IOException e) {e.printStackTrace();} finally {goon = false;}}//整合每個文件的未接收的文件片段，將它們整合到一個列表里public List<FileSection> getUnFileSection() {List<FileSection> sectionList = new ArrayList<FileSection>();Set<Integer> keys = unReceiveMap.keySet();for (Integer key : keys) {UnReceiveSection unRec = unReceiveMap.get(key);if (!unRec.isFinish()) {sectionList.addAll(unRec.getUnReceiveFileSection());}}return sectionList;}@Overridepublic void run() {while (goon) {try {Socket socket = receiveServer.accept();new DealReceive(socket, resource, unReceiveMap);} catch (IOException e) {//文件接收服務器異常掉線}}}}

2.處理接受的資源

package com.mec.ManyFile.receive;import java.io.DataInputStream; import java.io.IOException; import java.io.RandomAccessFile; import java.net.Socket; import java.util.LinkedList; import java.util.List; import java.util.Map;import com.mec.ManyFile.resource.FileSectionHandle; import com.mec.ManyFile.resource.Resource; import com.mec.ManyFile.resource.ResourceBaseInfo; import com.mec.ManyFile.resource.UnReceiveSection; import com.mec.ManyFile.Core.RafPool; import com.mec.ManyFile.resource.FileSection;/*** 處理每個發送端發來的文件片段* @author ty**/ public class DealReceive implements Runnable{private Socket socket;private DataInputStream dis;private static final int BUFFER_SIZE = 1 << 10;private boolean goon;private Resource resource;private List<FileSection> fileSectionPool;private Map<Integer, UnReceiveSection> unReceiveMap;DealReceive(Socket socket, Resource resource, Map<Integer, UnReceiveSection> unReceiveMap) {fileSectionPool = new LinkedList<FileSection>();this.resource = resource;this.socket = socket;this.unReceiveMap = unReceiveMap;try {dis = new DataInputStream(socket.getInputStream());} catch (IOException e) {e.printStackTrace();} new Thread(this).start();new Thread(new DealFileSection()).start();goon = true;}/*** 讀取len個字節，我們的緩沖區不一定能快速的收納len個字節* 采用以下方式能準確的讀取字節流* @param size* @return*/byte[] readBytes(int size) {int restLen = size;int readLen = 0;int len = size;int offset = 0;byte[] result = new byte[restLen];while(restLen > 0) {len = restLen < BUFFER_SIZE ? restLen : BUFFER_SIZE;try {readLen = dis.read(result, offset, len);restLen -= readLen;offset += readLen;} catch (IOException e) {goon = false;close();}}return result;}/*** 讀取一個文件片段* @return*/FileSection readFileSection() {FileSection fileSection = new FileSection();byte[] fileHand = readBytes(12);FileSectionHandle fileHandle = new FileSectionHandle(fileHand); byte[] value = readBytes(fileHandle.getLen());fileSection.setFileSectionHandle(fileHandle);fileSection.setValue(value);return fileSection;}void close() {if (dis != null) {try {dis.close();} catch (IOException e) {} finally {dis = null;}}if (socket != null) {try {socket.close();} catch (IOException e) {} finally {socket = null;}}}@Overridepublic void run() {while(goon) {FileSection fileSection = readFileSection();//每讀取一個，把他放入到緩沖區里，提高讀取效率，用另一個線程完成本地的寫fileSectionPool.add(fileSection);//dealFileSection(fileSection);}}/*** 將緩沖區中的文件片段根據資源信息慢慢的寫入到本地去* @author ty**/class DealFileSection implements Runnable{DealFileSection() {}@Overridepublic void run() {String absolutePath = resource.getAbsolutePath();RafPool rafPool = new RafPool();while(goon || !fileSectionPool.isEmpty()) {if (fileSectionPool.isEmpty()) {continue;}FileSection fileSection = fileSectionPool.remove(0);ResourceBaseInfo rbi = resource.getResourceBaseInfo(fileSection);String relativePath = rbi.getRelativePath();String filePath = absolutePath + "\\" + relativePath;RandomAccessFile raf = rafPool.get(filePath);readFileFromNative(raf, fileSection);}rafPool.closeAll();}private boolean readFileFromNative(RandomAccessFile raf, FileSection fileSection) {int fileHandle = fileSection.getFileHandle();int offset = fileSection.getOffSet();byte[] result = fileSection.getValue();try {raf.seek(offset);raf.write(result);UnReceiveSection unReceiveSection = unReceiveMap.get(fileHandle);unReceiveSection.addFileSection(fileSection);} catch (IOException e) {e.printStackTrace();}return true;}}}

總體來說，這寫內容有很多瑕疵，不過這已經榨干本博主能力的極限了，希望以后能在這條路上越走越遠！！！！

總結

以上是生活随笔為你收集整理的多文件云传输系统框架的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。