我正在编写一个新的Windows服务应用程序，该应用程序接受长时间运行的TCP/IP连接（即不像HTTP协议那样存在许多短连接，而是客户端连接并保持连接数小时、天甚至数周）。
我正在寻找最佳设计网络架构的思路。我将需要为此服务启动至少一个线程。我正在考虑使用异步API（如BeginRecieve等），因为我不知道在任何给定时间将有多少个客户端连接（可能达到数百个）。我绝对不希望为每个连接启动一个线程。
数据主要会从服务器向客户端流出，但偶尔也会有一些来自客户端的命令发送过来。这主要是一个监控应用程序，在其中我的服务器会定期向客户端发送状态数据。
最好的方法使其尽可能具有可扩展性是什么？基本工作流程是什么？
需要明确的是，我正在寻找基于.NET的解决方案（如果可能的话，是C＃，但任何.NET语言都可以）。
我需要一个工作示例，可以是指向可下载内容的指针或内置的简短示例。它必须是基于.NET和Windows的（任何.NET语言均可接受）。
private List<xConnection> _sockets;
同时，您需要确保套接字实际上正在监听传入的连接。
private System.Net.Sockets.Socket _serverSocket;
public bool Start()
{
System.Net.IPHostEntry localhost = System.Net.Dns.GetHostEntry(System.Net.Dns.GetHostName());
System.Net.IPEndPoint serverEndPoint;
try
{
serverEndPoint = new System.Net.IPEndPoint(localhost.AddressList[0], _port);
}
catch (System.ArgumentOutOfRangeException e)
{
throw new ArgumentOutOfRangeException("Port number entered would seem to be invalid, should be between 1024 and 65000", e);
}
try
{
_serverSocket = new System.Net.Sockets.Socket(serverEndPoint.Address.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
}
catch (System.Net.Sockets.SocketException e)
{
throw new ApplicationException("Could not create socket, check to make sure not duplicating port", e);
}
try
{
_serverSocket.Bind(serverEndPoint);
_serverSocket.Listen(_backlog);
}
catch (Exception e)
{
throw new ApplicationException("An error occurred while binding socket. Check inner exception", e);
}
try
{
//warning, only call this once, this is a bug in .net 2.0 that breaks if
// you're running multiple asynch accepts, this bug may be fixed, but
// it was a major pain in the rear previously, so make sure there is only one
//BeginAccept running
_serverSocket.BeginAccept(new AsyncCallback(acceptCallback), _serverSocket);
}
catch (Exception e)
{
throw new ApplicationException("An error occurred starting listeners. Check inner exception", e);
}
return true;
}
false，如果设置了配置选项，但为了简洁起见，我想删除它。 private void acceptCallback(IAsyncResult result)
{
xConnection conn = new xConnection();
try
{
//Finish accepting the connection
System.Net.Sockets.Socket s = (System.Net.Sockets.Socket)result.AsyncState;
conn = new xConnection();
conn.socket = s.EndAccept(result);
conn.buffer = new byte[_bufferSize];
lock (_sockets)
{
_sockets.Add(conn);
}
//Queue receiving of data from the connection
conn.socket.BeginReceive(conn.buffer, 0, conn.buffer.Length, SocketFlags.None, new AsyncCallback(ReceiveCallback), conn);
//Queue the accept of the next incoming connection
_serverSocket.BeginAccept(new AsyncCallback(acceptCallback), _serverSocket);
}
catch (SocketException e)
{
if (conn.socket != null)
{
conn.socket.Close();
lock (_sockets)
{
_sockets.Remove(conn);
}
}
//Queue the next accept, think this should be here, stop attacks based on killing the waiting listeners
_serverSocket.BeginAccept(new AsyncCallback(acceptCallback), _serverSocket);
}
catch (Exception e)
{
if (conn.socket != null)
{
conn.socket.Close();
lock (_sockets)
{
_sockets.Remove(conn);
}
}
//Queue the next accept, think this should be here, stop attacks based on killing the waiting listeners
_serverSocket.BeginAccept(new AsyncCallback(acceptCallback), _serverSocket);
}
}
BeginReceive ,当客户端发送数据时会运行一个回调函数，然后排队下一个 acceptCallback 用于接受下一个客户端连接。
BeginReceive 方法告诉套接字在接收到来自客户端的数据时该做什么。对于 BeginReceive ，您需要提供一个字节数组，这是客户端发送数据时它将复制数据的位置。我们使用 ReceiveCallback 方法来处理接收到的数据。private void ReceiveCallback(IAsyncResult result)
{
//get our connection from the callback
xConnection conn = (xConnection)result.AsyncState;
//catch any errors, we'd better not have any
try
{
//Grab our buffer and count the number of bytes receives
int bytesRead = conn.socket.EndReceive(result);
//make sure we've read something, if we haven't it supposadly means that the client disconnected
if (bytesRead > 0)
{
//put whatever you want to do when you receive data here
//Queue the next receive
conn.socket.BeginReceive(conn.buffer, 0, conn.buffer.Length, SocketFlags.None, new AsyncCallback(ReceiveCallback), conn);
}
else
{
//Callback run but no data, close the connection
//supposadly means a disconnect
//and we still have to close the socket, even though we throw the event later
conn.socket.Close();
lock (_sockets)
{
_sockets.Remove(conn);
}
}
}
catch (SocketException e)
{
//Something went terribly wrong
//which shouldn't have happened
if (conn.socket != null)
{
conn.socket.Close();
lock (_sockets)
{
_sockets.Remove(conn);
}
}
}
}
编辑：在这个模式中，我忘了提到在代码的这个区域：
//put whatever you want to do when you receive data here
//Queue the next receive
conn.socket.BeginReceive(conn.buffer, 0, conn.buffer.Length, SocketFlags.None, new AsyncCallback(ReceiveCallback), conn);
通常情况下，在任何你想要的代码中，我会将数据包重新组装成消息，并将其作为工作项添加到线程池中。这样，当处理消息的代码运行时，来自客户端的下一个数据块的BeginReceive不会被延迟。
接受回调通过调用EndReceive完成了对数据套接字的读取，这填充了在BeginReceive函数中提供的缓冲区。一旦你在我的注释处完成了你想要做的事情，我们就调用下一个BeginReceive方法，如果客户端发送更多数据，它将再次运行回调函数。
现在是真正棘手的部分：当客户端发送数据时，你的接收回调可能只会调用部分消息。重新组装可能变得非常复杂。我使用了自己的方法并创建了一种类似专有协议的方式来实现这一点。我省略了它，但如果你需要，我可以添加进去。这个处理程序实际上是我写过的最复杂的代码。
public bool Send(byte[] message, xConnection conn)
{
if (conn != null && conn.socket.Connected)
{
lock (conn.socket)
{
//we use a blocking mode send, no async on the outgoing
//since this is primarily a multithreaded application, shouldn't cause problems to send in blocking mode
conn.socket.Send(bytes, bytes.Length, SocketFlags.None);
}
}
else
return false;
return true;
}
Send调用。对我来说，由于消息大小和应用程序的多线程性质，这是可以接受的。如果您想要向每个客户端发送消息，只需要循环遍历 _sockets 列表。public class xConnection : xBase
{
public byte[] buffer;
public System.Net.Sockets.Socket socket;
}
另外作为参考，这里是我经常包含的using，因为当它们没有被包含时，我总是感到很烦。
using System.Net.Sockets;
BeginAccept 方法。很久以前，.NET 中就存在一个非常恼人的 bug，但具体细节我已经记不清了。
另外，在 ReceiveCallback 代码中，我们在将数据加入队列之前处理从套接字接收到的任何内容。这意味着对于单个套接字，我们实际上只在任何时候都在 ReceiveCallback 中一次，而且我们不需要使用线程同步。然而，如果你重新排列代码以便在提取数据后立即调用下一个接收操作（这可能会更快），则需要确保正确同步线程。
此外，我删去了很多代码，但留下了核心过程。这应该是你设计的良好起点。如果你有更多关于此问题的问题，请在下面留言。
public class Telnet
{
private readonly Pool<SocketAsyncEventArgs> m_EventArgsPool;
private Socket m_ListenSocket;
/// <summary>
/// This event fires when a connection has been established.
/// </summary>
public event EventHandler<SocketAsyncEventArgs> Connected;
/// <summary>
/// This event fires when a connection has been shutdown.
/// </summary>
public event EventHandler<SocketAsyncEventArgs> Disconnected;
/// <summary>
/// This event fires when data is received on the socket.
/// </summary>
public event EventHandler<SocketAsyncEventArgs> DataReceived;
/// <summary>
/// This event fires when data is finished sending on the socket.
/// </summary>
public event EventHandler<SocketAsyncEventArgs> DataSent;
/// <summary>
/// This event fires when a line has been received.
/// </summary>
public event EventHandler<LineReceivedEventArgs> LineReceived;
/// <summary>
/// Specifies the port to listen on.
/// </summary>
[DefaultValue(23)]
public int ListenPort { get; set; }
/// <summary>
/// Constructor for Telnet class.
/// </summary>
public Telnet()
{
m_EventArgsPool = new Pool<SocketAsyncEventArgs>();
ListenPort = 23;
}
/// <summary>
/// Starts the telnet server listening and accepting data.
/// </summary>
public void Start()
{
IPEndPoint endpoint = new IPEndPoint(0, ListenPort);
m_ListenSocket = new Socket(endpoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
m_ListenSocket.Bind(endpoint);
m_ListenSocket.Listen(100);
//
// Post Accept
//
StartAccept(null);
}
/// <summary>
/// Not Yet Implemented. Should shutdown all connections gracefully.
/// </summary>
public void Stop()
{
//throw (new NotImplementedException());
}
//
// ACCEPT
//
/// <summary>
/// Posts a requests for Accepting a connection. If it is being called from the completion of
/// an AcceptAsync call, then the AcceptSocket is cleared since it will create a new one for
/// the new user.
/// </summary>
/// <param name="e">null if posted from startup, otherwise a <b>SocketAsyncEventArgs</b> for reuse.</param>
private void StartAccept(SocketAsyncEventArgs e)
{
if (e == null)
{
e = m_EventArgsPool.Pop();
e.Completed += Accept_Completed;
}
else
{
e.AcceptSocket = null;
}
if (m_ListenSocket.AcceptAsync(e) == false)
{
Accept_Completed(this, e);
}
}
/// <summary>
/// Completion callback routine for the AcceptAsync post. This will verify that the Accept occured
/// and then setup a Receive chain to begin receiving data.
/// </summary>
/// <param name="sender">object which posted the AcceptAsync</param>
/// <param name="e">Information about the Accept call.</param>
private void Accept_Completed(object sender, SocketAsyncEventArgs e)
{
//
// Socket Options
//
e.AcceptSocket.NoDelay = true;
//
// Create and setup a new connection object for this user
//
Connection connection = new Connection(this, e.AcceptSocket);
//
// Tell the client that we will be echo'ing data sent
//
DisableEcho(connection);
//
// Post the first receive
//
SocketAsyncEventArgs args = m_EventArgsPool.Pop();
args.UserToken = connection;
//
// Connect Event
//
if (Connected != null)
{
Connected(this, args);
}
args.Completed += Receive_Completed;
PostReceive(args);
//
// Post another accept
//
StartAccept(e);
}
//
// RECEIVE
//
/// <summary>
/// Post an asynchronous receive on the socket.
/// </summary>
/// <param name="e">Used to store information about the Receive call.</param>
private void PostReceive(SocketAsyncEventArgs e)
{
Connection connection = e.UserToken as Connection;
if (connection != null)
{
connection.ReceiveBuffer.EnsureCapacity(64);
e.SetBuffer(connection.ReceiveBuffer.DataBuffer, connection.ReceiveBuffer.Count, connection.ReceiveBuffer.Remaining);
if (connection.Socket.ReceiveAsync(e) == false)
{
Receive_Completed(this, e);
}
}
}
/// <summary>
/// Receive completion callback. Should verify the connection, and then notify any event listeners
/// that data has been received. For now it is always expected that the data will be handled by the
/// listeners and thus the buffer is cleared after every call.
/// </summary>
/// <param name="sender">object which posted the ReceiveAsync</param>
/// <param name="e">Information about the Receive call.</param>
private void Receive_Completed(object sender, SocketAsyncEventArgs e)
{
Connection connection = e.UserToken as Connection;
if (e.BytesTransferred == 0 || e.SocketError != SocketError.Success || connection == null)
{
Disconnect(e);
return;
}
connection.ReceiveBuffer.UpdateCount(e.BytesTransferred);
OnDataReceived(e);
HandleCommand(e);
Echo(e);
OnLineReceived(connection);
PostReceive(e);
}
/// <summary>
/// Handles Event of Data being Received.
/// </summary>
/// <param name="e">Information about the received data.</param>
protected void OnDataReceived(SocketAsyncEventArgs e)
{
if (DataReceived != null)
{
DataReceived(this, e);
}
}
/// <summary>
/// Handles Event of a Line being Received.
/// </summary>
/// <param name="connection">User connection.</param>
protected void OnLineReceived(Connection connection)
{
if (LineReceived != null)
{
int index = 0;
int start = 0;
while ((index = connection.ReceiveBuffer.IndexOf('\n', index)) != -1)
{
string s = connection.ReceiveBuffer.GetString(start, index - start - 1);
s = s.Backspace();
LineReceivedEventArgs args = new LineReceivedEventArgs(connection, s);
Delegate[] delegates = LineReceived.GetInvocationList();
foreach (Delegate d in delegates)
{
d.DynamicInvoke(new object[] { this, args });
if (args.Handled == true)
{
break;
}
}
if (args.Handled == false)
{
connection.CommandBuffer.Enqueue(s);
}
start = index;
index++;
}
if (start > 0)
{
connection.ReceiveBuffer.Reset(0, start + 1);
}
}
}
//
// SEND
//
/// <summary>
/// Overloaded. Sends a string over the telnet socket.
/// </summary>
/// <param name="connection">Connection to send data on.</param>
/// <param name="s">Data to send.</param>
/// <returns>true if the data was sent successfully.</returns>
public bool Send(Connection connection, string s)
{
if (String.IsNullOrEmpty(s) == false)
{
return Send(connection, Encoding.Default.GetBytes(s));
}
return false;
}
/// <summary>
/// Overloaded. Sends an array of data to the client.
/// </summary>
/// <param name="connection">Connection to send data on.</param>
/// <param name="data">Data to send.</param>
/// <returns>true if the data was sent successfully.</returns>
public bool Send(Connection connection, byte[] data)
{
return Send(connection, data, 0, data.Length);
}
public bool Send(Connection connection, char c)
{
return Send(connection, new byte[] { (byte)c }, 0, 1);
}
/// <summary>
/// Sends an array of data to the client.
/// </summary>
/// <param name="connection">Connection to send data on.</param>
/// <param name="data">Data to send.</param>
/// <param name="offset">Starting offset of date in the buffer.</param>
/// <param name="length">Amount of data in bytes to send.</param>
/// <returns></returns>
public bool Send(Connection connection, byte[] data, int offset, int length)
{
bool status = true;
if (connection.Socket == null || connection.Socket.Connected == false)
{
return false;
}
SocketAsyncEventArgs args = m_EventArgsPool.Pop();
args.UserToken = connection;
args.Completed += Send_Completed;
args.SetBuffer(data, offset, length);
try
{
if (connection.Socket.SendAsync(args) == false)
{
Send_Completed(this, args);
}
}
catch (ObjectDisposedException)
{
//
// return the SocketAsyncEventArgs back to the pool and return as the
// socket has been shutdown and disposed of
//
m_EventArgsPool.Push(args);
status = false;
}
return status;
}
/// <summary>
/// Sends a command telling the client that the server WILL echo data.
/// </summary>
/// <param name="connection">Connection to disable echo on.</param>
public void DisableEcho(Connection connection)
{
byte[] b = new byte[] { 255, 251, 1 };
Send(connection, b);
}
/// <summary>
/// Completion callback for SendAsync.
/// </summary>
/// <param name="sender">object which initiated the SendAsync</param>
/// <param name="e">Information about the SendAsync call.</param>
private void Send_Completed(object sender, SocketAsyncEventArgs e)
{
e.Completed -= Send_Completed;
m_EventArgsPool.Push(e);
}
/// <summary>
/// Handles a Telnet command.
/// </summary>
/// <param name="e">Information about the data received.</param>
private void HandleCommand(SocketAsyncEventArgs e)
{
Connection c = e.UserToken as Connection;
if (c == null || e.BytesTransferred < 3)
{
return;
}
for (int i = 0; i < e.BytesTransferred; i += 3)
{
if (e.BytesTransferred - i < 3)
{
break;
}
if (e.Buffer[i] == (int)TelnetCommand.IAC)
{
TelnetCommand command = (TelnetCommand)e.Buffer[i + 1];
TelnetOption option = (TelnetOption)e.Buffer[i + 2];
switch (command)
{
case TelnetCommand.DO:
if (option == TelnetOption.Echo)
{
// ECHO
}
break;
case TelnetCommand.WILL:
if (option == TelnetOption.Echo)
{
// ECHO
}
break;
}
c.ReceiveBuffer.Remove(i, 3);
}
}
}
/// <summary>
/// Echoes data back to the client.
/// </summary>
/// <param name="e">Information about the received data to be echoed.</param>
private void Echo(SocketAsyncEventArgs e)
{
Connection connection = e.UserToken as Connection;
if (connection == null)
{
return;
}
//
// backspacing would cause the cursor to proceed beyond the beginning of the input line
// so prevent this
//
string bs = connection.ReceiveBuffer.ToString();
if (bs.CountAfterBackspace() < 0)
{
return;
}
//
// find the starting offset (first non-backspace character)
//
int i = 0;
for (i = 0; i < connection.ReceiveBuffer.Count; i++)
{
if (connection.ReceiveBuffer[i] != '\b')
{
break;
}
}
string s = Encoding.Default.GetString(e.Buffer, Math.Max(e.Offset, i), e.BytesTransferred);
if (connection.Secure)
{
s = s.ReplaceNot("\r\n\b".ToCharArray(), '*');
}
s = s.Replace("\b", "\b \b");
Send(connection, s);
}
//
// DISCONNECT
//
/// <summary>
/// Disconnects a socket.
/// </summary>
/// <remarks>
/// It is expected that this disconnect is always posted by a failed receive call. Calling the public
/// version of this method will cause the next posted receive to fail and this will cleanup properly.
/// It is not advised to call this method directly.
/// </remarks>
/// <param name="e">Information about the socket to be disconnected.</param>
private void Disconnect(SocketAsyncEventArgs e)
{
Connection connection = e.UserToken as Connection;
if (connection == null)
{
throw (new ArgumentNullException("e.UserToken"));
}
try
{
connection.Socket.Shutdown(SocketShutdown.Both);
}
catch
{
}
connection.Socket.Close();
if (Disconnected != null)
{
Disconnected(this, e);
}
e.Completed -= Receive_Completed;
m_EventArgsPool.Push(e);
}
/// <summary>
/// Marks a specific connection for graceful shutdown. The next receive or send to be posted
/// will fail and close the connection.
/// </summary>
/// <param name="connection"></param>
public void Disconnect(Connection connection)
{
try
{
connection.Socket.Shutdown(SocketShutdown.Both);
}
catch (Exception)
{
}
}
/// <summary>
/// Telnet command codes.
/// </summary>
internal enum TelnetCommand
{
SE = 240,
NOP = 241,
DM = 242,
BRK = 243,
IP = 244,
AO = 245,
AYT = 246,
EC = 247,
EL = 248,
GA = 249,
SB = 250,
WILL = 251,
WONT = 252,
DO = 253,
DONT = 254,
IAC = 255
}
/// <summary>
/// Telnet command options.
/// </summary>
internal enum TelnetOption
{
Echo = 1,
SuppressGoAhead = 3,
Status = 5,
TimingMark = 6,
TerminalType = 24,
WindowSize = 31,
TerminalSpeed = 32,
RemoteFlowControl = 33,
LineMode = 34,
EnvironmentVariables = 36
}
}
曾经有一篇由Coversant公司的Chris Mullins撰写的关于使用.NET实现可扩展TCP/IP的非常好的讨论。不幸的是，他的博客已经从之前的位置消失了，因此我将试图从记忆中拼凑出他的建议（他的一些有用评论出现在这个线程中：C++ vs. C#: Developing a highly scalable IOCP server）。
首先要注意的是，使用Socket类上的Begin/End和Async方法都利用I/O完成端口（IOCP）来提供可伸缩性。这比你选择哪种方法来实施你的解决方案更重要（当正确使用时；请参见下文）。
Chris Mullins的文章基于使用Begin/End，这是我个人有经验的方法。请注意，Chris提供了一个基于此的解决方案，可以在具有2 GB内存的32位机器上扩展到10,000个并发客户端连接，并在64位平台上具有足够内存的情况下扩展到超过100,000个。从我的经验来看（尽管远远达不到这种负载），我没有理由怀疑这些指示性数字。
你想使用一个底层使用IOCP机制的方法的原因是，它使用非常低级别的Windows线程池，在你尝试从其读取数据的I/O通道上没有实际数据之前不会唤醒任何线程（请注意，IOCP也可用于文件I/O）。这样做的好处是，Windows不必切换到一个线程，只需发现当前还没有可用数据就退出，这将减少你的服务器必须进行的上下文切换的数量到最低限度。
上下文切换一定会使“每个连接一个线程”的机制崩溃，尽管这是一个可行的解决方案，如果你只处理几十个连接。然而，这种机制绝对不是“可扩展的”。
内存
首先必须理解，在.NET下，如果你的实现太过天真，使用IOCP很容易导致内存问题。每次调用IOCP的BeginReceive方法将会导致你正在读取的缓冲区被“固定住”。关于为什么这是一个问题的很好的解释，请参见：Yun Jin's Weblog：OutOfMemoryException and Pinning。
幸运的是，这个问题可以避免，但它需要做出一些权衡。建议的解决方案是在应用程序启动时（或接近此时）分配一个大的byte[]缓冲区，大小至少为90 KB左右（在.NET 2中，所需大小在以后版本中可能更大）。之所以这样做，是因为大的内存分配自动进入一个不可压缩的内存段（大对象堆），这个内存段实际上是自动固定的。通过在启动时分配一个大缓冲区，你可以确保这个不可移动的内存块在一个相对“低地址”上，这样它就不会挡路并导致碎片化。
然后，你可以使用偏移量将这个大缓冲区分成用于每个需要读取一些数据的连接的单独区域。这里出现了一种权衡; 由于需要预先分配此缓冲区，因此你必须决定每个连接需要多少缓冲区空间，以及你要设置多少个连接的上限（或者，你可以实现一个抽象层来分配额外的固定缓冲区）。
最简单的解决方案是为每个连接分配一个唯一偏移量内的单个字节缓冲区。然后，您可以调用 BeginReceive 读取一个字节，并通过回调函数执行其余读取操作。
处理
当您从所做的 Begin 调用中获得回调时，非常重要的是意识到回调中的代码将在低级 IOCP 线程上执行。绝对必要避免在此回调中进行长时间的操作。在此回调中使用这些线程进行复杂处理将像使用“每个连接一个线程”一样有效地扼杀可伸缩性。
建议的解决方案是仅在回调中排队工作项以处理传入数据，该工作项将在其他线程上执行。避免在回调中进行任何可能阻塞的操作，以便 IOCP 线程能够尽快返回到其池中。在 .NET 4.0 中，我建议最简单的解决方案是生成一个 Task，并将客户端套接字的引用和由 BeginReceive 调用已读取的第一个字节的副本传递给它。然后，该任务负责从套接字读取表示正在处理的请求的所有数据，执行该请求，然后再次进行新的 BeginReceive 调用，以再次将套接字排队为 IOCP。在 .NET 4.0 之前，您可以使用线程池或创建自己的线程工作队列实现。
基本上，我建议使用 Kevin 的示例代码 进行此解决方案，并添加以下警告：
BeginReceive 的缓冲区已经“固定”BeginReceive 的回调仅做一件事，即排队处理传入数据的实际处理任务您说您想要做一个纯TCP服务器，所以我的下一个观点可能就不重要了。然而，如果您发现自己被迫使用WebRequest类及其派生类，请注意那里有一条龙守门：ServicePointManager。这是一个配置类，它的唯一目的是破坏您的性能。确保从人工强制的ServicePoint.ConnectionLimit中释放您的服务器，否则您的应用程序将永远无法扩展（我让您自己发现默认值是多少...）。您还可以重新考虑在HTTP请求中发送Expect100Continue标头的默认策略。
关于核心套接字管理API，发送方面的事情相当容易，但接收方面则要复杂得多。为了实现高吞吐量和可扩展性，必须确保套接字没有流量控制，因为你没有一个用于接收的缓冲区。理想情况下，为了获得高性能，你应该预先发布3-4个缓冲区，并在收到一个缓冲区（在处理返回的缓冲区之前）后立即发布新的缓冲区，这样可以确保套接字始终有地方存放来自网络的数据。不久你就会明白，你可能无法做到这一点。之后，您将使用BeginWrite发送消息，使用BeginRead接收消息。这就是我之前所说的问题，即由于AuthenticateStream类不支持此功能，您将无法发布多个接收缓冲区。 BeginRead操作会在您接收到整个帧之前自动管理所有I/O。否则，它将无法处理消息认证（解密帧并验证帧上的签名）。尽管在我的经验中，AuthenticatedStream类完成的工作非常出色，不应该有任何问题。即使如此，您也可以只使用4-5％的CPU来填满1Gbps网络。AuthenticatedStream类还会对协议特定的帧大小限制进行强制（SSL为16k，Kerberos为12k）。
这应该让您朝正确方向开始了。我不会在这里发布代码，因为MSDN上有非常好的示例。我已经完成了许多类似的项目，并且能够连接约1000个用户而没有任何问题。如果超过这个数量，您需要修改注册表键以允许内核处理更多的套接字句柄。并确保部署在服务器操作系统上，即Windows Server 2003，而不是Windows XP或Windows Vista（即客户端操作系统），这将产生很大的差异。
顺便提醒一下，如果您在服务器上进行数据库操作或文件I/O，请确保您还使用它们的异步版本，否则您很快就会耗尽线程池。对于SQL Server连接，请确保将“Asyncronous Processing=true”添加到连接字符串中。
我在我的一些解决方案中都有这样一个服务器运行。 这里提供了.NET中执行此操作的不同方法的非常详细的解释：使用.NET中的高性能套接字更接近底层。
最近我一直在寻找改进我们代码的方法，并会研究这个：版本3.5中的套接字性能增强，它专门用于“由使用异步网络I / O以实现最高性能的应用程序使用”。
“这些增强的主要特征是避免在高容量异步套接字I / O期间重复分配和同步对象。 Socket类当前为异步套接字I / O实现的Begin / End设计模式需要为每个异步套接字操作分配System.IAsyncResult对象。”
如果您跟随链接，可以继续阅读。我个人将测试他们的示例代码，以便与我已经拥有的进行基准测试。
这里 您可以找到使用新的3.5 SocketAsyncEventArgs的客户端和服务器的工作代码，因此您可以在几分钟内进行测试并查看代码。 这是一个简单的方法，但它是启动更大实现的基础。 此外，这篇 来自近两年前的MSDN Magazine的文章也很有趣。
为什么呢？自 Windows 2000 起，Windows 可以处理应用程序中数百个线程。我已经这样做过了，如果这些线程不需要同步，那么与它们一起工作真的很容易。特别是考虑到你正在进行大量的 I/O 操作（因此你不会受到 CPU 的限制，而且许多线程将被阻塞在磁盘或网络通信上），我不明白这种限制。我绝对不想为每个连接启动一个线程。
public class Server
{
private static readonly TcpListener listener = new TcpListener(IPAddress.Any, 9999);
public Server()
{
listener.Start();
Console.WriteLine("Started.");
while (true)
{
Console.WriteLine("Waiting for connection...");
var client = listener.AcceptTcpClient();
Console.WriteLine("Connected!");
// each connection has its own thread
new Thread(ServeData).Start(client);
}
}
private static void ServeData(object clientSocket)
{
Console.WriteLine("Started thread " + Thread.CurrentThread.ManagedThreadId);
var rnd = new Random();
try
{
var client = (TcpClient) clientSocket;
var stream = client.GetStream();
while (true)
{
if (rnd.NextDouble() < 0.1)
{
var msg = Encoding.ASCII.GetBytes("Status update from thread " + Thread.CurrentThread.ManagedThreadId);
stream.Write(msg, 0, msg.Length);
Console.WriteLine("Status update from thread " + Thread.CurrentThread.ManagedThreadId);
}
// wait until the next update - I made the wait time so small 'cause I was bored :)
Thread.Sleep(new TimeSpan(0, 0, rnd.Next(1, 5)));
}
}
catch (SocketException e)
{
Console.WriteLine("Socket exception in thread {0}: {1}", Thread.CurrentThread.ManagedThreadId, e);
}
}
}
服务器主程序：
namespace ManyThreadsServer
{
internal class Program
{
private static void Main(string[] args)
{
new Server();
}
}
}
客户端类：
public class Client
{
public Client()
{
var client = new TcpClient();
client.Connect(IPAddress.Loopback, 9999);
var msg = new byte[1024];
var stream = client.GetStream();
try
{
while (true)
{
int i;
while ((i = stream.Read(msg, 0, msg.Length)) != 0)
{
var data = Encoding.ASCII.GetString(msg, 0, i);
Console.WriteLine("Received: {0}", data);
}
}
}
catch (SocketException e)
{
Console.WriteLine("Socket exception in thread {0}: {1}", Thread.CurrentThread.ManagedThreadId, e);
}
}
}
using System;
using System.Threading;
namespace ManyThreadsClient
{
internal class Program
{
private static void Main(string[] args)
{
// first argument is the number of threads
for (var i = 0; i < Int32.Parse(args[0]); i++)
new Thread(RunClient).Start();
}
private static void RunClient()
{
new Client();
}
}
}
Microsoft的并发协调运行时是.NET库的一个例子，旨在简化这种编程难度。它看起来很不错，但由于我没有使用过，所以无法评论它的可扩展性。
对于我个人的项目，需要进行异步网络或磁盘I/O操作，我使用了一组.NET并发/I/O工具，这是我在过去一年中构建的，称为Squared.Task。它受到了像imvu.task和twisted这样的库的启发，并且我在存储库中包含了一些做网络I/O的工作示例。我还在我写的一些应用程序中使用它 - 最大的公开发布的应用程序是NDexer（它使用它进行无线程磁盘I/O）。该库基于我使用imvu.task的经验编写，并且有一组相当全面的单元测试，因此我强烈鼓励您尝试它。如果您有任何问题，我很乐意提供帮助。
在我看来，基于我的使用经验，在.NET平台上使用异步/无线程I/O而不是线程是值得尝试的，只要你准备好应对学习曲线。它可以避免由Thread对象成本带来的可伸缩性问题，在许多情况下，通过谨慎使用并发原语，如futures和promises，您可以完全避免使用锁和互斥体。
我使用了Kevin的解决方案，但他说这个方案缺少消息重组的代码。开发人员可以使用以下代码进行消息重组：
private static void ReceiveCallback(IAsyncResult asyncResult )
{
ClientInfo cInfo = (ClientInfo)asyncResult.AsyncState;
cInfo.BytesReceived += cInfo.Soket.EndReceive(asyncResult);
if (cInfo.RcvBuffer == null)
{
// First 2 byte is lenght
if (cInfo.BytesReceived >= 2)
{
//this calculation depends on format which your client use for lenght info
byte[] len = new byte[ 2 ] ;
len[0] = cInfo.LengthBuffer[1];
len[1] = cInfo.LengthBuffer[0];
UInt16 length = BitConverter.ToUInt16( len , 0);
// buffering and nulling is very important
cInfo.RcvBuffer = new byte[length];
cInfo.BytesReceived = 0;
}
}
else
{
if (cInfo.BytesReceived == cInfo.RcvBuffer.Length)
{
//Put your code here, use bytes comes from "cInfo.RcvBuffer"
//Send Response but don't use async send , otherwise your code will not work ( RcvBuffer will be null prematurely and it will ruin your code)
int sendLenghts = cInfo.Soket.Send( sendBack, sendBack.Length, SocketFlags.None);
// buffering and nulling is very important
//Important , set RcvBuffer to null because code will decide to get data or 2 bte lenght according to RcvBuffer's value(null or initialized)
cInfo.RcvBuffer = null;
cInfo.BytesReceived = 0;
}
}
ContinueReading(cInfo);
}
private static void ContinueReading(ClientInfo cInfo)
{
try
{
if (cInfo.RcvBuffer != null)
{
cInfo.Soket.BeginReceive(cInfo.RcvBuffer, cInfo.BytesReceived, cInfo.RcvBuffer.Length - cInfo.BytesReceived, SocketFlags.None, ReceiveCallback, cInfo);
}
else
{
cInfo.Soket.BeginReceive(cInfo.LengthBuffer, cInfo.BytesReceived, cInfo.LengthBuffer.Length - cInfo.BytesReceived, SocketFlags.None, ReceiveCallback, cInfo);
}
}
catch (SocketException se)
{
//Handle exception and Close socket here, use your own code
return;
}
catch (Exception ex)
{
//Handle exception and Close socket here, use your own code
return;
}
}
class ClientInfo
{
private const int BUFSIZE = 1024 ; // Max size of buffer , depends on solution
private const int BUFLENSIZE = 2; // lenght of lenght , depends on solution
public int BytesReceived = 0 ;
public byte[] RcvBuffer { get; set; }
public byte[] LengthBuffer { get; set; }
public Socket Soket { get; set; }
public ClientInfo(Socket clntSock)
{
Soket = clntSock;
RcvBuffer = null;
LengthBuffer = new byte[ BUFLENSIZE ];
}
}
public static void AcceptCallback(IAsyncResult asyncResult)
{
Socket servSock = (Socket)asyncResult.AsyncState;
Socket clntSock = null;
try
{
clntSock = servSock.EndAccept(asyncResult);
ClientInfo cInfo = new ClientInfo(clntSock);
Receive( cInfo );
}
catch (SocketException se)
{
clntSock.Close();
}
}
private static void Receive(ClientInfo cInfo )
{
try
{
if (cInfo.RcvBuffer == null)
{
cInfo.Soket.BeginReceive(cInfo.LengthBuffer, 0, 2, SocketFlags.None, ReceiveCallback, cInfo);
}
else
{
cInfo.Soket.BeginReceive(cInfo.RcvBuffer, 0, cInfo.BytesReceived, SocketFlags.None, ReceiveCallback, cInfo);
}
}
catch (SocketException se)
{
return;
}
catch (Exception ex)
{
return;
}
}