哪种线程方法最适合主服务器网络？

Question

根据这个问题，我决定从更具体的方法开始，而不是抛出我无法整合的理论和概念信息：Are Socket.*Async methods threaded?

关键是要为所有客户端保持流动性，同时优化服务器。这意味着以某种方式异步，以免阻塞主要操作。

这是我想出的一些方法。 “处理”是一种假设的方法，用于处理从客户端接收到的数据。考虑到这通常需要 1-5 毫秒，对于罕见数据库调用可能需要 500-2000 毫秒。

使用 Socket.*异步和循环

static void Main()
{
    Socket listener = new Socket(...);
    listener.Bind(new IPEndPoint(IPAddress.Any, 555));

    List<Socket> clients = new List<Socket>();

    SocketAsyncEventArgs e = new SocketAsyncEventArgs();

    while (true)
    {
        if (listener.AcceptAsync(e))
        {
            clients.Add(e.AcceptSocket);
        }

        foreach (Socket client in clients)
        {
            if (client.ReceiveAsync(e))
            {
                Process(e.Buffer);
            }
        }
    }
}

优点：

• 只有一个线程！
• 管理起来相当简单。

缺点：

• While(true)：CPU 太密集？
• 由于所有操作都相互跟踪，它们会减慢所有连接的客户端。

我认为这是一个好的开始，也许是我最好的解决方案。如果我可以混合在一个线程池中，将接受、接收和进程拆分到不同的线程中，我们可能会有所进展。

使用 Socket.Begin*/End* 和 ManualResetEvent

static class Server
{
    static Socket listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
    static ManualResetEvent acceptDone = new ManualResetEvent(false);

    static void Main()
    {
        listener.Bind(new IPEndPoint(IPAddress.Any, 555));

        while (true)
        {
            acceptDone.Reset();

            listener.BeginAccept(OnAccept, null);

            acceptDone.WaitOne();
        }
    }

    private static void OnAccept(IAsyncResult ar)
    {
        acceptDone.Set();

        new Receiver(listener.EndAccept(ar));
    }
}

class Receiver
{
    Socket socket;
    byte[] buffer = new byte[1024];
    static ManualResetEvent receiveDone = new ManualResetEvent(false);

    public Receiver(Socket socket)
    {
        this.socket = socket;

        new Thread
        (
            delegate()
            {
                while (true)
                {
                    receiveDone.Reset();

                    socket.BeginReceive(buffer, 0, 1024, SocketFlags.None, OnReceive, null);

                    receiveDone.WaitOne();
                }
            }
        ).Start();
    }

    private void OnReceive(IAsyncResult ar)
    {
        receiveDone.Set();

        int received = socket.EndReceive(ar);
        byte[] toProcess = new byte[received];
        Buffer.BlockCopy(buffer, 0, toProcess, 0, received);
        Process(toProcess);
    }
}

优点：

• 完全异步，客户端不会因其他操作而减慢速度。
• 使用 ManualResetEvents 可以很好地停止服务器。

缺点：

• 线程太多，每个客户端 1 个！
• 因为所有线程都被重置事件阻塞，所以浪费了处理时间？

最后是一种简化此解决方案的方法，无需手动重置事件。

使用阻塞调用和手动线程

static class Server
{
    static Socket listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);

    static void Main()
    {
        listener.Bind(new IPEndPoint(IPAddress.Any, 555));

        while (true)
        {
            new Receiver(listener.Accept());
        }
    }
}

class Receiver
{
    Socket socket;

    public Receiver(Socket socket)
    {
        this.socket = socket;

        new Thread
        (
            delegate()
            {
                while (true)
                {
                    byte[] buffer = new byte[1024];
                    int received = socket.Receive(buffer);
                    byte[] toProcess = new byte[received];
                    Buffer.BlockCopy(buffer, 0, toProcess, 0, received);
                    Process(toProcess);
                }
            }
        ).Start();
    }
}

使用线程池

我真的不知道如何使用它，有人可以给我一个例子吗？

建议

可能解决方案不是本文中的任何解决方案。你会怎么处理呢？

如您所见，我使用了 .*Async 方法、Begin*/End* 方法和阻塞方法，但都有相对较大的缺点。

在此先感谢 :) 我迫不及待地想看看 S/O 的代码示例。

Answer 1

您没有正确使用Begin/End。无需等待事件，让框架来处理。请注意，在the MSDN example 中，接收循环不使用显式事件，尽管接受循环是为了便于控制和展示。

static class Server
{
    static Socket listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);

    static void Main()
    {
        listener.Bind(new IPEndPoint(IPAddress.Any, 555));
        listener.BeginAccept(OnAccept, null);

        WaitUntilServerNeedsToShutdown () ;

        // worker threads will die because they are background
    }

    private static void OnAccept(IAsyncResult ar)
    {
        new Receiver(listener.EndAccept(ar));
        listener.BeginAccept(OnAccept, null);
    }
}

class Receiver
{
    Socket socket;
    byte[] buffer = new byte[1024];

    public Receiver(Socket socket)
    {
        this.socket = socket;
        socket.BeginReceive(buffer, 0, 1024, SocketFlags.None, OnReceive, null);
    }

    private void OnReceive(IAsyncResult ar)
    {
        int received = socket.EndReceive(ar);
        byte[] toProcess = new byte[received];
        Buffer.BlockCopy(buffer, 0, toProcess, 0, received);

        // TODO: detect EOF and error conditions
        socket.BeginReceive(buffer, 0, 1024, SocketFlags.None, OnReceive, null);

        // TODO: is it OK to process incomplete data?
        Process(toProcess);
    }
}