0

我正在编写一个服务器应用程序，它将接收来自多个 TCP 连接的数据。我们希望能够扩展到约 200 个连接。我为此编写的第一个算法如下：

while (keepListening)
{
    foreach (TcpClient client in clientList)
    {
        if (!client.Connected)
        {
            client.Close();
            deleteList.Add(client);
            continue;
        }

        int dataAvail = client.Available;
        if (dataAvail > 0)
        {
            NetworkStream netstr = client.GetStream();
            byte[] arry = new byte[dataAvail];
            netstr.Read(arry, 0, dataAvail);
            MemoryStream ms = new MemoryStream(arry);
            try
            {
                CommData data = dataDeserializer.Deserialize(ms) as CommData;
                beaconTable.BeaconReceived(data);
            }
            catch
            { }
        }
    }

    foreach (TcpClient clientToDelete in deleteList)
        clientList.Remove(clientToDelete);
    deleteList.Clear();

    while (connectionListener.Pending())
        clientList.Add(connectionListener.AcceptTcpClient());

    Thread.Sleep(20);
}

这工作得很好，虽然我发现我必须添加 Thread.Sleep 来减慢循环，否则它会占用整个核心，无论有多少或几个连接。我被告知 Thread.Sleep 通常被认为是不好的，所以我寻找了一些替代方案。在与此类似的问题中，建议我使用 WaitHandles 来使用 BeginRead 和 BeginAccept，因此我编写了一个算法来使用它来做同样的事情，并想出了这个：

while (keepListening)
{
    int waitResult = WaitHandle.WaitAny(waitList.Select(t => t.AsyncHandle.AsyncWaitHandle).ToArray(), connectionTimeout);

    if (waitResult == WaitHandle.WaitTimeout)
        continue;

    WaitObject waitObject = waitList[waitResult];
    Type waitType = waitObject.WaitingObject.GetType();

    if (waitType == typeof(TcpListener))
    {
        TcpClient newClient = (waitObject.WaitingObject as TcpListener).EndAcceptTcpClient(waitObject.AsyncHandle);
        waitList.Remove(waitObject);

        byte[] newBuffer = new byte[bufferSize];
        waitList.Add(new WaitObject(newClient.GetStream().BeginRead(newBuffer, 0, bufferSize, null, null), newClient, newBuffer));

        if (waitList.Count < 64)
            waitList.Add(new WaitObject(connectionListener.BeginAcceptTcpClient(null, null), connectionListener, null));
        else
        {
            connectionListener.Stop();
            listening = false;
        }
    }
    else if (waitType == typeof(TcpClient))
    {
        TcpClient currentClient = waitObject.WaitingObject as TcpClient;
        int bytesRead = currentClient.GetStream().EndRead(waitObject.AsyncHandle);

        if (bytesRead > 0)
        {
            MemoryStream ms = new MemoryStream(waitObject.DataBuffer, 0, bytesRead);
            try
            {
                CommData data = dataDeserializer.Deserialize(ms) as CommData;
                beaconTable.BeaconReceived(data);
            }
            catch
            { }
            byte[] newBuffer = new byte[bufferSize];
            waitList.Add(new WaitObject(currentClient.GetStream().BeginRead(newBuffer, 0, bufferSize, null, null), currentClient, newBuffer));
        }
        else
        {
            currentClient.Close();
        }

        waitList.Remove(waitObject);

        if (!listening && waitList.Count < 64)
        {
            listening = true;
            connectionListener.Start();
            waitList.Add(new WaitObject(connectionListener.BeginAcceptTcpClient(null, null), connectionListener, null));
        }
    }
    else
        throw new ApplicationException("An unknown type ended up in the wait list somehow: " + waitType.ToString());
}

这也很好，直到我打到 64 个客户。我写了一个限制，不接受超过 64 个客户端，因为这是 WaitAny 将接受的最大 WaitHandles 数量。我看不到任何绕过这个限制的好办法，所以我基本上不能像这样维持超过 64 个连接。Thread.Sleep 算法适用于 100 多个连接。

