C# 套接字 BeginAccept 事件停止触发

Question

以下代码创建套接字服务器和客户端

我启动服务器，如果我一个接一个地启动客户端，它工作正常

如果我立即启动三个客户端，则不会触发一个或多个客户端的 BeginAccept 事件

下面的结果是执行下面的代码后的结果

服务器已启动

服务器正在等待连接...

客户端 0.0.0.0:6352 请求连接

客户端 0.0.0.0:6353 请求连接

客户端 127.0.0.1:6351 请求连接

客户端 127.0.0.1:6351 已连接

客户端 127.0.0.1:6352 已连接

客户端 127.0.0.1:6353 已连接

ServerOnClientConnection 客户端：127.0.0.1:6351

服务器正在等待连接...

ServerOnClientConnection 客户端：127.0.0.1:6353

代码如下

    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Net;
    using System.Net.Sockets;
    using System.Text;
    using System.Threading;
    using System.Windows.Forms;

    namespace Test {
public class TestSockets {

    #region server
    Socket serverSocket;
    bool serverIsAlive;
    public ManualResetEvent waitForConnection = new ManualResetEvent(false);
    private Encoding encod = Encoding.Unicode;

    public void ServerStartInThread() {
        byte[] bytes = new Byte[1024];
        IPAddress ipAddress = IPAddress.Parse("127.0.0.1");
        IPEndPoint localEndPoint = new IPEndPoint(ipAddress, 5500);
        Socket socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
        serverSocket = socket;
        try {
            socket.Bind(localEndPoint);
            socket.Listen(100);
            Thread pollThread = new Thread(delegate () {
                serverIsAlive = true;    // needs if reopen
                SendMessage("Server Started");
                while (serverIsAlive) {
                    try {
                        SendMessage("Server is waiting for a connection...");
                        socket.BeginAccept(new AsyncCallback(ServerOnClientConnection), socket);
                        waitForConnection.Reset();
                        waitForConnection.WaitOne();
                    }
                    catch (Exception ex) {
                        SendMessage("Server: " + ex.ToString());
                    }
                }
                SendMessage("Server Stopped");
                socket.Close();
            }) {
                Name = "SocketServer"
            };
            pollThread.Start();
        }
        catch (Exception ex) {
            SendMessage("Server: " + ex.ToString());
        }
    }

    public void ServerOnClientConnection(IAsyncResult ar) {
        try {
            Socket listener = (Socket)ar.AsyncState;
            Socket clientSocket = listener.EndAccept(ar);
            SendMessage("ServerOnClientConnection Client: " + clientSocket.RemoteEndPoint.ToString());
            StateObject state = new StateObject() {
                socket = clientSocket
            };
            clientSocket.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ServerReceiveCallback), state);
            waitForConnection.Set();
        }
        catch (Exception ex) {
            SendMessage("ServerOnClientConnection: " + ex.ToString());
        }
    }

    public void ServerReceiveCallback(IAsyncResult ar) {
        StateObject state = (StateObject)ar.AsyncState;
        Socket socket = state.socket;
        try {
            if (socket == null) return;
            if (!socket.Connected) {
                return;
            }
            int bytesRead = socket.EndReceive(ar);
            if (bytesRead > 0) {
                state.sb.Append(encod.GetString(state.buffer, 0, bytesRead));
                socket.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ServerReceiveCallback), state);
            }
        }
        catch (Exception ex) {
            SendMessage("ServerReceiveCallback: " + ex.ToString());
        }
    }
    #endregion

    #region client
    private Socket client;
    private bool isAlive = false;
    private ManualResetEvent connectDone = new ManualResetEvent(false);

    public void StartInThread() {
        try {
            IPAddress ipAddress = IPAddress.Parse("127.0.0.1");
            IPEndPoint remoteEP = new IPEndPoint(ipAddress, 5500);
            Thread pollThread = new Thread(delegate () {
                isAlive = true;
                while (isAlive) {
                    try {
                        if (client != null && client.Connected) {
                            continue;
                        }
                        client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
                        client.BeginConnect(remoteEP, new AsyncCallback(ClientConnectCallback), client);
                        SendMessage(string.Format("Client {0} requests connection", client.LocalEndPoint.ToString()));
                        connectDone.Reset();
                        connectDone.WaitOne(3000, false);
                        if (client.Connected) {
                            StateObject state = new StateObject() {
                                socket = client
                            };
                            SendMessage(string.Format("Client {0} connected", client.LocalEndPoint.ToString()));
                            client.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ClientReceiveCallback), state);
                        }
                    }
                    catch (Exception ex) {
                        SendMessage("ClientStartInThread1: " + ex.ToString());
                    }
                }
                SendMessage("Client Disconnected");
            }) {
                Name = "ClientThread"
            };
            pollThread.Start();
        }
        catch (Exception ex) {
            SendMessage("ClientStartInThread2: " + ex.ToString());
        }
    }

    private void ClientConnectCallback(IAsyncResult ar) {
        try {
            Socket socket = (Socket)ar.AsyncState;
            socket.EndConnect(ar);
            connectDone.Set();
        }
        catch (Exception ex) {
            SendMessage("ClientConnectCallback: " + ex.ToString());
        }
    }

    private void ClientReceiveCallback(IAsyncResult ar) {
        StateObject state = (StateObject)ar.AsyncState;
        Socket socket = state.socket;
        if (socket == null || !socket.Connected) return;
        try {
            int bytesRead = socket.EndReceive(ar);
            if (bytesRead > 0) {
                state.sb.Append(encod.GetString(state.buffer, 0, bytesRead));
                socket.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ClientReceiveCallback), state);
            }
            else {
                socket.Close();
            }
        }
        catch (Exception ex) {
            SendMessage("ClientReceiveCallback: " + ex.ToString());
            socket.Close();
        }
    }

    #endregion

    private void SendMessage(string v) {
        System.Diagnostics.Debug.WriteLine(v);
    }

    public static void Start() {
        TestSockets server = new TestSockets();
        server.ServerStartInThread();
        TestSockets c1 = new TestSockets();
        c1.StartInThread();
        TestSockets c2 = new TestSockets();
        c2.StartInThread();
        TestSockets c3 = new TestSockets();
        c3.StartInThread();

    }
}
public class StateObject {
    public Socket socket = null;
    public const int BufferSize = 1024;
    public byte[] buffer = new byte[BufferSize];
    public StringBuilder sb = new StringBuilder();
}

}

