.Net开发笔记(十四) 基于“泵”的UDP通信(接上篇)
.Net开发笔记(十四) 基于“泵”的UDP通信(接上篇)
IT周见智 发表于3年前
.Net开发笔记(十四) 基于“泵”的UDP通信(接上篇)
  • 发表于 3年前
  • 阅读 3
  • 收藏 0
  • 点赞 0
  • 评论 0

新睿云服务器60天免费使用,快来体验!>>>   

上一篇中说到了“泵”在编程中的作用以及一些具体用处,但没有实际demo,可能不好理解,这篇文章我分享一个UDP通信的demo,大概实现了类似“飞鸽传书”在局域网中文本消息和文件传输的功能。功能不全也不是很完善,但足以说明“泵”在代码中的具体应用。

先来回忆一下上篇中“泵”的含义,首先它是可持续运作的,其次它可以将“数据”从一个地方传递到另外一个地方,供其他人使用。搬一张上篇的图:

图1 程序中“泵”结构图

如上图所示,每个泵有一个“待处理”的数据容器(缓冲区),泵循环里面还可以有一个“预处理”的地方,可以在数据被传递之前,先做处理,最后,泵会输出数据,供别人使用。完成“数据源”->“泵”->“使用者”这样一个过程,明显“泵”起到动力的作用。我们可以看见,“使用者”使用数据的过程包含在“泵循环”之内,也就是说,如果使用数据时,耗时太长,一次“泵循环”不能及时返回,那么,缓冲区中的数据就会累积,得不到及时处理,影响“泵”的工作效率。能解决该问题有两种途径:

  • 1)在使用数据时,不做长时间耗时操作;
  • 2)先不直接使用数据,先将“泵”传递出来的数据存入另外一个容器(缓冲区),再另外开辟线程去处理缓冲区中的数据。

其中1)明显治标不治本,不能解决实际问题,2)倒是可以一试,因为它将“使用数据”和“泵循环”分离开来,使“使用数据”不会直接影响到“泵”的工作。其实上一篇中说到“UDP通信结构”时就已经按照2)所说的实现的,我那时候说不能将数据处理、数据分析都放在“数据接收泵”中,因为有一个主要原因就是,那样做会影响数据接收的效率。后来我画了一张UDP通信结构图,里面用到了三个“泵”(数据接收泵、数据分析泵、数据处理泵),如下图:

图2 UDP通信中“泵”的使用

如上图,“数据分析泵”使用到了“数据接收泵”传递出来的数据,“数据处理泵”使用到了“数据分析泵”传递出来的数据,它们都没有直接使用,都是先将数据存入对应缓冲区,然后开辟新的“泵”(也是另开辟线程吧)去处理。这样一来,每个环节互不影响其他环节的工作效率。

在实现这个UDP通信demo时,我定义了三个“泵”类,一个数据发送类,和一个帮助类,分别如下:

  • UDPSocket:主要实现了数据接受泵,以及socket绑定等;
  • DataAnalyse:数据分析泵;
  • DataDeal:数据处理泵;
  • UDPSender:数据发送;
  • Help.cs中一些类型:辅助功能。

其中,DataAnalyse是UDPSocket中“泵”传递出来数据的使用者,DataDeal是DataAnalyse中“泵”传递出来数据的使用者,然后我们(开发者)是DataDeal中“泵”传递出来数据的使用者。

需要说明的是,默认请款下,DataDeal中的“泵”是不会传递出来任何数据,DataAnalyse中的“泵”也不会传递出来任何数据,我们需要根据具体需求,实现自己的DataDeal和DataAnalyse类,重写Deal和Analyse虚方法。我定义了两个类,分别为MyDataDeal和MyDataAnalyse,具体代码如下:

