至顶网›软件频道 ›基础软件›.NET可复用TCP通信层之消息分派器组件

.NET可复用TCP通信层之消息分派器组件

扫一扫
分享文章到微信
扫一扫
关注官方公众号
至顶头条

Tcp组件是整个通信层的消息驱动源，甚至，可以将Tcp组件看作是我们整个服务器系统的消息驱动源

作者：zhuweisky 来源：博客园 2007年11月4日

　上一篇主要讲到了Tcp通信层中的核心组件――Tcp组件的实现，Tcp组件是整个通信层的消息驱动源，甚至，可以将Tcp组件看作是我们整个服务器系统的消息驱动源，消息处理过程从这里引发。类似的消息驱动源还有发布的WebService接口、Remoting接口等。今天我们需要关注的是Tcp通信层中的“中央”组件――消息分派器组件ITcpReqStreamDispatcher，大家已经从前文的组件关系图中看到了消息分派器的大致位置和作用了，它是Tcp通信组件和消息处理器之间的“桥梁”。我们再对前文描述的通信层组件之间关系的一段话回顾一下：

　　“当网络（Tcp）组件从某个Tcp连接上接收到一个请求时，会将请求转发给消息分派器，消息分派器通过IDataStreamHelper组件获取请求消息的类型，然后根据此类型要求处理器工厂创建对应类型的请求处理器，请求处理器处理请求并返回结果。接下来再由网络组件把结果返回给终端用户。在消息分派器进行请求消息分派之前，可能涉及一系列的操作，像消息加密/解密、消息分裂/重组、消息验证等。”

　　上面的描述中已经体现出了消息分派器的主要职责，在理解了消息分派器职责的基础上，我们可以进一步来看看消息分派器的定义和实现了。

　　二．消息分派器组件

　　1．消息分派器组件接口的定义

　　消息分派器的接口很简单：

    public interface ITcpReqStreamDispatcher : IReqestStreamDispatcher
    {
        ArrayList DealRequestMessage(RequestData requestData ,out byte[] leftData ,ref RequestValidation validation) ;//同步回复
        bool      DealRequestMessage(RequestData requestData , NetworkStream userStream ,out byte[] leftData) ; //异步回复
    }

　　这个接口只有两个方法，第二个方法用于异步发送回复（即绕开Tcp组件发送回复），该方法的核心部分可以由第一个方法实现，我们把注意力放在第一个方法上，而Tcp组件与消息分派器进行交互的也正是第一个方法。我先解释一下这个方法的几个参数的含义：

　　RequestData是对请求消息的封装：

    //从网络接收到的原始数据的封装
    public class RequestData
    {
        public int  ConnectID = 0 ;
        public bool IsFirstMsg = false ; //标志是否为连接建立后的第一条消息
        public byte[] Buff     = null ; //接收数据缓冲区 ,可能其头部包含上次未处理完的数据
        public int ValidCount  = 0 ; //缓冲区中有效字节的个数 >= 本次接收的字节数
    }

　　前面已经提到过，ConnectID用于标志每一个Tcp连接，IsFirstMsg用于表明是否为tcp连接建立后的第一个消息，因为我们可能需要对第一个消息进行额外的验证，比如，果第一个消息不是登录请求，就关闭该Tcp连接。

　　第二个参数leftData，表示RequestData.Buff中的数据经过消息分裂器分裂之后余下的数据（一条非完整的消息），这些数据被Tcp组件用来放在下一次收到的数据的头部进行消息重组。

　　第三个参数validation，是个ref参数，用于通知Tcp组件对消息验证的结果，如果验证失败，Tcp组件将关闭对应的Tcp连接。

　　该方法的返回值是回复的集合，每一个回复对应一个请求，而RequestData.Buff中的数据可能分裂成多个请求。另外要注意，有些请求可能是没有回复消息的。

　　在我们的Tcp组件的两种实现中，都可以看到类似下面的与消息分派器交互的语句：

                //处理请求
                byte[] leftData = null ;
                ArrayList repondList = this.messageDispatcher.DealRequestMessage(key.RequestData  ,out leftData , ref key.Validation) ;

                if(this.validateRequest)
                {
                    if(key.Validation.gotoCloseConnection)
                    {
                        this.DisposeOneConnection(streamHashCode ,key.Validation.cause) ;
                    }
                }

　　2．消息分派器组件基本元素的实现

