using System;
using System.Text;
using System.Collections;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.Runtime.InteropServices;
using System.IO;
namespace Common.TcpLib
{
/// <summary>
/// 异步Socket处理服务器端
/// </summary>
public class TcpServerY : Common.TcpLib.ITcpServer
{
private Socket _listener;
private string _listenerIp = "127.0.0.1";
private int _port;
private List<_Bgz_ConnectionState> _connections = new List<_Bgz_ConnectionState>();
private int _maxConnections = 2000;
private int _backlog = 100;
private int _bufferSize = 1024;
private AsyncCallback ConnectionReady;
private WaitCallback AcceptConnection;
private AsyncCallback ReceivedDataReady;
private ManualResetEvent sendDone = new ManualResetEvent(false);
private bool _disposed;
private bool _debug = false;
private object _stopLock = new object();
private bool _sotpMessage = false;
#region define delegates
/// <summary>
/// 服务器绑定后处理事件
/// </summary>
public _Bgz_OnBindEventDelegate FOnBindEventDelegate;
/// <summary>
/// Accept后处理事件
/// </summary>
public _Bgz_OnAcceptEventDelegate FOnAcceptEventDelegate;
/// <summary>
/// 接收到客户端数据后处理事件
/// </summary>
public _Bgz_OnReceiveBeginEventDelegate FOnReceiveBeginEventDelegate;
public _Bgz_OnReceiveingEventDelegate FOnReceiveingEventDelegate;
public _Bgz_OnReceiveEndEventDelegate FOnReceiveEndEventDelegate;
/// <summary>
/// 报错信息处理
/// </summary>
public _Bgz_OnErrorEventDelegate FOnErrorEventDelegate;
/// <summary>
/// 断掉与客户机的连接前处理事件
/// </summary>
public _Bgz_OnDropConnectEventDelegate FOnDropConnectEventDelegate;
#endregion
#region Event
private void OnBindEvent(_Bgz_ConnectionState state)
{
if (FOnBindEventDelegate != null) FOnBindEventDelegate(state);
}
private void OnAcceptEvent(_Bgz_ConnectionState state)
{
if (FOnAcceptEventDelegate != null) FOnAcceptEventDelegate(state);
}
private void OnReceiveBeginEvent(_Bgz_ConnectionState state)
{
if (FOnReceiveBeginEventDelegate != null) FOnReceiveBeginEventDelegate(state);
}
private void OnReceiveingEvent(_Bgz_ConnectionState state)
{
if (FOnReceiveingEventDelegate != null) FOnReceiveingEventDelegate(state);
}
private void OnReceiveEndEvent(_Bgz_ConnectionState state)
{
if (FOnReceiveEndEventDelegate != null) FOnReceiveEndEventDelegate(state);
}
private void OnErrorEvent(ErrorType errortype, string msg, _Bgz_ConnectionState state)
{
if (FOnErrorEventDelegate != null) FOnErrorEventDelegate(errortype, msg, state);
}
private void OnDropConnectEvent(_Bgz_ConnectionState state)
{
if (FOnDropConnectEventDelegate != null) FOnDropConnectEventDelegate(state);
}
#endregion
#region Constructor and Destructor
public TcpServerY(int port)
{
_port = port;
}
public TcpServerY(int port, int backlog)
{
_port = port;
_backlog = backlog;
}
public TcpServerY(int port, int backlog, int maxConnection)
{
_port = port;
_backlog = backlog;
_maxConnections = maxConnection;
}
public TcpServerY(int port, int backlog, int maxConnection, int buffterSize)
{
_port = port;
_backlog = backlog;
_maxConnections = maxConnection;
_bufferSize = buffterSize;
}
public TcpServerY(string listenerIp, int port)
{
_listenerIp = listenerIp;
_port = port;
}
public TcpServerY(string listenerIp, int port, int backlog)
{
_listenerIp = listenerIp;
_port = port;
_backlog = backlog;
}
public TcpServerY(string listenerIp, int port, int backlog, int maxConnection)
{
_listenerIp = listenerIp;
_port = port;
_backlog = backlog;
_maxConnections = maxConnection;
}
public TcpServerY(string listenerIp, int port, int backlog, int maxConnection, int buffterSize)
{
_listenerIp = listenerIp;
_port = port;
_backlog = backlog;
_maxConnections = maxConnection;
_bufferSize = buffterSize;
}
~TcpServerY()
{
if (!_disposed)
{
Stop();
}
}
#endregion
#region Private Methods
private void Dispose()
{
if (!_disposed)
{
Stop();
GC.Collect();
GC.WaitForPendingFinalizers();
this.FOnAcceptEventDelegate = null;
this.FOnBindEventDelegate = null;
this.FOnDropConnectEventDelegate = null;
this.FOnErrorEventDelegate = null;
this.FOnReceiveBeginEventDelegate = null;
this.FOnReceiveEndEventDelegate = null;
this.FOnReceiveingEventDelegate = null;
_disposed = true;
}
}
private void ConnectionReady_Handler(IAsyncResult ar)
{
try
{
lock (this)
{
_Bgz_ConnectionState stx = new _Bgz_ConnectionState();
if (_listener == null)
{
lock (_stopLock)
{
if (!_sotpMessage)
{
OnErrorEvent(ErrorType.DisAccept, "服务器关闭了监听！", null);
_sotpMessage = true;
}
}
return;
}
stx._conn = _listener.EndAccept(ar);
#region 处理Socket网络异常断开
uint dummy = 0;
byte[] inOptionValues = new byte[Marshal.SizeOf(dummy) * 3];
BitConverter.GetBytes((uint)1).CopyTo(inOptionValues, 0);
BitConverter.GetBytes((uint)5000).CopyTo(inOptionValues, Marshal.SizeOf(dummy));
BitConverter.GetBytes((uint)5000).CopyTo(inOptionValues, Marshal.SizeOf(dummy) * 2);
_listener.IOControl(IOControlCode.KeepAliveValues, inOptionValues, null);
#endregion
if (_connections.Count >= _maxConnections)
{
OnErrorEvent(ErrorType.MaxConnect, "Server busy！", stx);
DropConnection(stx);
}
else
{
_connections.Add(stx);
ThreadPool.QueueUserWorkItem(AcceptConnection, stx);
}
_listener.BeginAccept(ConnectionReady, null);
}
}
catch (Exception ex)
{
if (Debug)
{
OnErrorEvent(ErrorType.Catch, "ConnectionReady Error！[Message]:\r\n" + ex.Message + "[StackTrace]:\r\n" + ex.StackTrace + "\r\n", null);
}
}
}
private void AcceptConnection_Handler(object state)
{
try
{
_Bgz_ConnectionState stx = state as _Bgz_ConnectionState;
try
{
OnAcceptEvent(stx);
if (stx.Connected)
stx._conn.BeginReceive(stx._buffer, 0, 0, SocketFlags.None, ReceivedDataReady, stx);
}
catch (Exception ex)
{
DropConnection(stx);
if (Debug)
{
OnErrorEvent(ErrorType.Catch, "AcceptConnection.1 Error！[Message]:\r\n" + ex.Message + "[StackTrace]:\r\n" + ex.StackTrace + "\r\n", stx);
}
}
}
catch (Exception ex)
{
if (Debug)
{
OnErrorEvent(ErrorType.Catch, "AcceptConnection.2 Error！[Message]:\r\n" + ex.Message + "[StackTrace]:\r\n" + ex.StackTrace + "\r\n", null);
}
}
}
private void ReceivedDataReady_Handler(IAsyncResult ar)
{
try
{
_Bgz_ConnectionState stx = ar.AsyncState as _Bgz_ConnectionState;
if (this._listener == null)
{
lock (_stopLock)
{
if (!_sotpMessage)
{
OnErrorEvent(ErrorType.DisAccept, "服务器关闭了监听！", null);
_sotpMessage = true;
}
}
return;
}
if (stx._conn == null) return;
stx._conn.EndReceive(ar);
if (stx.AvailableData == 0)
{
DropConnection(stx);
return;
}
try
{
lock (this)
{
OnReceiveBeginEvent(stx);
#region receive
stx._buffer = new byte[0];
stx._count = 0;
stx._dataStream.SetLength(0);
stx._dataStream.Position = 0;
if (stx._getonceall)
{
stx._buffer = new byte[stx.AvailableData];
int ret = stx.Read(stx._buffer, 0, stx._buffer.Length);
if (ret > 0)
{
stx._dataStream.Write(stx._buffer, 0, stx._buffer.Length);
stx._count++;
OnReceiveingEvent(stx);
}
}
else
{
while (stx.AvailableData > 0)
{
if (stx.AvailableData > _bufferSize)
stx._buffer = new byte[_bufferSize];
else
stx._buffer = new byte[stx.AvailableData];
int ret = stx.Read(stx._buffer, 0, stx._buffer.Length);
if (ret > 0)
{
stx._dataStream.Write(stx._buffer, 0, stx._buffer.Length);
stx._count++;
OnReceiveingEvent(stx);
}
}
}
#endregion
OnReceiveEndEvent(stx);
}
if (stx._conn == null) return;
if (stx._conn.Connected)
{
stx._buffer = new byte[0];
stx._count = 0;
stx._conn.BeginReceive(stx._buffer, 0, 0, SocketFlags.None, ReceivedDataReady, stx);
}
}
catch (Exception ex)
{
if (this._listener == null)
{
lock (_stopLock)
{
if (!_sotpMessage)
{
OnErrorEvent(ErrorType.DisAccept, "服务器关闭了监听！", null);
_sotpMessage = true;
}
}
return;
}
if (stx._conn != null)
DropConnection(stx);
if (Debug)
{
OnErrorEvent(ErrorType.Catch, "ReceivedDataReady.1 Error！[Message]:\r\n" + ex.Message + "[StackTrace]:\r\n" + ex.StackTrace + "\r\n", stx);
}
}
}
catch (Exception ex)
{
if (Debug)
{
OnErrorEvent(ErrorType.Catch, "ReceivedDataReady.2 Error！[Message]:\r\n" + ex.Message + "[StackTrace]:\r\n" + ex.StackTrace + "\r\n", null);
}
}
}
private void DropConnection(_Bgz_ConnectionState stx)
{
try
{
lock (this)
{
if (_listener == null)
{
lock (_stopLock)
{
if (!_sotpMessage)
{
OnErrorEvent(ErrorType.DisAccept, "服务器关闭了监听！", null);
_sotpMessage = true;
}
}
return;
}
OnDropConnectEvent(stx);
if (_connections.Contains(stx)) _connections.Remove(stx);
if (stx._conn != null)
{
IPEndPoint info = (IPEndPoint)stx._conn.RemoteEndPoint;
string str = info.Address.ToString() + ":" + info.Port.ToString();
OnErrorEvent(ErrorType.DisAccept, "关闭了远程连接[" + str + "]！", null);
if (stx._conn.Connected)
stx._conn.Close();
}
}
}
catch (Exception ex)
{
if (Debug)
{
OnErrorEvent(ErrorType.Catch, "DropConnection Error！[Message]:\r\n" + ex.Message + "[StackTrace]:\r\n" + ex.StackTrace + "\r\n", null);
}
}
}
private void SendCallback(IAsyncResult ar)
{
try
{
Socket client = (Socket)ar.AsyncState;
int bytesSent = client.EndSend(ar);
}
catch (Exception ex)
{
if (Debug)
{
OnErrorEvent(ErrorType.Catch, "Send Error！[Message]:\r\n" + ex.Message + "[StackTrace]:\r\n" + ex.StackTrace + "\r\n", null);
}
}
finally
{
sendDone.Set();
}
}
#endregion
#region Public Methods
public void Start()
{
Start(false);
}
public void Start(bool optionValue)
{
try
{
_listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
#region 处理Socket端口复用
_listener.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, optionValue);
#endregion
_connections = new List<_Bgz_ConnectionState>(); ;
ConnectionReady = new AsyncCallback(ConnectionReady_Handler);
AcceptConnection = new WaitCallback(AcceptConnection_Handler);
ReceivedDataReady = new AsyncCallback(ReceivedDataReady_Handler);
_listener.Bind(new IPEndPoint(IPAddress.Parse(_listenerIp), _port));
_listener.Listen(_backlog);
_Bgz_ConnectionState stx = new _Bgz_ConnectionState(_listener);
OnBindEvent(stx);
_sotpMessage = false;
_stopLock = new object();
_listener.BeginAccept(ConnectionReady, null);
}
catch (Exception ex)
{
if (Debug)
{
OnErrorEvent(ErrorType.Catch, "Start Error！[Message]:\r\n" + ex.Message + "[StackTrace]:\r\n" + ex.StackTrace + "\r\n", null);
}
}
}
public void Send(byte[] msg)
{
try
{
foreach (Socket s in ClientSockets)
{
Send(s, msg);
}
}
catch (Exception ex)
{
if (Debug)
{
OnErrorEvent(ErrorType.Catch, "Send Error！[Message]:\r\n" + ex.Message + "[StackTrace]:\r\n" + ex.StackTrace + "\r\n", null);
}
}
}
public void Send(Socket ClientSocket, byte[] msg)
{
try
{
if (ClientSocket.Connected == false)
{
OnErrorEvent(ErrorType.DisConnect, "未与服务器建立链接，导致发送数据失败！", null);
}
ClientSocket.Send(msg);
}
catch (Exception ex)
{
if (Debug)
{
OnErrorEvent(ErrorType.Catch, "Send Error！[Message]:\r\n" + ex.Message + "[StackTrace]:\r\n" + ex.StackTrace + "\r\n", null);
}
}
}
public void SendAsync(Socket ClientSocket, byte[] msg)
{
try
{
if (ClientSocket.Connected == false)
{
throw (new Exception("没有连接客户端不可以发送信息!"));
}
byte[] buffer;
int count = msg.Length;
int cur = 0;
while (count > 0)
{
if (count > this._bufferSize)
buffer = new byte[this._bufferSize];
else
buffer = new byte[count];
for (int i = 0; i < buffer.Length; i++)
{
buffer[i] = msg[cur + i];
}
count -= buffer.Length;
cur += buffer.Length;
ClientSocket.BeginSend(buffer, 0, buffer.Length, 0, new AsyncCallback(SendCallback), ClientSocket);
sendDone.WaitOne();
}
}
catch (Exception ex)
{
if (Debug)
{
OnErrorEvent(ErrorType.Catch, "Send Error！[Message]:\r\n" + ex.Message + "[StackTrace]:\r\n" + ex.StackTrace + "\r\n", null);
}
}
}
public void SendAsync(Socket ClientSocket, Stream Astream)
{
try
{
if (ClientSocket.Connected == false)
{
throw (new Exception("没有连接客户端不可以发送信息!"));
}
byte[] buffer;
long count = Astream.Length;
while (count > 0)
{
if (count > this._bufferSize)
buffer = new byte[this._bufferSize];
else
buffer = new byte[count];
Astream.Read(buffer, 0, buffer.Length);
count -= buffer.Length;
ClientSocket.BeginSend(buffer, 0, buffer.Length, 0, new AsyncCallback(SendCallback), ClientSocket);
sendDone.WaitOne();
}
}
catch (Exception ex)
{
if (Debug)
{
OnErrorEvent(ErrorType.Catch, "Send Error！[Message]:\r\n" + ex.Message + "[StackTrace]:\r\n" + ex.StackTrace + "\r\n", null);
}
}
}
public void Stop()
{
try
{
lock (this)
{
foreach (object obj in _connections)
{
_Bgz_ConnectionState stx = obj as _Bgz_ConnectionState;
OnDropConnectEvent(stx);
stx._conn.Shutdown(SocketShutdown.Both);
stx._conn.Close();
}
_connections.Clear();
_listener.Close();
_listener = null;
}
}
catch (Exception ex)
{
if (Debug)
{
OnErrorEvent(ErrorType.Catch, "Stop Error！[Message]:\r\n" + ex.Message + "[StackTrace]:\r\n" + ex.StackTrace + "\r\n", null);
}
}
}
#endregion
#region property
/// <summary>
/// 获取最大的连接池大小
/// </summary>
public int MaxConnections
{
get
{
return _maxConnections;
}
}
/// <summary>
/// 获取挂起连接队列的最大长度
/// </summary>
public int BackLog
{
get
{
return _backlog;
}
}
public int BufferSize
{
get
{
return _bufferSize;
}
}
/// <summary>
/// 获取当前连接池的容量
/// </summary>
public int CurrentConnections
{
get
{
lock (this) { return _connections.Count; }
}
}
public Socket[] ClientSockets
{
get
{
Socket[] sockets = new Socket[this._connections.Count];
for (int i = 0; i < this._connections.Count; i++)
{
sockets[i] = this._connections[i]._conn;
}
return sockets;
}
}
public bool Debug
{
get
{
return _debug;
}
set
{
_debug = value;
}
}
#endregion
}
}
==========================================================
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Text;
using System.Windows.Forms;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using Common.TcpLib;
using Common;
namespace WindowsServer
{
public partial class ServerY : Form
{
ThreadView tv;
TcpServerY tcpserver;
public ServerY()
{
InitializeComponent();
tv = new ThreadView(this,0);
tv.OnUpdateUIEventDelegate = new _Bgz_OnUpdateUI(_UpdateUI);
}
private void btConnect_Click(object sender, EventArgs e)
{
tcpserver = new TcpServerY(int.Parse(this.textBox1.Text));
tcpserver.FOnAcceptEventDelegate = new _Bgz_OnAcceptEventDelegate(_OnAccept);
tcpserver.FOnBindEventDelegate = new _Bgz_OnBindEventDelegate(_OnBind);
tcpserver.FOnDropConnectEventDelegate = new _Bgz_OnDropConnectEventDelegate(_OnDropConnect);
tcpserver.FOnErrorEventDelegate = new _Bgz_OnErrorEventDelegate(_OnError);
tcpserver.FOnReceiveBeginEventDelegate = new _Bgz_OnReceiveBeginEventDelegate(_OnReceiveBegin);
tcpserver.FOnReceiveingEventDelegate = new _Bgz_OnReceiveingEventDelegate(_OnReceiveing);
tcpserver.FOnReceiveEndEventDelegate = new _Bgz_OnReceiveEndEventDelegate(_OnReceiveEnd);
tcpserver.Debug = true;
tcpserver.Start(true);
this.btConnect.Enabled = false;
}
private void btClose_Click(object sender, EventArgs e)
{
tcpserver.Stop();
this.btConnect.Enabled = true;
}
public void _OnAccept(_Bgz_ConnectionState state)
{
string str = "远程端口[" + state.StrRemoteEndPoint + "]访问服务器!#";
tv.ShowState(str, "rtbReceiveMessage");
}
public void _OnBind(_Bgz_ConnectionState state)
{
string str = "建立服务器监听[" + state.StrLocalEndPoint + "]#";
tv.ShowState( str , "rtbReceiveMessage");
}
public void _OnDropConnect(_Bgz_ConnectionState state)
{
//string str = "关闭了["+state.StrRemoteEndPoint +"]的连接！#";
//tv.ShowState(str , "rtbReceiveMessage");
}
public void _OnError(ErrorType errortype, string errormsg, _Bgz_ConnectionState state)
{
tv.ShowState(errormsg, "rtbReceiveMessage");
}
public void _OnReceiveBegin(_Bgz_ConnectionState state)
{
//设置接收数据包的处理
//count： 包的接收次数
//getonceall: 是否一次性接收 如果设置为false则接所设置的默认缓冲区大小bufferSize分批次接收
state._count = 0;
state._getonceall = true;
string str = ">>>>>>> Receive Begin from [" + state.StrRemoteEndPoint + "]#";
tv.ShowState(str, "rtbReceiveMessage");
}
public void _OnReceiveing(_Bgz_ConnectionState state)
{
//如非必要建议不要在此处理 服务器接收数据片段
//string str = Encoding.UTF8.GetString(state._buffer);
//tv.ShowState(state._count.ToString() + ">>>", "rtbReceiveMessage");
//tv.ShowState(str, "rtbReceiveMessage");
}
long i = 0;
public void _OnReceiveEnd(_Bgz_ConnectionState state)
{
string str = Encoding.UTF8.GetString(state._buffer);
tv.ShowState(str, "rtbReceiveMessage");
tv.ShowState(">>>>>>> Receive End frm [" + state.StrRemoteEndPoint + "]#", "rtbReceiveMessage");
//回写客户端
System.Threading.Interlocked.Increment(ref i);
str = i.ToString() + ":" + str;
tcpserver.Send(Encoding.UTF8.GetBytes(str));
}
#region 跨线程窗体控件操作
private void _UpdateUI(object sender, MessageEvents e)
{
switch (e.Control)
{
case "rtbReceiveMessageApp":
this.rtbReceiveMessage.AppendText(e.Msg);
this.rtbReceiveMessage.ScrollToCaret();
break;
case "rtbReceiveMessage":
default:
this.rtbReceiveMessage.AppendText(e.Msg + "\r\n");
this.rtbReceiveMessage.ScrollToCaret();
break;
}
}
#endregion
}
}