/// <summary>
    /// 自定义数据处理类 重写Deal方法 /// 负责处理某一些数据,其余交给基类 /// 可以继续从MyDataDeal派生新的类,继续重写Deal方法 /// </summary>
    public class MyDataDeal : DataDeal { public event LoginEventHandler Login; public event LoginBackEventHandler LoginBack; public event TxtMessageReceivedEventHandler TxtMessageReceived; public event TxtMessageBackEventHandler TxtMessageBack; public event LogoutEventHandler Logout; public event AskSendFileEventHandler AskSendFile; public event AskSendFileBackEventHandler AskSendFileBack; public event SendFileDataEventHandler SendFileData; public event ReceiveFileOKEventHandler ReceiveFileOK; public event SendFileStopEventHandler SendFileStop; public event ReceiveFileStopEventHandler ReceiveFileStop; public event SendFileDataBackEventHandler SendFileDataBack; public MyDataDeal(DataAnalyse dataAnalyse) : base(dataAnalyse) { } protected override void Deal(Data data) { switch (data.Msg) { case Msg.ZMsg1: { if (Login != null) { Login(data.RemoteIP, data.RemotePort, data.Content.ToString()); } break; } case Msg.ZMsg2: { if (LoginBack != null) { LoginBack(data.RemoteIP, data.RemotePort, data.Content.ToString()); } break; } case Msg.ZMsg3: { if (TxtMessageReceived != null) { TxtMessageReceived(data.RemoteIP, data.RemotePort, data.Content.ToString()); } break; } case Msg.ZMsg4: { if(TxtMessageBack != null) { TxtMessageBack(data.RemoteIP, data.RemotePort, data.Content.ToString()); } break; } case Msg.ZMsg5: { if (Logout != null) { Logout(data.RemoteIP, data.RemotePort, data.Content.ToString()); } break; } case Msg.ZMsg6: { if (AskSendFile != null) { object[] objs = data.Content as object[]; int uid = (int)objs[0]; long length = (long)objs[1]; string filename = objs[2].ToString(); AskSendFile(data.RemoteIP, data.RemotePort, filename, length, uid); } break; } case Msg.ZMsg7: { if (AskSendFileBack != null) { object[] objs = data.Content as object[]; int uid = (int)objs[0]; int result = (int)objs[1]; AskSendFileBack(data.RemoteIP, data.RemotePort, result, uid); } break; } case Msg.ZMsg8: { if (SendFileData != null) { object[] objs = data.Content as object[]; int uid = (int)objs[0]; int part = (int)objs[1]; byte[] dat = objs[2] as byte[]; SendFileData(data.RemoteIP, data.RemotePort, dat, uid, part); } break; } case Msg.ZMsg9: { if (ReceiveFileOK != null) { int uid = (int)data.Content; ReceiveFileOK(data.RemoteIP, data.RemotePort, uid); } break; } case Msg.ZMsg10: { if (SendFileStop != null) { int uid = (int)data.Content; SendFileStop(data.RemoteIP, data.RemotePort, uid); } break; } case Msg.ZMsg11: { if (ReceiveFileStop != null) { int uid = (int)data.Content; ReceiveFileStop(data.RemoteIP, data.RemotePort, uid); } break; } case Msg.ZMsg12: { if (SendFileDataBack != null) { object[] objs = data.Content as object[]; int uid = (int)objs[0]; int part = (int)objs[1]; SendFileDataBack(data.RemoteIP, data.RemotePort, uid, part); } break; } default: { base.Deal(data); //其他由基类处理
                        break; } } } } public delegate void LoginEventHandler(string remoteIP,int remotePort,string remoteName); public delegate void LoginBackEventHandler(string remoteIP,int remotePort,string remoteName); public delegate void TxtMessageReceivedEventHandler(string remoteIP,int remotePort,string txtMsg); public delegate void TxtMessageBackEventHandler(string remoteIP,int remotePort,string txtMsgBack); public delegate void LogoutEventHandler(string remoteIP,int remotePort,string remoteName); public delegate void AskSendFileEventHandler(string remoteIP,int remotePort,string filename,long length,int uid); public delegate void AskSendFileBackEventHandler(string remoteIP,int remotePort,int result,int uid); public delegate void SendFileDataEventHandler(string remoteIP,int remotePort,byte[] data,int uid,int part); public delegate void ReceiveFileOKEventHandler(string remoteIP,int remotePort,int uid); public delegate void SendFileStopEventHandler(string remoteIP,int remotePort,int uid); public delegate void ReceiveFileStopEventHandler(string remoteIP,int remotePort,int uid); public delegate void SendFileDataBackEventHandler(string remoteIP,int remotePort,int uid,int part);
