文档章节

.Net开发笔记(十四) 基于“泵”的UDP通信(接上篇)

IT周见智
 IT周见智
发布于 2015/06/05 17:17
字数 2478
阅读 4
收藏 0

上一篇中说到了“泵”在编程中的作用以及一些具体用处,但没有实际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周见智

IT周见智

粉丝 10
博文 61
码字总数 185891
作品 0
西青
TCP/UDP、IP 、Socket、HTTP笔记

1.1 TCP/IP协议组 TCP/IP协议(传输控制协议)由网络层的IP协议和传输层的TCP协议组成 IP层负责网络主机的定位,数据传输的路由,由IP地址可以唯一的确定Internet上的一台主机。 TCP层负责面向应...

X堇色
2017/06/26
0
0
Linux多线程并发服务器编程(线程池,FTP服务器)

分享网盘下载:https://pan.baidu.com/s/1gfNCcXt 密码: irfk 内容简介 本课程从最基础的进程、线程概念讲起逐步深入,通过理论与实践结合的方式,使学员快说掌握linux多线程网络编程技术,并...

人气王子333
06/26
0
0
CoAP协议学习笔记 2.2 阿里云IoT平台接入 基础版CoAP协议接入测试

这篇笔记涉及 阿里云CoAP接入,CoAP协议学习笔记可点此查看。 1 前言 上篇笔记完成了云端产品功能定义,并生成了相应的设备证书。 这篇笔记将会跟着教程走,使用 Ubuntu 对 SDK 编译验证,先...

iotisan
05/11
0
0
网络编程懒人入门(五):快速理解为什么说UDP有时比TCP更有优势

本文观点仅作参考,请根据自已系统的应用场景合理地选择数据传输层协议即可,无需盲目崇拜大牛言论。 1、前言 对于即时通讯开者新手来说,在开始着手编写IM或消息推送系统的代码前,最头疼的...

JackJiang2011
2017/12/19
0
0
Python就业班笔记整理(一)

什么是socket? socket(简称 套接字) 是进程间通信的一种方式,它与其他进程间通信的一个主要不同是:它能实现不同主机间的进程间通信,socket属于全双工。 关于创建套接字的参数说明: sock...

我是小谷粒
07/03
0
0

没有更多内容

加载失败,请刷新页面

加载更多

如何通过 J2Cache 实现分布式 session 存储

做 Java Web 开发的人多数都会需要使用到 session (会话),我们使用 session 来保存一些需要在两个不同的请求之间共享数据。一般 Java 的 Web 容器像 Tomcat、Resin、Jetty 等等,它们会在...

红薯
今天
1
0
C++ std::thread

C++11提供了std::thread类来表示一个多线程对象。 1,首先介绍一下std::this_thread命名空间: (1)std::this_thread::get_id():返回当前线程id (2)std::this_thread::yield():用户接口...

yepanl
今天
2
0
Nignx缓存文件与动态文件自动均衡的配置

下面这段nginx的配置脚本的作用是,自动判断是否存在缓存文件,如果有优先输出缓存文件,不经过php,如果没有,则回到php去处理,同时生成缓存文件。 PHP框架是ThinkPHP,最后一个rewrite有关...

swingcoder
今天
1
0
20180920 usermod命令与用户密码管理

命令 usermod usermod 命令的选项和 useradd 差不多。 一个用户可以属于多个组,但是gid只有一个;除了gid,其他的组(groups)叫做扩展组。 usermod -u 1010 username # 更改用户idusermod ...

野雪球
今天
1
0
Java网络编程基础

1. 简单了解网络通信协议TCP/IP网络模型相关名词 应用层(HTTP,FTP,DNS等) 传输层(TCP,UDP) 网络层(IP,ICMP等) 链路层(驱动程序,接口等) 链路层:用于定义物理传输通道,通常是对...

江左煤郎
今天
1
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部