在網上看過許多程式（QQ、Azureus、Ants、PPStream）都實作了 P2P，以前覺得技術很高深。透過這段時間學習才發現，單純實作 P2P 在區域網路通訊很容易，但要實現外網穿透 (NAT) 感覺很困難。最近看了 Azureus 和 eMule 原始碼，分別是 Java 和 C++ 版，對這兩門語言都不熟悉，看起來很吃力。最後只好根據 VC++ 實作的 P2PDemo 程式改版，依設計思路用 C# 寫了一個 Demo。經測試，多個客戶端在區域網路能脫離服務端實現端到端工作。外網情況要透過路由器，看 Azureus 要用 uPnP 進行連接埠對應，在 CodeProject 下載 uPnP 原始碼測試，結果沒啟用 uPnP 路由器。現在郁悶了，不知道下一步怎麼測試，是否用 uPnP 實現連接埠自動對應成功就能像 QQ 那樣通訊。
Packet.cs - 基礎封包類別
[Serializable()]
public abstract class Packet {
public virtual int GetCommandType() { return -1; }
public string UserName { get; set; }
public Packet() { }
public Packet(string username) { this.UserName = username; }
}
MassTextPacket.cs - 分片傳輸類別
[Serializable()]
public class MassTextPacket : TextPacket {
public int SeqID { get; set; } // 封包序列
public int SeqCount { get; set; } // 封包數量
public int CLSD { get; set; } // 識別碼
}
P2P 訊息發送
private bool SendMessageTo(string toUserName, Packet packet) {
PeerEntity toUser = userList.Single(c => c.UserName == toUserName);
if (toUser == null) return false;
ReceivedACK = false;
for (int i = 0; i < MAXRETRY; i++) {
// 若對方 P2P 位址有效，嘗試直接發送
if (toUser.P2PAddress != null && toUser.P2PAddress.Port != 0) {
if (packet.GetType() == typeof(TextPacket)) {
// 文字訊息發送邏輯，支援大訊息分片
TextPacket msgPacket = new TextPacket(toUserName, (packet as TextPacket).Message);
byte[] buffer = UtilityHelper.Serialize(msgPacket);
if (buffer.Length > MAXBUFFERSIZE) {
// 分片傳輸邏輯
MassTextPacket mtp = new MassTextPacket();
mtp.SeqID = 0;
mtp.SeqCount = (int)Math.Ceiling(buffer.Length / (decimal)MAXBUFFERSIZE);
mtp.CLSD = mtp.GetHashCode();
long pos = 0;
long count = buffer.Length < MAXBUFFERSIZE ? buffer.Length : MAXBUFFERSIZE;
while (pos < buffer.Length && pos > 0) {
byte[] bytes = new byte[count];
for (int k = 0; k < count; k++)
bytes[k] = buffer[pos + k];
mtp.SeqID++;
mtp.Message = Convert.ToBase64String(bytes);
byte[] buf = UtilityHelper.Serialize(mtp);
client.Send(buf, buf.Length, toUser.P2PAddress);
Thread.Sleep(100);
}
} else {
client.Send(buffer, buffer.Length, toUser.P2PAddress);
}
} else if (packet.GetType() == typeof(FileStreamPacket)) {
// 檔案傳輸邏輯，支援分片與進度控制
FileStreamPacket fsp = packet as FileStreamPacket;
using (FileStream fs = new FileStream(fsp.FileName, FileMode.Open, FileAccess.Read, FileShare.Read)) {
handle1.Reset();
fsp.SeqID = 0;
fsp.SeqCount = (int)Math.Ceiling(fs.Length / (decimal)MAXBUFFERSIZE);
fsp.CLSD = fsp.GetHashCode();
long pos = 0;
long count = fs.Length < MAXBUFFERSIZE ? fs.Length : MAXBUFFERSIZE;
while (pos < fs.Length && count > 0) {
byte[] buffer = new byte[count];
fs.Seek(pos, SeekOrigin.Begin);
fs.Read(buffer, 0, (int)count);
pos += count;
count = pos + MAXBUFFERSIZE < fs.Length ? MAXBUFFERSIZE : fs.Length - pos;
fsp.SeqID++;
fsp.Message = Convert.ToBase64String(buffer);
byte[] buf = UtilityHelper.Serialize(fsp);
client.Send(buf, buf.Length, toUser.P2PAddress);
Thread.Sleep(300);
}
handle1.Set();
}
}
// 等待 ACK 確認
for (int j = 0; j < 10; j++) {
if (this.ReceivedACK) {
this.ReceivedACK = false;
return true;
}
Thread.Sleep(300);
}
}
// 建立 P2P 打洞封包，透過伺服器轉發請求對方向自己打洞
P2PConnectionPacket transMsg = new P2PConnectionPacket(UserName, toUserName);
byte[] msgBuffer = UtilityHelper.Serialize(transMsg);
client.Send(msgBuffer, msgBuffer.Length, hostPoint);
// 等待對方 P2PCONNECTACK 訊息
for (int j = 0; j < 10; ++j) {
toUser = userList.Single(c => c.UserName == toUserName);
if (toUser.P2PAddress != null && toUser.P2PAddress.Port != 0)
break;
Thread.Sleep(300);
}
}
return false;
}
訊息接收執行緒
private void RecvThreadProc() {
byte[] buffer;
while (true) {
buffer = client.Receive(ref remotePoint);
Packet packet = UtilityHelper.Deserialize(buffer) as Packet;
Type msgType = packet.GetType();
if (msgType == typeof(UserListAckPacket)) {
// 更新使用者列表
UserListAckPacket usersMsg = (UserListAckPacket)packet;
userList.Clear();
foreach (PeerEntity user in usersMsg.Users)
userList.Add(user);
bUserListComplete = true;
}
else if (msgType == typeof(UserLoginAckPacket)) {
ProcUserLogAckMsg(packet);
}
else if (msgType == typeof(TextPacket)) {
TextPacket txtPacket = (TextPacket)packet;
printf("Receive a message: {0}", txtPacket.Message);
// 發送 ACK
P2PAckPacket ackMsg = new P2PAckPacket();
buffer = UtilityHelper.Serialize(ackMsg);
client.Send(buffer, buffer.Length, remotePoint);
}
else if (msgType == typeof(MassTextPacket)) {
lock (this) {
MassTextPacket fPacket = (MassTextPacket)packet;
if (packets.ContainsKey(fPacket.CLSD))
packets[fPacket.CLSD].Add(fPacket);
else
packets.Add(fPacket.CLSD, new List() { fPacket });
printf("PacketID:{0}--SeqNo:{1}--progress: {2}%", fPacket.CLSD, fPacket.SeqID,
(int)(Math.Round(packets[fPacket.CLSD].Count / (decimal)(fPacket as MassTextPacket).SeqCount, 2) * 100));
// 組裝完整訊息
if ((fPacket as MassTextPacket).SeqCount == packets[fPacket.CLSD].Count) {
List temp = packets[fPacket.CLSD].OrderBy(c => c.SeqID).ToList();
List values = new List();
foreach (MassTextPacket mt in temp) {
byte[] buf = Convert.FromBase64String(mt.Message);
values.AddRange(buf);
}
MassTextPacket value = UtilityHelper.Deserialize(values.ToArray()) as MassTextPacket;
printf("Receive a message: {0}", value.Message);
// 發送 ACK
P2PAckPacket ackMsg = new P2PAckPacket();
buffer = UtilityHelper.Serialize(ackMsg);
client.Send(buffer, buffer.Length, remotePoint);
}
}
}
else if (msgType == typeof(FileStreamPacket)) {
// 檔案接收與重組邏輯
lock (this) {
FileStreamPacket fPacket = (FileStreamPacket)packet;
if (packets.ContainsKey(fPacket.CLSD))
packets[fPacket.CLSD].Add(fPacket);
else
packets.Add(fPacket.CLSD, new List() { fPacket });
// 進度顯示與檔案重組...
if ((fPacket as FileStreamPacket).SeqCount == packets[fPacket.CLSD].Count) {
// 寫入檔案、發送 ACK 等
}
}
}
else if (msgType == typeof(P2PAckPacket)) {
this.ReceivedACK = true;
}
else if (msgType == typeof(P2PPurchHolePacket)) {
ProcP2PPurchHoleMsg(packet, remotePoint);
}
else if (msgType == typeof(P2PPurchHoleAckPacket)) {
PeerEntity touser = userList.SingleOrDefault(c => c.UserName == (packet as P2PPurchHoleAckPacket).ToUserName);
touser.P2PAddress = touser.RemoteEndPoint;
}
Thread.Sleep(100);
}
}
建立 P2P 會話（打洞邏輯）
private void ProcP2PPurchHoleMsg(Packet packet, IPEndPoint remoteEP) {
P2PPurchHolePacket purchReqMsg = (P2PPurchHolePacket)packet;
PeerEntity toUser = userList.Single(c => c.UserName == purchReqMsg.ToUserName);
PeerEntity user = userList.Single(c => c.UserName == purchReqMsg.UserName);
toUser.P2PAddress = toUser.RemoteEndPoint;
printf("Set P2P Address for {0}->[{1}]", user.UserName, toUser.P2PAddress.ToString());
// uPnP 連接埠對應
if (NAT.AddPortMapping(toUser.P2PAddress.Port, ProtocolType.Udp, "AddPortMapping"))
printf("Port mapping successed!");
// 發送打洞訊息到遠端主機
P2PPurchHoleAckPacket trashMsg = new P2PPurchHoleAckPacket(purchReqMsg.UserName, purchReqMsg.ToUserName);
byte[] buffer = UtilityHelper.Serialize(trashMsg);
client.Send(buffer, buffer.Length, user.RemoteEndPoint);
}
訊息處理執行緒
private void RecvThreadProc() {
IPEndPoint remotePoint = null;
byte[] msgBuffer = null;
while (true) {
msgBuffer = server.Receive(ref remotePoint);
try {
object msgObj = UtilityHelper.Deserialize(msgBuffer);
switch ((msgObj as Packet).GetCommandType()) {
case Command.MSG_USERLOGIN:
ProcUserLoginMsg(msgObj as UserLoginPacket, remotePoint);
break;
case Command.MSG_USERLOGOUT:
ProcUserLogoutMsg(msgObj as UserLogoutPacket, remotePoint);
break;
case Command.MSG_GETUSERLIST:
ProcGetUserListMsg(msgObj as UserListPacket, remotePoint);
break;
case Command.MSG_P2PCONNECT:
ProcP2PConnectMsg(msgObj as P2PConnectionPacket, remotePoint);
break;
case Command.MSG_USERACTIVEQUERY:
ProcUserActiveQueryMsg(msgObj as UserActiveQueryPacket, remotePoint);
break;
}
Thread.Sleep(100);
} catch { }
}
}
伺服端轉發 P2P 連線請求
private void ProcP2PConnectMsg(Packet packet, IPEndPoint remoteEP) {
P2PConnectionPacket transMsg = (P2PConnectionPacket)packet;
printf("{0}({1}) wants to p2p {2}", remoteEP.Address.ToString(), transMsg.UserName, transMsg.ToUserName);
PeerEntity toUser = userList.SingleOrDefault(c => c.UserName == transMsg.ToUserName);
// 轉發打洞請求
P2PPurchHolePacket transMsg2 = new P2PPurchHolePacket(transMsg.UserName, toUser.UserName);
byte[] buffer = UtilityHelper.Serialize(transMsg2);
server.Send(buffer, buffer.Length, toUser.RemoteEndPoint);
}
希望大家共同討論、共同進步！本文附配套原始碼供下載參考。
