广播与P2P通道（下） -- 方案实现
Posted 在西天取经的路上……
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了广播与P2P通道（下） -- 方案实现相关的知识，希望对你有一定的参考价值。
在广播与P2P通道（上） -- 问题与方案 一文中，我们已经找到了最优的模型，即将广播与P2P通道相结合的方案，这样能使服务器的带宽消耗降到最低，最大节省服务器的宽带支出。当然，如果从零开始实现这种方案无疑是非常艰巨的，但基于ESFramework提供的通信功能和P2P功能来做，就不再那么遥不可及了。
1.P2P通道状态
根据上文模型3的讨论，要实现该模型，每个客户端需要知道自己与哪些用户创建了P2P通道，服务器也要知道每个客户端已建立的P2P通道的状态。
使用ESFramework，在客户端已经可以通过IRapidPassiveEngine.P2PController接口知道当前客户端与哪些其它客户端成功建立了P2P通道，并且可以通过P2PController接口发起与新的客户端建立新的P2P通道的尝试。但在服务端，对于每个客户端建立了哪些P2P通道，服务端是一无所知的。所以，基于ESFramework实现模型3的第一件事情，就是客户端要实时把自己的P2P状态变化报告给服务端，而服务端也要管理每个客户端的P2P通道状态。（注意。下面的所有实现，需要引用ESFramework.dll、ESPlus.dll、ESBasic.dll）
（1）P2PChannelManager
我们在服务端设计P2PChannelManager类来管理每个在线客户端已成功创建的所有P2P通道。
public class P2PChannelManager
{
//key 表示P2P通道的起始点用户ID，value 表示P2P通道的目的点用户列表。（单向，因为某些P2P通道就是单向的）
private SortedArray<string, SortedArray<string>> channels = new SortedArray<string, SortedArray<string>>();
public void Initialize(IUserManager userManager)
{
userManager.SomeOneDisconnected += new ESBasic.CbGeneric<UserData, ESFramework.Server.DisconnectedType>(userManager_SomeOneDisconnected);
}
void userManager_SomeOneDisconnected(UserData user, ESFramework.Server.DisconnectedType obj2)
{
this.channels.RemoveByKey(user.UserID);
}
public void Register(string startUserID, string destUserID)
{
if (!this.channels.ContainsKey(startUserID))
{
this.channels.Add(startUserID, new SortedArray<string>());
}
this.channels[startUserID].Add(destUserID);
}
public void Unregister(string startUserID, string destUserID)
{
if (this.channels.ContainsKey(startUserID))
{
this.channels[startUserID].Remove(destUserID);
}
}
public bool IsP2PChannelExist(string startUserID, string destUserID)
{
if (!this.channels.ContainsKey(startUserID))
{
return false;
}
return this.channels[startUserID].Contains(destUserID);
}
}
P2PChannelManager提供了注册P2P通道、注销P2P通道、以及查询P2P通道是否存在的方法。其内部使用类似字典的SortedArray来管理每个用户的已经成功建立的P2P通道（即与哪些其它用户打通了P2P）。另外，P2PChannelManager预定了IUserManager的SomeOneDisconnected事件，这样，当某个用户掉线时，就可以清除其所有的P2P状态。因为，在ESFramework中，当客户端与服务器的TCP连接断开时，客户端会自动关闭所有的P2P通道。
（2）客户端实时报告自己的P2P状态变化给服务端
当客户端每次成功创建一个P2P通道、或者已有P2P通道中断时，客户端要发消息告诉服务端。这样，我们就需要定义这个消息的类型：
public static class MyInfoTypes
{
public const int P2PChannelOpen = 1;
public const int P2PChannelClose = 2;
}
再定义消息协议：
public class P2PChannelReportContract
{
public P2PChannelReportContract() { }
public P2PChannelReportContract(string dest)
{
this.destUserID = dest;
}
#region DestUserID
private string destUserID;
public string DestUserID
{
get { return destUserID; }
set { destUserID = value; }
}
#endregion
}
定好了消息类型和contract类，我们在客户端预定P2P通道的状态变化，并报告给服务端：
public void Initialize(IRapidPassiveEngine rapidPassiveEngine)
{
rapidPassiveEngine.P2PController.P2PChannelOpened += new CbGeneric<P2PChannelState>(P2PController_P2PChannelOpened);
rapidPassiveEngine.P2PController.P2PChannelClosed += new CbGeneric<P2PChannelState>(P2PController_P2PChannelClosed);
}
void P2PController_P2PChannelClosed(P2PChannelState state)
{
this.P2PChannelReport(false, state.DestUserID);
}
void P2PController_P2PChannelOpened(P2PChannelState state)
{
this.P2PChannelReport(true, state.DestUserID);
}
private void P2PChannelReport(bool open, string destUserID)
{
P2PChannelReportContract contract = new P2PChannelReportContract(destUserID);
int messageType = open ? MyInfoTypes.P2PChannelOpen : MyInfoTypes.P2PChannelClose;
this.rapidPassiveEngine.CustomizeOutter.Send(messageType, CompactPropertySerializer.Default.Serialize(contract));
}
在服务端，我们需要处理这两种类型的消息（实现ICustomizeHandler接口的HandleInformation方法）：
private P2PChannelManager p2PChannelManager = new P2PChannelManager();
public void HandleInformation(string sourceUserID, int informationType, byte[] information)
{
if (informationType == MyInfoTypes.P2PChannelOpen)
{
P2PChannelReportContract contract = CompactPropertySerializer.Default.Deserialize<P2PChannelReportContract>(information, 0);
this.p2PChannelManager.Register(sourceUserID, contract.DestUserID);
return ;
}
if (informationType == MyInfoTypes.P2PChannelClose)
{
P2PChannelReportContract contract = CompactPropertySerializer.Default.Deserialize<P2PChannelReportContract>(information, 0);
this.p2PChannelManager.Unregister(sourceUserID, contract.DestUserID);
return ;
}
｝
这样，服务端就实时地知道每个客户端的P2P状态了。
2.与广播结合
同样的，我们首先为广播消息定义一个消息类型：
public static class MyInfoTypes
{
public const int P2PChannelOpen = 1;
public const int P2PChannelClose = 2;
public const int Broadcast = 3; //广播消息
}
再定义对应的协议类：
public class BroadcastContract
{
#region Ctor
public BroadcastContract() { }
public BroadcastContract(string _broadcasterID, string _groupID, int infoType ,byte[] info )
{
this.broadcasterID = _broadcasterID;
this.groupID = _groupID;
this.content = info;
this.informationType = infoType;
this.actionTypeOnChannelIsBusy = action;
}
#endregion
#region BroadcasterID
private string broadcasterID = null;
/// <summary>
/// 发出广播的用户ID。
/// </summary>
public string BroadcasterID
{
get { return broadcasterID; }
set { broadcasterID = value; }
}
#endregion
#region GroupID
private string groupID = "";
/// <summary>
/// 接收广播的组ID
/// </summary>
public string GroupID
{
get { return groupID; }
set { groupID = value; }
}
#endregion
#region InformationType
private int informationType = 0;
/// <summary>
/// 广播信息的类型。
/// </summary>
public int InformationType
{
get { return informationType; }
set { informationType = value; }
}
#endregion
#region Content
private byte[] content;
public byte[] Content
{
get { return content; }
set { content = value; }
}
#endregion
}
（1）在客户端发送广播消息
在客户端，我们根据与组内成员的P2P通道的状态，来判断发送的方案，就像依据上文提到的，可细分为三种情况：
a.当某个客户端发现自己和组内的所有其它成员都建立了P2P通道时，那么，它就不用把广播消息发送给服务器了。
b.如果客户端与组内的所有其它成员的P2P通道都没有建立成功，那么，它只需要将广播消息发送给服务器。
c.如果客户端与部分组内的成员建立了P2P通道，那么，它不仅需要将广播消息发送给服务器，还需要将该广播消息经过每个P2P通道发送一次。
public void Broadcast(string currentUserID, string groupID, int broadcastType, byte[] broadcastContent)
{
BroadcastContract contract = new BroadcastContract(currentUserID, groupID, broadcastType, broadcastContent);
byte[] info = CompactPropertySerializer.Default.Serialize(contract);
List<string> members = this.groupManager.GetGroupMembers(groupID);
if (members == null)
{
return;
}
bool allP2P = true;
foreach (string memberID in members)
{
if (memberID == this.currentUserID)
{
continue;
}
if (rapidPassiveEngine.P2PController.IsP2PChannelExist(memberID))
{
rapidPassiveEngine.CustomizeOutter.SendByP2PChannel(memberID, MyInfoTypes.Broadcast, info, ActionTypeOnNoP2PChannel.Discard, true, ActionTypeOnChannelIsBusy.Continue);
}
else
{
allP2P = false;
}
}
if (!allP2P) //只要有一个组成员没有成功建立P2P，就要发给服务端。
{
this.rapidPassiveEngine.CustomizeOutter.Send(null, this.groupInfoTypes.Broadcast, info, true, action);
}
}
（2）服务端转发广播
当服务器收到一个广播消息时，首先，查看目标组中的用户，然后，根据广播消息的发送者的P2P通道状态，来综合决定该广播消息需要转发给哪些客户端。我们只需在上面的HandleInformation方法中增加代码就可以了：
if (informationType == MyInfoTypes.Broadcast)
{
BroadcastContract contract = CompactPropertySerializer.Default.Deserialize<BroadcastContract>(information, 0);
string groupID = contract.GroupID;
List<string> members = this.groupManager.GetGroupMembers(groupID);
if (members != null)
{
foreach (string memberID in members)
{
bool useP2PChannel = this.p2PChannelManager.IsP2PChannelExist(sourceUserID, memberID);
if (memberID != sourceUserID && !useP2PChannel)
{
this.customizeController.Send(memberID, MyInfoTypes.Broadcast, information, true, ActionTypeOnChannelIsBusy.Continue);
}
}
}
return;
}
（3）客户端处理接收到的广播消息
客户端也只要实现ICustomizeHandler接口的HandleInformation方法，就可以处理来自P2P通道或者转发自服务端的广播消息了（即处理MyInfoTypes.Broadcast类型的消息），这里就不赘述了。
实际上，本文的实现还可以进一步优化，特别是在高频的广播消息时（如前文举的视频会议的例子），这种优化效果是很明显的。那就是，比如，我们在客户端可以将组内的成员分成两类管理起来，一类是P2P已经打通的，一类是没有通的，并根据实际的P2P状态变化而调整。这样，客户端每次发送广播消息时，就不用遍历自己与每个组员的P2P通道的状态，这可以节省不少的cpu时间。同理，服务端也可以如此处理。
以上是关于广播与P2P通道（下） -- 方案实现的主要内容，如果未能解决你的问题，请参考以下文章
SpringCloud系列十一：SpringCloudStream（SpringCloudStream 简介创建消息生产者创建消息消费者自定义消息通道分组与持久化设置 RoutingKey）(代码片段