Answer 1

您的代码中有一个竞争条件，导致 waitForConnection 事件句柄被设置两次，第二次调用 Set() 无效，在主服务器线程有机会通过 @987654324 之前@ 并重置句柄。这会导致它错过第二组。

解决它的一种方法是切换到信号量对象。例如：

#region server
Socket serverSocket;
bool serverIsAlive;
SemaphoreSlim waitForConnection = new SemaphoreSlim(0);
private Encoding encod = Encoding.Unicode;

public void ServerStartInThread()
{
    byte[] bytes = new Byte[1024];
    IPAddress ipAddress = IPAddress.Parse("127.0.0.1");
    IPEndPoint localEndPoint = new IPEndPoint(ipAddress, 5500);
    Socket socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
    serverSocket = socket;
    try
    {
        socket.Bind(localEndPoint);
        socket.Listen(100);
        Thread pollThread = new Thread(delegate () {
            serverIsAlive = true;    // needs if reopen
            SendMessage("Server Started");
            while (serverIsAlive)
            {
                try
                {
                    SendMessage("Server is waiting for a connection...");
                    socket.BeginAccept(new AsyncCallback(ServerOnClientConnection), socket);
                    waitForConnection.Wait();
                }
                catch (Exception ex)
                {
                    SendMessage("Server: " + ex.ToString());
                }
            }
            SendMessage("Server Stopped");
            socket.Close();
        })
        {
            Name = "SocketServer"
        };
        pollThread.Start();
    }
    catch (Exception ex)
    {
        SendMessage("Server: " + ex.ToString());
    }
}

public void ServerOnClientConnection(IAsyncResult ar)
{
    try
    {
        Socket listener = (Socket)ar.AsyncState;
        Socket clientSocket = listener.EndAccept(ar);
        SendMessage("ServerOnClientConnection Client: " + clientSocket.RemoteEndPoint.ToString());
        StateObject state = new StateObject()
        {
            socket = clientSocket
        };
        clientSocket.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0, new AsyncCallback(ServerReceiveCallback), state);
        waitForConnection.Release();
    }
    catch (Exception ex)
    {
        SendMessage("ServerOnClientConnection: " + ex.ToString());
    }
}

这将允许ServerOnClientConnection() 方法增加信号量计数，以便主服务器线程可以继续循环，直到观察到所有接受的连接。

但是，坦率地说，使用这些事件句柄的 MSDN 示例（这是我假设您的代码直接或间接基于的）恕我直言，它被简单地破坏了。它们在实际上不需要线程同步的情况下引入了复杂性，从而使代码变得不必要地复杂化，并使代码更难正常工作。

更惯用的方法是根本没有额外的线程，在 ServerStartInThread() 方法中调用一次 BeginAccept()（您可能将其重命名为 ServerStart()）并在 @ 中调用 BeginAccept()处理当前接受的客户端后的 987654331@ 方法。 IE。就像你现在处理BeginReceive() 一样。

也就是说，恕我直言，即使是惯用的方法也已过时。最初设计 API 时它运行良好，但从那时起 .NET 已经取得了长足的进步，并且具有更好的机制来处理此类异步操作。我在this answer 中发布了一个简单聊天服务器的示例，它展示了如何使用async/await 以更简单、更易于阅读的方式实现此类事情。您可能想了解如何将相同的技术应用于您自己的代码。