当前位置: 开发笔记 > 运维 > 正文

在C#中对TCP客户端的状态封装详解

作者：小妖 | 来源：互联网 | 2022-03-18 01:57

TCP客户端连接TCP服务器端有几种应用状态：1.与服务器的连接已建立2.与服务器的连接已断开3.与服务器的连接发生异常应用程序可按需求合理

TCP客户端连接TCP服务器端有几种应用状态：
1.与服务器的连接已建立
2.与服务器的连接已断开
3.与服务器的连接发生异常

应用程序可按需求合理处理这些逻辑，比如：
1.连接断开后自动重连
2.连接断开后选择备用地址重连
3.所有状态变化上报告警
本文描述的TcpClient实现了状态变化的事件通知机制。

代码如下:

/// 
   /// 异步TCP客户端
   /// 
   public class AsyncTcpClient : IDisposable
   {
     #region Fields

     private TcpClient tcpClient;
     private bool disposed = false;
     private int retries = 0;

     #endregion

     #region Ctors

     /// 
     /// 异步TCP客户端
     /// 
     /// 远端服务器终结点
     public AsyncTcpClient(IPEndPoint remoteEP)
       : this(new[] { remoteEP.Address }, remoteEP.Port)
     {
     }

     /// 
     /// 异步TCP客户端
     /// 
     /// 远端服务器终结点
     /// 本地客户端终结点
     public AsyncTcpClient(IPEndPoint remoteEP, IPEndPoint localEP)
       : this(new[] { remoteEP.Address }, remoteEP.Port, localEP)
     {
     }

     /// 
     /// 异步TCP客户端
     /// 
     /// 远端服务器IP地址
     /// 远端服务器端口
     public AsyncTcpClient(IPAddress remoteIPAddress, int remotePort)
       : this(new[] { remoteIPAddress }, remotePort)
     {
     }

     /// 
     /// 异步TCP客户端
     /// 
     /// 远端服务器IP地址
     /// 远端服务器端口
     /// 本地客户端终结点
     public AsyncTcpClient(
       IPAddress remoteIPAddress, int remotePort, IPEndPoint localEP)
       : this(new[] { remoteIPAddress }, remotePort, localEP)
     {
     }

     /// 
     /// 异步TCP客户端
     /// 
     /// 远端服务器主机名
     /// 远端服务器端口
     public AsyncTcpClient(string remoteHostName, int remotePort)
       : this(Dns.GetHostAddresses(remoteHostName), remotePort)
     {
     }

     /// 
     /// 异步TCP客户端
     /// 
     /// 远端服务器主机名
     /// 远端服务器端口
     /// 本地客户端终结点
     public AsyncTcpClient(
       string remoteHostName, int remotePort, IPEndPoint localEP)
       : this(Dns.GetHostAddresses(remoteHostName), remotePort, localEP)
     {
     }

     /// 
     /// 异步TCP客户端
     /// 
     /// 远端服务器IP地址列表
     /// 远端服务器端口
     public AsyncTcpClient(IPAddress[] remoteIPAddresses, int remotePort)
       : this(remoteIPAddresses, remotePort, null)
     {
     }

     /// 
     /// 异步TCP客户端
     /// 
     /// 远端服务器IP地址列表
     /// 远端服务器端口
     /// 本地客户端终结点
     public AsyncTcpClient(
       IPAddress[] remoteIPAddresses, int remotePort, IPEndPoint localEP)
     {
       this.Addresses = remoteIPAddresses;
       this.Port = remotePort;
       this.LocalIPEndPoint = localEP;
       this.Encoding = Encoding.Default;

       if (this.LocalIPEndPoint != null)
       {
         this.tcpClient = new TcpClient(this.LocalIPEndPoint);
       }
       else
       {
         this.tcpClient = new TcpClient();
       }

       Retries = 3;
       RetryInterval = 5;
     }

     #endregion

     #region Properties

     /// 
     /// 是否已与服务器建立连接
     /// 
     public bool Connected { get { return tcpClient.Client.Connected; } }
     /// 
     /// 远端服务器的IP地址列表
     /// 
     public IPAddress[] Addresses { get; private set; }
     /// 
     /// 远端服务器的端口
     /// 
     public int Port { get; private set; }
     /// 
     /// 连接重试次数
     /// 
     public int Retries { get; set; }
     /// 
     /// 连接重试间隔
     /// 
     public int RetryInterval { get; set; }
     /// 
     /// 远端服务器终结点
     /// 
     public IPEndPoint RemoteIPEndPoint 
     { 
       get { return new IPEndPoint(Addresses[0], Port); } 
     }
     /// 
     /// 本地客户端终结点
     /// 
     protected IPEndPoint LocalIPEndPoint { get; private set; }
     /// 
     /// 通信所使用的编码
     /// 
     public Encoding Encoding { get; set; }

     #endregion

     #region Connect

     /// 
     /// 连接到服务器
     /// 
     /// 异步TCP客户端
     public AsyncTcpClient Connect()
     {
       if (!Connected)
       {
         // start the async connect operation
         tcpClient.BeginConnect(
           Addresses, Port, HandleTcpServerConnected, tcpClient);
       }

       return this;
     }

     /// 
     /// 关闭与服务器的连接
     /// 
     /// 异步TCP客户端
     public AsyncTcpClient Close()
     {
       if (Connected)
       {
         retries = 0;
         tcpClient.Close();
         RaiseServerDisconnected(Addresses, Port);
       }

       return this;
     }

     #endregion

     #region Receive

     private void HandleTcpServerConnected(IAsyncResult ar)
     {
       try
       {
         tcpClient.EndConnect(ar);
         RaiseServerConnected(Addresses, Port);
         retries = 0;
       }
       catch (Exception ex)
       {
         ExceptionHandler.Handle(ex);
         if (retries > 0)
         {
           Logger.Debug(string.Format(CultureInfo.InvariantCulture, 
             "Connect to server with retry {0} failed.", retries));
         }

         retries++;
         if (retries > Retries)
         {
           // we have failed to connect to all the IP Addresses, 
           // connection has failed overall.
           RaiseServerExceptionOccurred(Addresses, Port, ex);
           return;
         }
         else
         {
           Logger.Debug(string.Format(CultureInfo.InvariantCulture, 
             "Waiting {0} seconds before retrying to connect to server.", 
             RetryInterval));
           Thread.Sleep(TimeSpan.FromSeconds(RetryInterval));
           Connect();
           return;
         }
       }

       // we are connected successfully and start asyn read operation.
       byte[] buffer = new byte[tcpClient.ReceiveBufferSize];
       tcpClient.GetStream().BeginRead(
         buffer, 0, buffer.Length, HandleDatagramReceived, buffer);
     }

     private void HandleDatagramReceived(IAsyncResult ar)
     {
       NetworkStream stream = tcpClient.GetStream();

       int numberOfReadBytes = 0;
       try
       {
         numberOfReadBytes = stream.EndRead(ar);
       }
       catch
       {
         numberOfReadBytes = 0;
       }

       if (numberOfReadBytes == 0)
       {
         // connection has been closed
         Close();
         return;
       }

       // received byte and trigger event notification
       byte[] buffer = (byte[])ar.AsyncState;
       byte[] receivedBytes = new byte[numberOfReadBytes];
       Buffer.BlockCopy(buffer, 0, receivedBytes, 0, numberOfReadBytes);
       RaiseDatagramReceived(tcpClient, receivedBytes);
       RaisePlaintextReceived(tcpClient, receivedBytes);

       // then start reading from the network again
       stream.BeginRead(
         buffer, 0, buffer.Length, HandleDatagramReceived, buffer);
     }

     #endregion

     #region Events

     /// 
     /// 接收到数据报文事件
     /// 
     public event EventHandler> DatagramReceived;
     /// 
     /// 接收到数据报文明文事件
     /// 
     public event EventHandler> PlaintextReceived;

     private void RaiseDatagramReceived(TcpClient sender, byte[] datagram)
     {
       if (DatagramReceived != null)
       {
         DatagramReceived(this, 
           new TcpDatagramReceivedEventArgs(sender, datagram));
       }
     }

     private void RaisePlaintextReceived(TcpClient sender, byte[] datagram)
     {
       if (PlaintextReceived != null)
       {
         PlaintextReceived(this, 
           new TcpDatagramReceivedEventArgs(
             sender, this.Encoding.GetString(datagram, 0, datagram.Length)));
       }
     }

     /// 
     /// 与服务器的连接已建立事件
     /// 
     public event EventHandler ServerConnected;
     /// 
     /// 与服务器的连接已断开事件
     /// 
     public event EventHandler ServerDisconnected;
     /// 
     /// 与服务器的连接发生异常事件
     /// 
     public event EventHandler ServerExceptionOccurred;

     private void RaiseServerConnected(IPAddress[] ipAddresses, int port)
     {
       if (ServerConnected != null)
       {
         ServerConnected(this, 
           new TcpServerConnectedEventArgs(ipAddresses, port));
       }
     }

     private void RaiseServerDisconnected(IPAddress[] ipAddresses, int port)
     {
       if (ServerDisconnected != null)
       {
         ServerDisconnected(this, 
           new TcpServerDisconnectedEventArgs(ipAddresses, port));
       }
     }

     private void RaiseServerExceptionOccurred(
       IPAddress[] ipAddresses, int port, Exception innerException)
     {
       if (ServerExceptionOccurred != null)
       {
         ServerExceptionOccurred(this, 
           new TcpServerExceptionOccurredEventArgs(
             ipAddresses, port, innerException));
       }
     }

     #endregion

     #region Send

     /// 
     /// 发送报文
     /// 
     /// 报文
     public void Send(byte[] datagram)
     {
       if (datagram == null)
         throw new ArgumentNullException("datagram");

       if (!Connected)
       {
         RaiseServerDisconnected(Addresses, Port);
         throw new InvalidProgramException(
           "This client has not connected to server.");
       }

       tcpClient.GetStream().BeginWrite(
         datagram, 0, datagram.Length, HandleDatagramWritten, tcpClient);
     }

     private void HandleDatagramWritten(IAsyncResult ar)
     {
       ((TcpClient)ar.AsyncState).GetStream().EndWrite(ar);
     }

     /// 
     /// 发送报文
     /// 
     /// 报文
     public void Send(string datagram)
     {
       Send(this.Encoding.GetBytes(datagram));
     }

     #endregion

     #region IDisposable Members

     /// 
     /// Performs application-defined tasks associated with freeing, 
     /// releasing, or resetting unmanaged resources.
     /// 
     public void Dispose()
     {
       Dispose(true);
       GC.SuppressFinalize(this);
     }

     /// 
     /// Releases unmanaged and - optionally - managed resources
     /// 
     /// true to release both managed 
     /// and unmanaged resources; false 
     /// to release only unmanaged resources.
     /// 
     protected virtual void Dispose(bool disposing)
     {
       if (!this.disposed)
       {
         if (disposing)
         {
           try
           {
             Close();

             if (tcpClient != null)
             {
               tcpClient = null;
             }
           }
           catch (SocketException ex)
           {
             ExceptionHandler.Handle(ex);
           }
         }

         disposed = true;
       }
     }

     #endregion
   }

