zoukankan      html  css  js  c++  java
  • 【转载】C#实现线程安全的网络流

     http://www.csharpwin.com/csharpspace/10274r6060.shtml

    http://www.csharpwin.com/search.aspx?ChID=0&AID=0&KW=ESFramework

    前面我们已经讨论了客户端与服务器通信的一般模式,即Client通过Tcp连接向Server递交请求,Server处理请求后,使用同一Tcp连接将服务结果发送给Client。这一Tcp连接在程序中被抽象成网络流,也就是System.Net.Sockets.NetworkStream类,这个类是非线程安全的。

     

    一.线程安全

    如果一个类是线程安全的,表明该类的任何一个实例是线程安全的。也就是说,在多线程的环境中,对任何一个这样的实例的并发访问将被自动同步。由于,同步是由类自己完成的,所以,类的使用者不需要再使用额外的同步机制。

     二.线程安全网络流的意义

    按照前面讲述的通信的一般模式,我们也许觉得NetworkStream不需要线程安全机制,因为好像其总是只被单线程访问。简单的情况下确实如此。但是,更常见的需求是,NetworkStream必须支持并发访问。

    设想这样一个场景,当一个Client通过Tcp连接上Server后,需要每隔一分钟定时向服务器发送一个Check消息,以表明自己还在线上,这通常是在一个后台线程中完成的。假设某个时刻,主线程正在向Server发送某个功能请求,而同时,后台线程也在发送Check消息。如果网络流是非线程安全的,那么Server端接收到的将是一堆无法解析的数据。有两种方法可以解决这个问题,一是仍然使用System.Net.Sockets.NetworkStream类,只不过由该类的使用者提供同步机制,就像这样,每次调用的NetworkStream方法时先lock:

    lock(this.networkStream)
    {

    this.networkStream.Write(buffer ,offset ,size ) ;
    }

         从客户端开发者的角度来看,这个方法是可以勉强一用的。但是,如果你是一个服务器开发者了,你需要管理的可能是成千上万个NetworkStream,并且,任何时刻都可能有新的NetworkStream建立,也会有NetworkStream被销毁。在这种情况下,仍然使用第一种方法将会死的很难看!
          “所有的软件问题都可以通过引入一个间接层来得到解决”,这是我经常引用的一句话,它可以导出上述问题的第二种解决方案,那就是设计一个线程安全的网络流类SafeNetworkStream,它将自动同步所有的并发访问,它把所有分散的同步代码(比如,lock、Monitor等)全部简化并集中在自己的身体里,极大的方便了使用者。

    三.线程安全的网络流的接口与实现

    线程安全的网络流的接口ISafeNetworkStream中的方法覆盖了类NetworkStream的几个常用方法,在我们的通信框架中,这个接口中的内容已经够我们使用了。ISafeNetworkStream接口定义如下:

    /// <summary>
    /// INetworkStreamSafe 线程安全的网络流 。
    /// 注意:如果调用的异步的begin方法,就一定要调用对应的End方法,否则锁将得不到释放。
    /// 作者:朱伟 sky.zhuwei@163.com
    /// 2005.04.22
    /// </summary>

    //用于在TCP连接上发送数据,支持同步和异步
    public interface ITcpSender
    {
    void Write(byte[] buffer ,int offset ,int size) ;
    IAsyncResult BeginWrite(
    byte[] buffer, int offset, int size, AsyncCallback callback, object state );
    void EndWrite(IAsyncResult asyncResult );
    }

    //用于在TCP连接上接收数据,支持同步和异步
    public interface ITcpReciever
    {
    int Read (byte[] buffer ,int offset ,int size) ;
    IAsyncResult BeginRead(
    byte[] buffer, int offset, int size, AsyncCallback callback, object state );
    int EndRead(IAsyncResult asyncResult );
    }

    public interface ISafeNetworkStream :ITcpSender ,ITcpReciever
    {
    void Flush();
    void Close() ;

    NetworkStream NetworkStream{
    get ;}
    }

        接口中各方法的含义与NetworkStream类的方法含义相同。从接口的NetworkStream属性可以推断出,SafeNetworkStream是对NetworkStream的一层薄薄的封装,在这封装的夹层中存在的就是同步机制。
          ISafeNetworkStream接口的实现SafeNetworkStream主要是通过lock来进行同步处理的,其实现代码如下:    整个实现中,没有什么难的,所以就不多费口舌了。

    //SafeNetworkStream
    public class SafeNetworkStream : ISafeNetworkStream
    {
    private System.Net.Sockets.NetworkStream stream = null;
    private object lockForRead = null ;
    private object lockForWrite = null ;
    private volatile bool isReading = false ;
    private volatile bool isWriting = false ;
    private int retryCount = 10 ;

    public SafeNetworkStream(NetworkStream netStream)
    {
    this.stream = netStream ;
    if(netStream == null)
    {
    throw new NullReferenceException("the wrapped netStream is null in SafeNetworkStream's ctor !") ;
    }


    this.lockForRead = this ;
    this.lockForWrite = this.stream ;
    }


    #region StartReadAction ,EndReadAction ,StartWriteAction ,EndWriteAction
    private bool StartReadAction()
    {
    lock(this.lockForRead)
    {
    if(this.isReading)
    {
    return false ;
    }


    this.isReading = true ;
    return true ;
    }

    }


    private void EndReadAction()
    {
    lock(this.lockForRead)
    {
    this.isReading = false ;
    }

    }


    private bool StartWriteAction()
    {
    lock(this.lockForWrite)
    {
    if(this.isWriting)
    {
    return false ;
    }


    this.isWriting = true ;
    return true ;
    }

    }


    private void EndWriteAction()
    {
    lock(this.lockForWrite)
    {
    this.isWriting = false ;
    }

    }

    #endregion


    #region ISafeNetworkStream 成员

    #region Write ,BeginWrite ,EndWrite
    public void Write(byte[] buffer, int offset, int size)
    {
    int count = 0 ;
    while(! this.StartWriteAction())
    {
    if(count >= this.retryCount)
    {
    return ;
    }

    Thread.Sleep(
    200) ;
    ++ count ;
    }


    this.stream.Write(buffer ,offset ,size ) ;

    this.EndWriteAction() ;
    }


    public IAsyncResult BeginWrite(byte[] buffer, int offset, int size, System.AsyncCallback callback, object state)
    {
    int count = 0 ;
    while(! this.StartWriteAction())
    {
    if(count >= this.retryCount)
    {
    return null ;
    }

    Thread.Sleep(
    200) ;
    ++ count ;
    }


    return this.stream.BeginWrite(buffer, offset, size, callback, state) ;

    }


    public void EndWrite(IAsyncResult asyncResult)
    {
    this.EndWrite(asyncResult) ;
    this.EndWriteAction() ;
    }

    #endregion


    #region Read ,BeginRead ,EndRead
    public int Read(byte[] buffer, int offset, int size)
    {
    int count = 0 ;
    while(! this.StartReadAction())
    {
    if(count >= this.retryCount)
    {
    return -1 ;
    }

    Thread.Sleep(
    200) ;
    ++ count ;
    }


    int readCount = this.stream.Read(buffer ,offset ,size ) ;

    this.EndReadAction() ;

    return readCount ;
    }


    public IAsyncResult BeginRead(byte[] buffer, int offset, int size, System.AsyncCallback callback, object state)
    {
    int count = 0 ;
    while(! this.StartReadAction())
    {
    if(count >= this.retryCount)
    {
    return null ;
    }

    Thread.Sleep(
    200) ;
    ++ count ;
    }


    return this.stream.BeginRead(buffer, offset, size, callback, state) ;
    }


    public int EndRead(IAsyncResult asyncResult)
    {
    int count = this.stream.EndRead(asyncResult) ;
    this.EndReadAction() ;

    return count ;
    }

    #endregion


    #region Flush ,Close
    public void Flush()
    {
    this.stream.Flush() ;
    }


    public void Close()
    {
    this.stream.Close() ;
    }

    #endregion


    #region property
    public NetworkStream NetworkStream
    {
    get
    {
    return this.stream ;
    }

    }

    #endregion


    #endregion

    }
  • 相关阅读:
    第04组(64)需求分析报告
    实验 7:OpenDaylight 实验——Python 中的 REST API 调用
    结对编程作业
    第01组 Alpha冲刺(5/6)
    第01组 Alpha冲刺(4/6)
    第01组 Alpha冲刺 (3/6)
    第01组 Alpha冲刺(2/6)
    第01组 Alpha冲刺(1/6)
    第01组(17)需求分析报告
    第01组(17)团队展示
  • 原文地址:https://www.cnblogs.com/fx2008/p/2236485.html
Copyright © 2011-2022 走看看