我也不太喜欢预先分配任意大小的接收数组，而不是在接收到数据后以接收到的数据的确切大小进行分配。而且我无论如何都必须给 WaitAny 一个超时，否则它会阻止运行它的线程在我关闭应用程序时加入（如果没有连接）。而且它通常更长，更复杂。

那么为什么 Thread.Sleep 是更糟糕的解决方案呢？有什么方法至少可以让 WaitAny 版本处理超过 64 个连接？是否有一些我没有看到的完全不同的处理方式？

4

1 回答 1

0

Jim 给出了使用异步回调而不是 WaitHandles 的明显建议。我最初认为这太复杂了，但是当我意识到我可以在状态对象中传递对调用 TcpListener 或 TcpClient 的引用时，它变得简单多了。有了这些以及对线程安全的一些更改，它就可以开始了。它可以通过 100 多个连接进行测试，并且干净地退出没有问题。不过，我仍然想要一个替代方案，不必预先分配数据缓冲区。这是任何尝试类似事情的人的代码：

public class NetworkReceiver : IDisposable
{
    private IReceiver beaconTable;
    private XmlSerializer dataDeserializer;
    private HashSet<TcpClient> ClientTable;
    private TcpListener connectionListener;
    private int bufferSize = 1000;

    public NetworkReceiver(IReceiver inputTable)
    {
        beaconTable = inputTable;
        dataDeserializer = new XmlSerializer(typeof(CommData));

        ClientTable = new HashSet<TcpClient>();
        connectionListener = new TcpListener(IPAddress.Any, SharedData.connectionPort);
        connectionListener.Start();
        connectionListener.BeginAcceptTcpClient(ListenerCallback, connectionListener);
    }

    private void ListenerCallback(IAsyncResult callbackResult)
    {
        TcpListener listener = callbackResult.AsyncState as TcpListener;
        TcpClient client;

        try
        {
            client = listener.EndAcceptTcpClient(callbackResult);

            lock (ClientTable)
                ClientTable.Add(client);

            ClientObject clientObj = new ClientObject() { AsyncClient = client, Buffer = new byte[bufferSize] };
            client.GetStream().BeginRead(clientObj.Buffer, 0, bufferSize, ClientReadCallback, clientObj);

            listener.BeginAcceptTcpClient(ListenerCallback, listener);
        }
        catch (ObjectDisposedException)
        {
            return;
        }
    }

    private void ClientReadCallback(IAsyncResult callbackResult)
    {
        ClientObject clientObj = callbackResult.AsyncState as ClientObject;
        TcpClient client = clientObj.AsyncClient;

        if (!client.Connected)
            return;

        try
        {
            int bytesRead = client.GetStream().EndRead(callbackResult);
            if (bytesRead > 0)
            {
                MemoryStream ms = new MemoryStream(clientObj.Buffer, 0, bytesRead);
                try
                {
                    CommData data;
                    lock (dataDeserializer)
                        data = dataDeserializer.Deserialize(ms) as CommData;
                    lock (beaconTable)
                        beaconTable.BeaconReceived(data);
                }
                catch
                { }

                client.GetStream().BeginRead(clientObj.Buffer, 0, bufferSize, ClientReadCallback, clientObj);
            }
            else
            {
                client.Close();
                lock (ClientTable)
                    ClientTable.Remove(client);
            }
        }
        catch (Exception ex)
        {
            if (ex.GetType() == typeof(ObjectDisposedException) || ex.GetType() == typeof(InvalidOperationException))
                return;
            else
                throw;
        }
    }

    class ClientObject
    {
        public TcpClient AsyncClient;
        public byte[] Buffer;
    }

    public void Dispose()
    {
        connectionListener.Stop();
        foreach (TcpClient client in ClientTable)
            client.Close();
    }
}
于 2013-03-29T22:00:13.830 回答