使用举例

代码如下:

class Program
   {
     static AsyncTcpClient client;

     static void Main(string[] args)
     {
       LogFactory.Assign(new ConsoleLogFactory());

       // 测试用，可以不指定由系统选择端口
       IPEndPoint remoteEP = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 9999);
       IPEndPoint localEP = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 9998); 
       client = new AsyncTcpClient(remoteEP, localEP);
       client.Encoding = Encoding.UTF8;
       client.ServerExceptionOccurred += 
         new EventHandler(client_ServerExceptionOccurred);
       client.ServerConnected += 
         new EventHandler(client_ServerConnected);
       client.ServerDisconnected += 
         new EventHandler(client_ServerDisconnected);
       client.PlaintextReceived += 
         new EventHandler>(client_PlaintextReceived);
       client.Connect();

       Console.WriteLine("TCP client has connected to server.");
       Console.WriteLine("Type something to send to server...");
       while (true)
       {
         try
         {
           string text = Console.ReadLine();
           client.Send(text);
         }
         catch (Exception ex)
         {
           Console.WriteLine(ex.Message);
         }
       }
     }

     static void client_ServerExceptionOccurred(
       object sender, TcpServerExceptionOccurredEventArgs e)
     {
       Logger.Debug(string.Format(CultureInfo.InvariantCulture, 
         "TCP server {0} exception occurred, {1}.", 
         e.ToString(), e.Exception.Message));
     }

     static void client_ServerConnected(
       object sender, TcpServerConnectedEventArgs e)
     {
       Logger.Debug(string.Format(CultureInfo.InvariantCulture, 
         "TCP server {0} has connected.", e.ToString()));
     }

     static void client_ServerDisconnected(
       object sender, TcpServerDisconnectedEventArgs e)
     {
       Logger.Debug(string.Format(CultureInfo.InvariantCulture, 
         "TCP server {0} has disconnected.", e.ToString()));
     }

     static void client_PlaintextReceived(
       object sender, TcpDatagramReceivedEventArgs e)
     {
       Console.Write(string.Format("Server : {0} --> ", 
         e.TcpClient.Client.RemoteEndPoint.ToString()));
       Console.WriteLine(string.Format("{0}", e.Datagram));
     }
   }

