cyf/lserp_cs_6.0
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

基线 SVN r240

Browse Source 
SVN-Revision: r240
This commit is contained in:
cyf
2025-02-06 06:46:06 +00:00
commit ab56a9bcf7
5317 changed files with 902274 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
插件库/Lskj.SocketService/AsyncSocketCore/AsyncSocketServer.cs
+427
View File
@@ -0,0 +1,427 @@
using System;
using System.Text;
using System.Net;
using System.Net.Sockets;
using System.Threading;
namespace Lskj.SocketService
{
/// <summary>
/// 服务入口，建立Socket监听，负责接收连接，绑定连接对象，处理异步事件返回的接收和发送事件
/// </summary>
public class AsyncSocketServer
{
private Socket _listenSocket;
/// <summary>
/// 最大支持连接个数
/// </summary>
private int _numConnections;
/// <summary>
/// 每个连接接收缓存大小
/// </summary>
private int _receiveBufferSize;
/// <summary>
/// 限制访问接收连接的线程数，用来控制最大并发数
/// </summary>
private Semaphore _maxNumberAcceptedClients;
/// <summary>
/// Socket最大超时时间，单位为MS
/// </summary>
private int _socketTimeOutMS;
public int SocketTimeOutMS { get { return _socketTimeOutMS; } set { _socketTimeOutMS = value; } }
private AsyncSocketUserTokenPool _asyncSocketUserTokenPool;
private AsyncSocketUserTokenList _asyncSocketUserTokenList;
private LogOutputSocketProtocolMgr _logOutputSocketProtocolMgr;
private UploadSocketProtocolMgr _uploadSocketProtocolMgr;
private DownloadSocketProtocolMgr _downloadSocketProtocolMgr;
/// <summary>
/// 守护进程，用于关闭超时连接
/// </summary>
private DaemonThread _daemonThread;
/// <summary>
/// 管理所有正在执行的AsyncSocketUserToken，是一个列表
/// </summary>
/// <value>The asynchronous socket user token list.</value>
public AsyncSocketUserTokenList AsyncSocketUserTokenList
{
get { return _asyncSocketUserTokenList; }
}
/// <summary>
/// LogOutputSocketProtocol的管理对象
/// </summary>
/// <value>The log output socket protocol MGR.</value>
public LogOutputSocketProtocolMgr LogOutputSocketProtocolMgr
{
get { return _logOutputSocketProtocolMgr; }
}
/// <summary>
/// UploadSocketProtocol的管理对象，用于检测是否同时上传同一个文件
/// </summary>
/// <value>The upload socket protocol MGR.</value>
public UploadSocketProtocolMgr UploadSocketProtocolMgr { get { return _uploadSocketProtocolMgr; } }
/// <summary>
/// DownloadSocketProtocol的管理对象
/// </summary>
/// <value>The download socket protocol MGR.</value>
public DownloadSocketProtocolMgr DownloadSocketProtocolMgr { get { return _downloadSocketProtocolMgr; } }
/// <summary>
/// <para>说明：处理Socket客户端发送的请求</para>
/// <para>创建人：龚宇超</para>
/// <para>创建日期：2018-01-24 </para>
/// <para>修改人：</para>
/// <para>修改日期：</para>
/// <para>修改备注：</para>
/// <para>版本：1.0</para>
/// </summary>
/// <param name="sender">The source of the event.</param>
/// <param name="acceptEventArgs">The <see cref="SocketAsyncEventArgs"/> instance containing the event data.</param>
private void AcceptEventArg_Completed(object sender, SocketAsyncEventArgs acceptEventArgs)
{
try
{
ProcessAccept(acceptEventArgs);
}
catch (Exception E)
{
Program.Logger.ErrorFormat("Accept client {0} error, message: {1}", acceptEventArgs.AcceptSocket, E.Message);
Program.Logger.Error(E.StackTrace);
}
}
/// <summary>
/// <para>说明：完成异步操作的事件</para>
/// <para>创建人：龚宇超</para>
/// <para>创建日期：2018-01-24 </para>
/// <para>修改人：</para>
/// <para>修改日期：</para>
/// <para>修改备注：</para>
/// <para>版本：1.0</para>
/// </summary>
/// <param name="sender">The source of the event.</param>
/// <param name="asyncEventArgs">The <see cref="SocketAsyncEventArgs"/> instance containing the event data.</param>
/// <exception cref="System.ArgumentException">The last operation completed on the socket was not a receive or send</exception>
private void IO_Completed(object sender, SocketAsyncEventArgs asyncEventArgs)
{
AsyncSocketUserToken userToken = asyncEventArgs.UserToken as AsyncSocketUserToken;
userToken.ActiveDateTime = DateTime.Now;
try
{
lock (userToken)
{
if (asyncEventArgs.LastOperation == SocketAsyncOperation.Receive)
ProcessReceive(asyncEventArgs);
else if (asyncEventArgs.LastOperation == SocketAsyncOperation.Send)
ProcessSend(asyncEventArgs);
else
throw new ArgumentException("The last operation completed on the socket was not a receive or send");
}
}
catch (Exception E)
{
Program.Logger.ErrorFormat("IO_Completed {0} error, message: {1}", userToken.ConnectSocket, E.Message);
Program.Logger.Error(E.StackTrace);
}
}
/// <summary>
/// <para>说明：创建对应传输协议</para>
/// <para>创建人：龚宇超</para>
/// <para>创建日期：2018-01-24 </para>
/// <para>修改人：</para>
/// <para>修改日期：</para>
/// <para>修改备注：</para>
/// <para>版本：1.0</para>
/// </summary>
/// <param name="userToken">The user token.</param>
private void BuildingSocketInvokeElement(AsyncSocketUserToken userToken)
{
byte flag = userToken.ReceiveEventArgs.Buffer[userToken.ReceiveEventArgs.Offset];
if (flag == (byte)ProtocolFlag.Upload)
userToken.AsyncSocketInvokeElement = new UploadSocketProtocol(this, userToken);
else if (flag == (byte)ProtocolFlag.Download)
userToken.AsyncSocketInvokeElement = new DownloadSocketProtocol(this, userToken);
else if (flag == (byte)ProtocolFlag.RemoteStream)
userToken.AsyncSocketInvokeElement = new RemoteStreamSocketProtocol(this, userToken);
else if (flag == (byte)ProtocolFlag.Throughput)
userToken.AsyncSocketInvokeElement = new ThroughputSocketProtocol(this, userToken);
else if (flag == (byte)ProtocolFlag.Control)
userToken.AsyncSocketInvokeElement = new ControlSocketProtocol(this, userToken);
else if (flag == (byte)ProtocolFlag.LogOutput)
userToken.AsyncSocketInvokeElement = new LogOutputSocketProtocol(this, userToken);
if (userToken.AsyncSocketInvokeElement != null)
{
Program.Logger.InfoFormat("Building socket invoke element {0}.Local Address: {1}, Remote Address: {2}",
userToken.AsyncSocketInvokeElement, userToken.ConnectSocket.LocalEndPoint, userToken.ConnectSocket.RemoteEndPoint);
}
}
/// <summary>
/// <para>说明：处理接收请求</para>
/// <para>创建人：龚宇超</para>
/// <para>创建日期：2018-01-24 </para>
/// <para>修改人：</para>
/// <para>修改日期：</para>
/// <para>修改备注：</para>
/// <para>版本：1.0</para>
/// </summary>
/// <param name="acceptEventArgs">The <see cref="SocketAsyncEventArgs"/> instance containing the event data.</param>
private void ProcessAccept(SocketAsyncEventArgs acceptEventArgs)
{
Program.Logger.InfoFormat("Client connection accepted. Local Address: {0}, Remote Address: {1}",
acceptEventArgs.AcceptSocket.LocalEndPoint, acceptEventArgs.AcceptSocket.RemoteEndPoint);
AsyncSocketUserToken userToken = _asyncSocketUserTokenPool.Pop();
_asyncSocketUserTokenList.Add(userToken); // 添加到正在连接列表
userToken.ConnectSocket = acceptEventArgs.AcceptSocket;
userToken.ConnectDateTime = DateTime.Now;
try
{
bool willRaiseEvent = userToken.ConnectSocket.ReceiveAsync(userToken.ReceiveEventArgs); // 投递接收请求
if (!willRaiseEvent)
{
lock (userToken)
{
ProcessReceive(userToken.ReceiveEventArgs);
}
}
}
catch (Exception E)
{
Program.Logger.ErrorFormat("Accept client {0} error, message: {1}", userToken.ConnectSocket, E.Message);
Program.Logger.Error(E.StackTrace);
}
StartAccept(acceptEventArgs); //把当前异步事件释放，等待下次连接
}
/// <summary>
/// <para>说明：处理接收的套接字</para>
/// <para>创建人：龚宇超</para>
/// <para>创建日期：2018-01-24 </para>
/// <para>修改人：</para>
/// <para>修改日期：</para>
/// <para>修改备注：</para>
/// <para>版本：1.0</para>
/// </summary>
/// <param name="receiveEventArgs">The <see cref="SocketAsyncEventArgs"/> instance containing the event data.</param>
private void ProcessReceive(SocketAsyncEventArgs receiveEventArgs)
{
AsyncSocketUserToken userToken = receiveEventArgs.UserToken as AsyncSocketUserToken;
if (userToken.ConnectSocket == null)
return;
userToken.ActiveDateTime = DateTime.Now;
if (userToken.ReceiveEventArgs.BytesTransferred > 0 && userToken.ReceiveEventArgs.SocketError == SocketError.Success)
{
int offset = userToken.ReceiveEventArgs.Offset;
int count = userToken.ReceiveEventArgs.BytesTransferred;
if ((userToken.AsyncSocketInvokeElement == null) & (userToken.ConnectSocket != null)) //存在Socket对象，并且没有绑定协议对象，则进行协议对象绑定
{
BuildingSocketInvokeElement(userToken);
offset = offset + 1;
count = count - 1;
}
if (userToken.AsyncSocketInvokeElement == null) //如果没有解析对象，提示非法连接并关闭连接
{
Program.Logger.WarnFormat("Illegal client connection. Local Address: {0}, Remote Address: {1}", userToken.ConnectSocket.LocalEndPoint,
userToken.ConnectSocket.RemoteEndPoint);
CloseClientSocket(userToken);
}
else
{
if (count > 0) //处理接收数据
{
if (!userToken.AsyncSocketInvokeElement.ProcessReceive(userToken.ReceiveEventArgs.Buffer, offset, count))
{ // 如果处理数据返回失败，则断开连接
CloseClientSocket(userToken);
}
else //否则投递下次介绍数据请求
{
bool willRaiseEvent = userToken.ConnectSocket.ReceiveAsync(userToken.ReceiveEventArgs); //投递接收请求
if (!willRaiseEvent)
ProcessReceive(userToken.ReceiveEventArgs);
}
}
else
{
bool willRaiseEvent = userToken.ConnectSocket.ReceiveAsync(userToken.ReceiveEventArgs); //投递接收请求
if (!willRaiseEvent)
ProcessReceive(userToken.ReceiveEventArgs);
}
}
}
else
{
CloseClientSocket(userToken);
}
}
/// <summary>
/// <para>说明：发送指令</para>
/// <para>创建人：龚宇超</para>
/// <para>创建日期：2018-01-24 </para>
/// <para>修改人：</para>
/// <para>修改日期：</para>
/// <para>修改备注：</para>
/// <para>版本：1.0</para>
/// </summary>
/// <param name="sendEventArgs">The <see cref="SocketAsyncEventArgs"/> instance containing the event data.</param>
/// <returns><c>true</c> if XXXX, <c>false</c> otherwise.</returns>
private bool ProcessSend(SocketAsyncEventArgs sendEventArgs)
{
AsyncSocketUserToken userToken = sendEventArgs.UserToken as AsyncSocketUserToken;
if (userToken.AsyncSocketInvokeElement == null)
return false;
userToken.ActiveDateTime = DateTime.Now;
if (sendEventArgs.SocketError == SocketError.Success)
return userToken.AsyncSocketInvokeElement.SendCompleted(); //调用子类回调函数
else
{
CloseClientSocket(userToken);
return false;
}
}
public AsyncSocketServer(int numConnections)
{
_numConnections = numConnections;
_receiveBufferSize = ProtocolConst.ReceiveBufferSize;
_asyncSocketUserTokenPool = new AsyncSocketUserTokenPool(numConnections);
_asyncSocketUserTokenList = new AsyncSocketUserTokenList();
_maxNumberAcceptedClients = new Semaphore(numConnections, numConnections);
_logOutputSocketProtocolMgr = new LogOutputSocketProtocolMgr();
_uploadSocketProtocolMgr = new UploadSocketProtocolMgr();
_downloadSocketProtocolMgr = new DownloadSocketProtocolMgr();
}
/// <summary>
/// <para>说明：初始化</para>
/// <para>创建人：龚宇超</para>
/// <para>创建日期：2018-01-24 </para>
/// <para>修改人：</para>
/// <para>修改日期：</para>
/// <para>修改备注：</para>
/// <para>版本：1.0</para>
/// </summary>
public void Init()
{
AsyncSocketUserToken userToken;
for (int i = 0; i < _numConnections; i++) // 按照连接数建立读写对象
{
userToken = new AsyncSocketUserToken(_receiveBufferSize);
userToken.ReceiveEventArgs.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
userToken.SendEventArgs.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
_asyncSocketUserTokenPool.Push(userToken);
}
}
/// <summary>
/// <para>说明：启动Socket服务监听</para>
/// <para>创建人：龚宇超</para>
/// <para>创建日期：2018-01-24 </para>
/// <para>修改人：</para>
/// <para>修改日期：</para>
/// <para>修改备注：</para>
/// <para>版本：1.0</para>
/// </summary>
/// <param name="localEndPoint">The local end point.</param>
public void Start(IPEndPoint localEndPoint)
{
_listenSocket = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
_listenSocket.Bind(localEndPoint);
_listenSocket.Listen(_numConnections);
Program.Logger.InfoFormat("Start listen socket {0} success", localEndPoint.ToString());
StartAccept(null);
_daemonThread = new DaemonThread(this);
}
/// <summary>
/// <para>说明：释放异步事件</para>
/// <para>创建人：龚宇超</para>
/// <para>创建日期：2018-01-24 </para>
/// <para>修改人：</para>
/// <para>修改日期：</para>
/// <para>修改备注：</para>
/// <para>版本：1.0</para>
/// </summary>
/// <param name="acceptEventArgs">The <see cref="SocketAsyncEventArgs"/> instance containing the event data.</param>
public void StartAccept(SocketAsyncEventArgs acceptEventArgs)
{
if (acceptEventArgs == null)
{
acceptEventArgs = new SocketAsyncEventArgs();
acceptEventArgs.Completed += new EventHandler<SocketAsyncEventArgs>(AcceptEventArg_Completed);
}
else
{
acceptEventArgs.AcceptSocket = null; // 释放上次绑定的Socket，等待下一个Socket连接
}
_maxNumberAcceptedClients.WaitOne(); //获取信号量
bool willRaiseEvent = _listenSocket.AcceptAsync(acceptEventArgs);
if (!willRaiseEvent)
{
ProcessAccept(acceptEventArgs);
}
}
/// <summary>
/// <para>说明：发送异步数据</para>
/// <para>创建人：龚宇超</para>
/// <para>创建日期：2018-01-24 </para>
/// <para>修改人：</para>
/// <para>修改日期：</para>
/// <para>修改备注：</para>
/// <para>版本：1.0</para>
/// </summary>
/// <param name="connectSocket">The connect socket.</param>
/// <param name="sendEventArgs">The <see cref="SocketAsyncEventArgs"/> instance containing the event data.</param>
/// <param name="buffer">The buffer.</param>
/// <param name="offset">The offset.</param>
/// <param name="count">The count.</param>
/// <returns><c>true</c> if XXXX, <c>false</c> otherwise.</returns>
public bool SendAsyncEvent(Socket connectSocket, SocketAsyncEventArgs sendEventArgs, byte[] buffer, int offset, int count)
{
if (connectSocket == null)
return false;
sendEventArgs.SetBuffer(buffer, offset, count);
bool willRaiseEvent = connectSocket.SendAsync(sendEventArgs);
if (!willRaiseEvent)
{
return ProcessSend(sendEventArgs);
}
else
return true;
}
/// <summary>
/// <para>说明：关闭指定的Socket连接</para>
/// <para>创建人：龚宇超</para>
/// <para>创建日期：2018-01-24 </para>
/// <para>修改人：</para>
/// <para>修改日期：</para>
/// <para>修改备注：</para>
/// <para>版本：1.0</para>
/// </summary>
/// <param name="userToken">The user token.</param>
public void CloseClientSocket(AsyncSocketUserToken userToken)
{
if (userToken.ConnectSocket == null)
return;
string socketInfo = string.Format("Local Address: {0} Remote Address: {1}", userToken.ConnectSocket.LocalEndPoint,
userToken.ConnectSocket.RemoteEndPoint);
Program.Logger.InfoFormat("Client connection disconnected. {0}", socketInfo);
try
{
userToken.ConnectSocket.Shutdown(SocketShutdown.Both);
}
catch (Exception E)
{
Program.Logger.ErrorFormat("CloseClientSocket Disconnect client {0} error, message: {1}", socketInfo, E.Message);
}
userToken.ConnectSocket.Close();
userToken.ConnectSocket = null; //释放引用，并清理缓存，包括释放协议对象等资源
_maxNumberAcceptedClients.Release();
_asyncSocketUserTokenPool.Push(userToken);
_asyncSocketUserTokenList.Remove(userToken);
}
}
}