　　正如在实现Tcp组件之前需要构建一些基本元素，在实现消息分派器之前也是如此，用于支持消息分派器实现的基本元素包括：IDataStreamHelper、消息分裂器、消息处理器工厂、ITcpStreamDispatcherHook等。

　　（1）IDataStreamHelper消息分裂器

　　IDataStreamHelper，前文中已经提到，IDataStreamHelper用于从请求/回复消息中提取消息的“元数据”，并提供一些辅助方法，每个特定的应用，它们对IDataStreamHelper的实现可能是不一样的。IDataStreamHelper接口定义如下：

     /// <summary>
    /// IDataStreamHelper 通信协议的面向流辅助设施。
    /// </summary>
    public interface IDataStreamHelper :IStringEncoder
    {
        int MaxRecieveBuffSize{get ;} //接收缓冲区的大小
        int MessageHeaderLength{get ;} //消息头的长度
        int OffsetOfLengthField{get ;} //表示消息长度的字段在消息头中的偏移

        IDataStreamHeader ParseMessageHeader(byte[] data ,int offset) ; //解析消息头

        LengthTypeInHeader LengthTypeInHeader{get ;}

        byte[] GetRespondWhenFailure(byte[] reqData ,ServiceFailureType failType) ;    //根据服务失败类型获取失败回复消息
        byte[] GetRespondWhenFailure(byte[] reqData ,string errorMsg) ;
    }

    /// <summary>
    /// StringEncoder 限定字符串编码格式
    /// </summary>
    public interface IStringEncoder
    {
        string GetStrFromStream(byte[] stream ,int offset ,int len) ;
        byte[] GetBytesFromStr(string ss) ;
    }

    /// <summary>
    /// ServiceFailureType 服务失败类型
    /// </summary>
    public enum ServiceFailureType
    {
        InvalidMessge ,ParseFailure ,HandleFailure ,ServiceStopped ,ServiceIsNotExit ,ServerIsBusy
    }

　　IDataStreamHeader即是我们所说的消息的“元数据”，如其名所示，它也是消息的“消息头”。请让我补充说明一下，依照我的经验，消息由消息头Header和消息主体Body组成，消息头用于存放消息的“元数据”等信息，而消息主体用于存放与特定请求相关的数据。消息头的长度固定，比如都是64字节或都是128字节。请求消息和回复消息公用相同格式的消息头。我们来看看消息头接口IDataStreamHeader的定义：

    public interface IDataStreamHeader
    {
        int MessageLength    {get ;set ;} //本消息长度
        int TypeKey            {get ;set ;} //请求的目录类型
        int ServiceKey        {get ;set ;} //请求类型
        int ServiceItemIndex{get ;set ;} //请求细分索引
        int RandomNum        {get ;set ;} //用于将回复与请求一一对应起来
        int Result            {get ;set ;} //服务结果

        string UserID        {get ;set ;} //发出请求的用户编号

        byte[] ToDataStream() ;              //将消息头转化为流，流的长度位消息头的长度
        void   ToDataStream(byte[] buff ,int offset);
    }

　　需要解释一下TypeKey、ServiceKey、ServiceItemIndex，我们实际上将服务类型分为三级，可以举个不太恰当的例子让大家有个感性的认识。比如，生活中的衣、食、住、行可以作为不同的TypeKey，而“衣”中的春装、冬装可作为ServiceKey，而“春装”中的T恤、夹克可作为ServiceItemIndex。对于服务的类型，你可以根据自己的意愿分成任意层级，但据我的经验，通常情况下，三层已经够用了。

　　（2）消息分裂器

　　前面已经多次提到消息分裂器MessageSplitter，它用于将接收缓冲区中的数据分裂成一个个完整的消息，并且把余下的非完整数据返回，其接口定义如下：

public interface IMessageSplitter
    {
        void Initialize(int maxBuffSize ,int headerLen ,int offSetLenField ,LengthTypeInHeader lenType) ;
        ArrayList SplitRequestMsgs(byte[] buff ,int validCount , out byte[] leftData) ;//ArrayList 中每条记录都是是byte[]，表示一个完整的请求
    }

    //消息头中的长度是body长度还是总长度
    public enum LengthTypeInHeader
    {
        TotalLen ,BodyLen
    }

　　其中，Initialize方法中的参数都可以由IDataStreamHeader提供。leftData是余下的非完整消息的数据。SplitRequestMsgs方法返回的集合中是一条条完整的请求消息。