补充代码

代码如下:

/// 
   /// Internal class to join the TCP client and buffer together
   /// for easy management in the server
   /// 
   internal class TcpClientState
   {
     /// 
     /// Constructor for a new Client
     /// 
     /// The TCP client
     /// The byte array buffer
     public TcpClientState(TcpClient tcpClient, byte[] buffer)
     {
       if (tcpClient == null)
         throw new ArgumentNullException("tcpClient");
       if (buffer == null)
         throw new ArgumentNullException("buffer");

       this.TcpClient = tcpClient;
       this.Buffer = buffer;
     }

     /// 
     /// Gets the TCP Client
     /// 
     public TcpClient TcpClient { get; private set; }

     /// 
     /// Gets the Buffer.
     /// 
     public byte[] Buffer { get; private set; }

     /// 
     /// Gets the network stream
     /// 
     public NetworkStream NetworkStream
     {
       get { return TcpClient.GetStream(); }
     }
   }

代码如下:

/// 
   /// 与客户端的连接已建立事件参数
   /// 
   public class TcpClientConnectedEventArgs : EventArgs
   {
     /// 
     /// 与客户端的连接已建立事件参数
     /// 
     /// 客户端
     public TcpClientConnectedEventArgs(TcpClient tcpClient)
     {
       if (tcpClient == null)
         throw new ArgumentNullException("tcpClient");

       this.TcpClient = tcpClient;
     }

     /// 
     /// 客户端
     /// 
     public TcpClient TcpClient { get; private set; }
   }