MyDataDeal
 1     /// <summary>
 2     /// 自定义数据分析类 重写Analyse方法  3     /// 负责分析某一些数据,其余交给基类  4     /// 可以继续从MyDataAnalyse派生新的类,继续重写Analyse方法  5     /// </summary>
 6     public class MyDataAnalyse : DataAnalyse  7  {  8         public MyDataAnalyse(UDPSocket udpSocket)  9             : base(udpSocket)  10  {  11 
 12  }  13         protected override void Analyse(RawData rawData)  14  {  15             string ip = (rawData.RemoteEP as IPEndPoint).Address.ToString(); //远端IP
 16             int port = (rawData.RemoteEP as IPEndPoint).Port; //远端端口
 17             if (rawData.Buffer.Length > 0)  18  {  19                 Msg msg = (Msg)(rawData.Buffer[0]); //
 20                 switch (msg)  21  {  22                     case Msg.ZMsg1: //登录
 23  {  24                             string data = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);  //从1开始 因为前面有一个 头 下同,注意是 rawData.Length(实际接收数据) 不是rawData.Buffer.Length
 25  OnAnalysed(msg, data, ip, port);  26                             break;  27  }  28                     case Msg.ZMsg2: //登录反馈
 29  {  30                             string data = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);  31  OnAnalysed(msg, data, ip, port);  32                             break;  33  }  34                     case Msg.ZMsg3: //文本信息
 35  {  36                             string data = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);  37  OnAnalysed(msg, data, ip, port);  38                             break;  39  }  40                     case Msg.ZMsg4: //文本信息反馈
 41  {  42                             string data = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);  43  OnAnalysed(msg, data, ip, port);  44                             break;  45  }  46                     case Msg.ZMsg5: //退出
 47  {  48                             string data = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);  49  OnAnalysed(msg, data, ip, port);  50                             break;  51  }  52                     case Msg.ZMsg6: //请求发送文件
 53  {  54                             int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
 55                             long length = BitConverter.ToInt64(rawData.Buffer, 5); //文件大小
 56                             string filename = Encoding.Unicode.GetString(rawData.Buffer, 13, rawData.Length - 13); //文件名称
 57                             OnAnalysed(msg, new object[] { uid, length, filename }, ip, port); //  58                             break;  59  }  60                     case Msg.ZMsg7: //请求发送文件反馈
 61  {  62                             int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
 63                             int result = BitConverter.ToInt32(rawData.Buffer, 5); //反馈结果
 64                             OnAnalysed(msg, new object[] { uid, result }, ip, port);  65                             break;  66  }  67                     case Msg.ZMsg8: //发送文件数据
 68  {  69                             int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
 70                             int part = BitConverter.ToInt32(rawData.Buffer, 5); //段ID
 71                             byte[] data = new byte[rawData.Length - 9]; //实际文件数据
 72                             Buffer.BlockCopy(rawData.Buffer, 9, data, 0, data.Length);  73                             OnAnalysed(msg, new object[] { uid, part, data }, ip, port);  74                             break;  75  }  76                     case Msg.ZMsg9: //接收文件完毕
 77  {  78                             int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
 79  OnAnalysed(msg, uid, ip, port);  80                             break;  81  }  82                     case Msg.ZMsg10: //发送方中止发送文件
 83  {  84                             int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
 85  OnAnalysed(msg, uid, ip, port);  86                             break;  87  }  88                     case Msg.ZMsg11: //接收方中止接收文件
 89  {  90                             int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
 91  OnAnalysed(msg, uid, ip, port);  92                             break;  93  }  94                     case Msg.ZMsg12: //发送文件数据反馈
 95  {  96                             int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
 97                             int part = BitConverter.ToInt32(rawData.Buffer, 5); //段ID
 98                             OnAnalysed(msg, new object[] { uid, part }, ip, port);  99                             break; 100  } 101                     default: 102  { 103                             base.Analyse(rawData);  //其他由基类分析
104                             break; 105  } 106  } 107  } 108  } 109     }
MyDataAnalyse

如代码所示,各位下载源码后,可以将UDPSocket、UDPSender、DataDeal、DataAnalyse以及Help.cs文件中的一些类型封装起来,生成一个通用程序集,实际使用中,只需要引用该程序集,并且实现自己的DataDeal和DataAnalyse就行。

另外,demo中我还定义了一套自己的“通信协议”,如下:

 1 ZMsg1:用户新上线                  头+用户信息  2 ZMsg2:用户新上线反馈              头+反馈用户信息  3 ZMsg3:文本消息                    头+密码+消息正文  4 ZMsg4:文本消息反馈                头+原始消息  5 ZMsg5:离线                        头+任意文本  6 ZMsg6:请求发送文件                头+发送文件的唯一标示+大小+文件名  7 ZMsg7:请求发送文件反馈            头+文件的唯一标示+1或0 1代表接收 0代表拒绝  8 ZMsg8:发送文件数据                头+文件唯一标示+段ID+文件数据(byte[])  9 ZMsg9:接收文件完毕                头+文件唯一标示 10 ZMsg10:发送方中止发送文件         头+文件唯一标示 11 ZMsg11:接收方中止接收文件         头+文件唯一标示 12 ZMsg12:发送文件数据反馈           头+文件唯一标示+段ID
消息协议

通信协议根据具体需求而不同。

总结一下,在其他项目中具体使用步骤如下:

  1. 根据具体需求,定义一套“消息协议”;
  2. 从DataAnalyse派生出一个新的类,根据“消息协议”重写Analyse方法,进行数据分析;
  3. 从DataDeal派生出一个新的类,根据具体业务逻辑重写Deal方法,进行数据处理;
  4. 数据发送方必须严格按照“消息协议”发送数据;
  5. 最后,你要做的就是注册DataDeal派生类的一些事件,使用“泵”传递出来的程序可识别数据。

源码下载地址:http://files.cnblogs.com/xiaozhi_5638/UDPMessager.rar

希望对各位有帮助。附几张效果图

  • 打赏
  • 点赞
  • 收藏
  • 分享
共有 人打赏支持
IT周见智
粉丝 10
博文 61
码字总数 185891
×
IT周见智
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: