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

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

    /// <summary>
    /// 自定义数据分析类 重写Analyse方法
    /// 负责分析某一些数据，其余交给基类
    /// 可以继续从MyDataAnalyse派生新的类，继续重写Analyse方法
    /// </summary>
    public class MyDataAnalyse : DataAnalyse
    {
        public MyDataAnalyse(UDPSocket udpSocket)
            : base(udpSocket)
        {

        }
        protected override void Analyse(RawData rawData)
        {
            string ip = (rawData.RemoteEP as IPEndPoint).Address.ToString(); //远端IP
            int port = (rawData.RemoteEP as IPEndPoint).Port; //远端端口
            if (rawData.Buffer.Length > 0)
            {
                Msg msg = (Msg)(rawData.Buffer[0]); //头
                switch (msg)
                {
                    case Msg.ZMsg1: //登录
                        {
                            string data = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);  //从1开始  因为前面有一个 头  下同，注意是 rawData.Length(实际接收数据)  不是rawData.Buffer.Length
                            OnAnalysed(msg, data, ip, port);
                            break;
                        }
                    case Msg.ZMsg2: //登录反馈
                        {
                            string data = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);
                            OnAnalysed(msg, data, ip, port);
                            break;
                        }
                    case Msg.ZMsg3: //文本信息
                        {
                            string data = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);
                            OnAnalysed(msg, data, ip, port);
                            break;
                        }
                    case Msg.ZMsg4: //文本信息反馈
                        {
                            string data = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);
                            OnAnalysed(msg, data, ip, port);
                            break;
                        }
                    case Msg.ZMsg5: //退出
                        {
                            string data = Encoding.Unicode.GetString(rawData.Buffer, 1, rawData.Length - 1);
                            OnAnalysed(msg, data, ip, port);
                            break;
                        }
                    case Msg.ZMsg6: //请求发送文件
                        {
                            int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
                            long length = BitConverter.ToInt64(rawData.Buffer, 5); //文件大小
                            string filename = Encoding.Unicode.GetString(rawData.Buffer, 13, rawData.Length - 13); //文件名称
                            OnAnalysed(msg, new object[] { uid, length, filename }, ip, port); //
                            break;
                        }
                    case Msg.ZMsg7: //请求发送文件反馈
                        {
                            int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
                            int result = BitConverter.ToInt32(rawData.Buffer, 5); //反馈结果
                            OnAnalysed(msg, new object[] { uid, result }, ip, port); 
                            break;
                        }
                    case Msg.ZMsg8: //发送文件数据
                        {
                            int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
                            int part = BitConverter.ToInt32(rawData.Buffer, 5); //段ID
                            byte[] data = new byte[rawData.Length - 9]; //实际文件数据
                            Buffer.BlockCopy(rawData.Buffer, 9, data, 0, data.Length);
                            OnAnalysed(msg, new object[] { uid, part, data }, ip, port);
                            break;
                        }
                    case Msg.ZMsg9: //接收文件完毕
                        {
                            int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
                            OnAnalysed(msg, uid, ip, port);
                            break;
                        }
                    case Msg.ZMsg10: //发送方中止发送文件
                        {
                            int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
                            OnAnalysed(msg, uid, ip, port);
                            break;
                        }
                    case Msg.ZMsg11: //接收方中止接收文件
                        {
                            int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
                            OnAnalysed(msg, uid, ip, port);
                            break;
                        }
                    case Msg.ZMsg12: //发送文件数据反馈
                        {
                            int uid = BitConverter.ToInt32(rawData.Buffer, 1); //文件唯一标示
                            int part = BitConverter.ToInt32(rawData.Buffer, 5); //段ID
                            OnAnalysed(msg, new object[] { uid, part }, ip, port);
                            break;
                        }
                    default:
                        {
                            base.Analyse(rawData);  //其他由基类分析
                            break;
                        }
                }
            }
        }
    }

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

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

ZMsg1：用户新上线                  头+用户信息
ZMsg2：用户新上线反馈              头+反馈用户信息
ZMsg3：文本消息                    头+密码+消息正文
ZMsg4：文本消息反馈                头+原始消息
ZMsg5：离线                        头+任意文本
ZMsg6：请求发送文件                头+发送文件的唯一标示+大小+文件名
ZMsg7：请求发送文件反馈            头+文件的唯一标示+1或0           1代表接收 0代表拒绝
ZMsg8：发送文件数据                头+文件唯一标示+段ID+文件数据（byte[]）
ZMsg9：接收文件完毕                头+文件唯一标示
ZMsg10：发送方中止发送文件         头+文件唯一标示
ZMsg11：接收方中止接收文件         头+文件唯一标示
ZMsg12：发送文件数据反馈           头+文件唯一标示+段ID

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

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

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

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

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