代码如下:

///

/// 与客户端的连接已断开事件参数
///

public class TcpClientDisconnectedEventArgs : EventArgs
{
///

/// 与客户端的连接已断开事件参数
///

    /// 客户端
    public TcpClientDisconnectedEventArgs(TcpClient tcpClient)
    {
      if (tcpClient == null)
        throw new ArgumentNullException("tcpClient");

this.TcpClient = tcpClient;
}

///

/// 客户端
///

public TcpClient TcpClient { get; private set; }
}

代码如下:

/// 
   /// 与服务器的连接发生异常事件参数
   /// 
   public class TcpServerExceptionOccurredEventArgs : EventArgs
   {
     /// 
     /// 与服务器的连接发生异常事件参数
     /// 
     /// 服务器IP地址列表
     /// 服务器端口
     /// 内部异常
     public TcpServerExceptionOccurredEventArgs(
       IPAddress[] ipAddresses, int port, Exception innerException)
     {
       if (ipAddresses == null)
         throw new ArgumentNullException("ipAddresses");

       this.Addresses = ipAddresses;
       this.Port = port;
       this.Exception = innerException;
     }

     /// 
     /// 服务器IP地址列表
     /// 
     public IPAddress[] Addresses { get; private set; }
     /// 
     /// 服务器端口
     /// 
     public int Port { get; private set; }
     /// 
     /// 内部异常
     /// 
     public Exception Exception { get; private set; }

     /// 
     /// Returns a  that represents this instance.
     /// 
     /// 
     /// A  that represents this instance.
     /// 
     public override string ToString()
     {
       string s = string.Empty;
       foreach (var item in Addresses)
       {
         s = s + item.ToString() + ',';
       }
       s = s.TrimEnd(',');
       s = s + ":" + Port.ToString(CultureInfo.InvariantCulture);

       return s;
     }
   }