几天前在博问中看到一个C# Socket问题,就想到笔者2004年做的一个省级交通流量接收服务器项目,当时的基本求如下:
- 接收自动观测设备通过无线网卡、Internet和Socket上报的交通量数据包
- 全年365*24运行的自动观测设备5分钟上报一次观测数据,每笔记录约2K大小
- 规划全省将有100个左右的自动观测设备(截止2008年10月还只有30个)
当时,VS2003才发布年多,笔者也是接触C#不久。于是Google了国内国外网,希望找点应用C#解决Socket通信问题的思路和代码。最后,找到了两篇帮助最大的文章:一篇是国人写的Socket接收器框架,应用了独立的客户端Socket会话(Session)概念,给笔者提供了一个接收服务器的总体框架思路;另一篇是美国人写的,提出了多线程、分段接收数据包的技术方案,描述了多线程、异步Socket的许多实现细节,该文坚定了笔者采用多线程和异步方式处理Socket接收器的技术路线。
具体实现和测试时笔者还发现,在Internet环境下的Socket应用中,需要系统有极强的容错能力:没有办法控制异常,就必须允许它们存在(附加源代码中可以看到,try{}catch{}语句较多)。对此,笔者设计了一个专门的检查和清理线程,完成无效或超时会话的清除和资源释放工作。
依稀记得,国内框架作者的名称空间有ibm,认为是IBM公司职员,通过邮件后才知道其人在深圳。笔者向他请教了几个问题,相互探讨了几个技术关键点。可惜,现在再去找,已经查不到原文和邮件了。只好借此机会,将本文献给这两个素未谋面的技术高人和同行,也盼望拙文或源码能给读者一点有用的启发和帮助。
1、主要技术思路
整个系统由三个核心线程组成,并由.NET线程池统一管理:
侦听客户端连接请求线程:ListenClientRequest(),循环侦听客户端连接请求。如果有,检测该客户端IP,看是否是同一观测设备,然后建立一个客户端TSession对象,并通过Socket异步调用方法BeginReceive()接收数据包、EndReceive()处理数据包 数据包处理线程:HandleDatagrams(),循环检测数据包队列_datagramQueue,完成数据包解析、判断类型、存储等工作 客户端状态检测线程:CheckClientState(),循环检查客户端会话表_sessionTable,判断会话对象是否有效,设置超时会话关闭标志,清楚无效会话对象及释放其资源
2、主要类简介
系统主要由3个类组成:
TDatagramReceiver(数据包接收服务器):系统的核心进程类,建立Socket连接、处理与存储数据包、清理系统资源,该类提供全部的public属性和方法 TSession(客户端会话):由每个客户端的Socket对象组成,有自己的数据缓冲区,清理线程根据该对象的最近会话时间判断是否超时 TDatagram(数据包类):判断数据包类别、解析数据包
3、关键函数和代码
下面简介核心类TDatagramReceiver的关键实现代码。
3.1 系统启动
系统启动方法StartReceiver()首先清理资源、创建数据库连接、初始化若干计数值,然后创建服务器端侦听Socket对象,最后调用静态方法ThreadPool.QueueUserWorkItem()在线程池中创建3个核心处理线程。
Code- /// <summary>
- /// 启动接收器
- /// </summary>
- public bool StartReceiver()
- {
- try
- {
- _stopReceiver = true;
- this.Close();
- if (!this.ConnectDatabase()) return false;
- _clientCount = 0;
- _datagramQueueCount = 0;
- _datagramCount = 0;
- _errorDatagramCount = 0;
- _exceptionCount = 0;
- _sessionTable = new Hashtable(_maxAllowClientCount);
- _datagramQueue = new Queue<TDatagram>(_maxAllowDatagramQueueCount);
- _stopReceiver = false; // 循环中均要该标志
- if (!this.CreateReceiverSocket()) //建立服务器端 Socket 对象
- {
- return false;
- }
- // 侦听客户端连接请求线程, 使用委托推断, 不建 CallBack 对象
- if (!ThreadPool.QueueUserWorkItem(ListenClientRequest))
- {
- return false;
- }
- // 处理数据包队列线程
- if (!ThreadPool.QueueUserWorkItem(HandleDatagrams))
- {
- return false;
- }
- // 检查客户会话状态, 长时间未通信则清除该对象
- if (!ThreadPool.QueueUserWorkItem(CheckClientState))
- {
- return false;
- }
- _stopConnectRequest = false; // 启动接收器,则自动允许连接
- }
- catch
- {
- this.OnReceiverException();
- _stopReceiver = true;
- }
- return !_stopReceiver;
- }
复制代码 下面是创建侦听Socket对象的方法代码。
Code- /// <summary>
- /// 创建接收服务器的 Socket, 并侦听客户端连接请求
- /// </summary>
- private bool CreateReceiverSocket()
- {
- try
- {
- _receiverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
- _receiverSocket.Bind(new IPEndPoint(IPAddress.Any, _tcpSocketPort)); // 绑定端口
- _receiverSocket.Listen(_maxAllowListenQueueLength); // 开始监听
- return true;
- }
- catch
- {
- this.OnReceiverException();
- return false;
- }
- }
复制代码 3.2 侦听客户端连接请求
服务器端循环等待客户端连接请求。一旦有请求,先判断客户端连接数是否超限,接着检测该客户端IP地址,一切正常后建立TSession对象,并调用异步方法接收客户端Socket数据包。
代码中,Socket读到数据时的回调AsyncCallback委托方法EndReceiveData()完成数据接收工作,正常情况下启动另一个异步BeginReceive()调用。
.NET中,每个异步方法都有自己的独立线程,异步处理其实也基于多线程机制的。下面代码中的异步套异步调用,既占用较大的系统资源,也给处理带来意想不到的结果,更是出现异常时难以控制和处理的关键所在。
Code- /// <summary>
- /// 循环侦听客户端请求,由于要用线程池,故带一个参数
- /// </summary>
- private void ListenClientRequest(object state)
- {
- Socket client = null;
- while (!_stopReceiver)
- {
- if (_stopConnectRequest) // 停止客户端连接请求
- {
- if (_receiverSocket != null)
- {
- try
- {
- _receiverSocket.Close(); // 强制关闭接收器
- }
- catch
- {
- this.OnReceiverException();
- }
- finally
- {
- // 必须为 null,否则 disposed 对象仍然存在,将引发下面的错误
- _receiverSocket = null;
- }
- }
- continue;
- }
- else
- {
- if (_receiverSocket == null)
- {
- if (!this.CreateReceiverSocket())
- {
- continue;
- }
- }
- }
- try
- {
- if (_receiverSocket.Poll(_loopWaitTime, SelectMode.SelectRead))
- {
- // 频繁关闭、启动时,这里容易产生错误(提示套接字只能有一个)
- client = _receiverSocket.Accept();
- if (client != null && client.Connected)
- {
- if (this._clientCount >= this._maxAllowClientCount)
- {
- this.OnReceiverException();
- try
- {
- client.Shutdown(SocketShutdown.Both);
- client.Close();
- }
- catch { }
- }
- else if (CheckSameClientIP(client)) // 已存在该 IP 地址
- {
- try
- {
- client.Shutdown(SocketShutdown.Both);
- client.Close();
- }
- catch { }
- }
- else
- {
- TSession session = new TSession(client);
- session.LoginTime = DateTime.Now;
- lock (_sessionTable)
- {
- int preSessionID = session.ID;
- while (true)
- {
- if (_sessionTable.ContainsKey(session.ID)) // 有可能重复该编号
- {
- session.ID = 100000 + preSessionID;
- }
- else
- {
- break;
- }
- }
- _sessionTable.Add(session.ID, session); // 登记该会话客户端
- Interlocked.Increment(ref _clientCount);
- }
- this.OnClientRequest();
- try // 客户端连续连接或连接后立即断开,易在该处产生错误,系统忽略之
- {
- // 开始接受来自该客户端的数据
- session.ClientSocket.BeginReceive(session.ReceiveBuffer, 0,
- session.ReceiveBufferLength, SocketFlags.None, EndReceiveData, session);
- }
- catch
- {
- session.DisconnectType = TDisconnectType.Exception;
- session.State = TSessionState.NoReply;
- }
- }
- }
- else if (client != null) // 非空,但没有连接(connected is false)
- {
- try
- {
- client.Shutdown(SocketShutdown.Both);
- client.Close();
- }
- catch { }
- }
- }
- }
- catch
- {
- this.OnReceiverException();
- if (client != null)
- {
- try
- {
- client.Shutdown(SocketShutdown.Both);
- client.Close();
- }
- catch { }
- }
- }
- // 该处可以适当暂停若干毫秒
- }
- // 该处可以适当暂停若干毫秒
- }
- /// <summary>
- /// 循环侦听客户端请求,由于要用线程池,故带一个参数
- /// </summary>
- private void ListenClientRequest(object state)
- {
- Socket client = null;
- while (!_stopReceiver)
- {
- if (_stopConnectRequest) // 停止客户端连接请求
- {
- if (_receiverSocket != null)
- {
- try
- {
- _receiverSocket.Close(); // 强制关闭接收器
- }
- catch
- {
- this.OnReceiverException();
- }
- finally
- {
- // 必须为 null,否则 disposed 对象仍然存在,将引发下面的错误
- _receiverSocket = null;
- }
- }
- continue;
- }
- else
- {
- if (_receiverSocket == null)
- {
- if (!CreateReceiverSocket())
- {
- continue;
- }
- }
- }
- try
- {
- if (_receiverSocket.Poll(_loopWaitTime, SelectMode.SelectRead))
- {
- // 频繁关闭、启动时,这里容易产生错误(提示套接字只能有一个)
- client = _receiverSocket.Accept();
- if (client != null && client.Connected)
- {
- if (this._clientCount >= this._maxAllowClientCount)
- {
- this.OnReceiverException();
- try
- {
- client.Shutdown(SocketShutdown.Both);
- client.Close();
- }
- catch { }
- }
- else if (CheckSameClientIP(client)) // 已存在该 IP 地址
- {
- try
- {
- client.Shutdown(SocketShutdown.Both);
- client.Close();
- }
- catch { }
- }
- else
- {
- TSession session = new TSession(client);
- session.LoginTime = DateTime.Now;
- lock (_sessionTable)
- {
- int preSessionID = session.ID;
- while (true)
- {
- if (_sessionTable.ContainsKey(session.ID)) // 有可能重复该编号
- {
- session.ID = 100000 + preSessionID;
- }
- else
- {
- break;
- }
- }
- _sessionTable.Add(session.ID, session); // 登记该会话客户端
- Interlocked.Increment(ref _clientCount);
- }
- OnClientRequest();
- try // 客户端连续连接或连接后立即断开,易在该处产生错误,系统忽略之
- {
- // 开始接受来自该客户端的数据
- session.ClientSocket.BeginReceive(session.ReceiveBuffer, 0,
- session.ReceiveBufferLength, SocketFlags.None, EndReceiveData, session);
- }
- catch
- {
- session.DisconnectType = TDisconnectType.Exception;
- session.State = TSessionState.NoReply;
- }
- }
- }
- else if (client != null) // 非空,但没有连接(connected is false)
- {
- try
- {
- client.Shutdown(SocketShutdown.Both);
- client.Close();
- }
- catch { }
- }
- }
- }
- catch
- {
- this.OnReceiverException();
- if (client != null)
- {
- try
- {
- client.Shutdown(SocketShutdown.Both);
- client.Close();
- }
- catch { }
- }
- }
- // 该处可以适当暂停若干毫秒
- }
- // 该处可以适当暂停若干毫秒
- }
复制代码 3.3 处理数据包
该线程循环查看数据包队列,完成数据包的解析与存储等工作。具体实现时,如果队列中没有数据包,可以考虑等待若干毫秒,提高CPU利用率。
Code- private void HandleDatagrams(object state)
- {
- while (!_stopReceiver)
- {
- this.HandleOneDatagram(); // 处理一个数据包
- if (!_stopReceiver)
- {
- // 如果连接关闭,则重新建立,可容许几个连接错误出现
- if (_sqlConnection.State == ConnectionState.Closed)
- {
- this.OnReceiverWork();
- try
- {
- _sqlConnection.Open();
- }
- catch
- {
- this.OnReceiverException();
- }
- }
- }
- }
- }
- /// <summary>
- /// 处理一个包数据,包括:验证、存储
- /// </summary>
- private void HandleOneDatagram()
- {
- TDatagram datagram = null;
- lock (_datagramQueue)
- {
- if (_datagramQueue.Count > 0)
- {
- datagram = _datagramQueue.Dequeue(); // 取队列数据
- Interlocked.Decrement(ref _datagramQueueCount);
- }
- }
- if (datagram == null) return;
- datagram.Clear();
- datagram = null; // 释放对象
- }
复制代码 3.4 检查与清理会话
本线程负责处理建立连接后的客户端会话TSession或Socket对象的关闭与资源清理工作,其它方法中出现异常等情况,尽可能标记相关TSession对象的属性NoReply=true,表示该会话已经无效、需要清理。
检查会话队列并清理资源分3步:第一步,Shutdown()客户端Socket,此时可能立即触发某些Socket的异步方法EndReceive();第二步,Close()客户端Socket,释放占用资源;第三步,从会话表中清除该会话对象。其中,第一步完成后,某个TSession也许不会立即到第二步,因为可能需要处理其异步结束方法。
需要指出, 由于涉及多线程处理,需要频繁加解锁操作,清理工作前先建立一个会话队列列副本sessionTable2,检查与清理该队副本列列的TSession对象。
Code- /// <summary>
- /// 检查客户端状态(扫描方式,若长时间无数据,则断开)
- /// </summary>
- private void CheckClientState(object state)
- {
- while (!_stopReceiver)
- {
- DateTime thisTime = DateTime.Now;
- // 建立一个副本 ,然后对副本进行操作
- Hashtable sessionTable2 = new Hashtable();
- lock (_sessionTable)
- {
- foreach (TSession session in _sessionTable.Values)
- {
- if (session != null)
- {
- sessionTable2.Add(session.ID, session);
- }
- }
- }
- foreach (TSession session in sessionTable2.Values) // 对副本进行操作
- {
- Monitor.Enter(session);
- try
- {
- if (session.State == TSessionState.NoReply) // 分三步清除一个 Session
- {
- session.State = TSessionState.Closing;
- if (session.ClientSocket != null)
- {
- try
- {
- // 第一步:shutdown
- session.ClientSocket.Shutdown(SocketShutdown.Both);
- }
- catch { }
- }
- }
- else if (session.State == TSessionState.Closing)
- {
- session.State = TSessionState.Closed;
- if (session.ClientSocket != null)
- {
- try
- {
- // 第二步: Close
- session.ClientSocket.Close();
- }
- catch { }
- }
- }
- else if (session.State == TSessionState.Closed)
- {
- lock (_sessionTable)
- {
- // 第三步:remove from table
- _sessionTable.Remove(session.ID);
- Interlocked.Decrement(ref _clientCount);
- }
- this.OnClientRequest();
- session.Clear(); // 清空缓冲区
- }
- else if (session.State == TSessionState.Normal) // 正常的会话
- {
- TimeSpan ts = thisTime.Subtract(session.LastDataReceivedTime);
- if (Math.Abs(ts.TotalSeconds) > _maxSocketDataTimeout) // 超时,则准备断开连接
- {
- session.DisconnectType = TDisconnectType.Timeout;
- session.State = TSessionState.NoReply; // 标记为将关闭、准备断开
- }
- }
- }
- finally
- {
- Monitor.Exit(session);
- }
- } // end foreach
- sessionTable2.Clear();
- } // end while
- }
复制代码 4 、结语
基于多线程处理的系统代价是比较大的,需要经常调用加/解锁方法lock()或Monitor.Enter(),需要经常创建处理线程等。从实际运行效果看,笔者的实现方案有较好的稳定性:2005年4月到5月间,在一个普通PC机器上连续运行30多天不出一点故障。同时,笔者采用了时序区间判重等算法,有效地提高了系统处理与响应速度。测试表明,在普通的PC机器(P4 2.0)上,可以做到0.5秒处理一个数据包,如果优化代码和服务器,还有较大的性能提升空间。
上面的代码是笔者实现的省级公路交通流量数据服务中心(DSC)项目中的接收服务器框架部分,整个系统还包括:数据转发交通部的转发服务器、数据远程查询客户端、综合报表数据处理系统、数据在线发布系统、系统运行监控系统等。
实际的接收服务器类及其辅助类超过3K行,整个系统则超过了60K。因为是早期实现的程序,难免有代码粗糙、方法欠妥的感觉,只有留待下个版本完善扩充了。由于与甲方有保密合同和版权保护等,不可能公开全部源代码,删减也有不当之处,读者发现时请不吝指正。下面是带详细注释的代码下载URL。
|