不在本地主机上时的 TcpClient/TcpListener 通信问题（多线程）

Question

我正在做一个项目，我通过 TCP/IP 通信序列化命令。当我在本地主机上时它可以工作，但是当我在不同的服务器上运行我的侦听器时，它在尝试反序列化侦听器端的命令时随机失败。 抛出的异常是：“尝试反序列化一个空流。”和“在解析完成之前遇到流结束。”来自序列化程序

当我单独运行一系列命令时，它可以正常工作，但是当我创建线程并同时运行多个序列时，它会失败。

监听器在 4 个不同的端口上创建监听器，客户端为每个端口运行 1 个线程。当其中一个线程到达序列的末尾时，它就会终止。

我尝试让我的客户单例，也尝试了互斥锁。但还是同样的问题。

这是我的客户：

public class TcpIpCommunicator : ICommunicator, IDisposable
{
    private Dictionary<int,TcpClient> clientSockets = new Dictionary<int,TcpClient>();
    public IInverterCommand ReadAsyncCommand { set; get; }
    private static TcpIpCommunicator tcpIpCommunicator;

    private TcpIpCommunicator()
    {
    }

    public static TcpIpCommunicator GetInstance()
    {
        if(tcpIpCommunicator == null)
            tcpIpCommunicator = new TcpIpCommunicator();

        return tcpIpCommunicator;
    }

    public void Send(IInverterCommand command, int id)
    {
        var serializer = new Serializer();
        MemoryStream stream = serializer.SerializeMultipleObjects(command);
        var _bytes = stream.GetBuffer();

        var networkStream = clientSockets[id].GetStream();
        networkStream.Write(_bytes, 0, _bytes.Length);
        networkStream.Flush();
    }

    public IInverterCommand Read(int id)
    {
        var memoryStream = new MemoryStream();

        byte[] buffer;
        var networkStream = clientSockets[id].GetStream();
        do
        {
            buffer = new byte[clientSockets[id].ReceiveBufferSize];
            int sizeRead = networkStream.Read(buffer, 0, buffer.Length);
            memoryStream.Write(buffer, 0, sizeRead);
        } while (networkStream.DataAvailable);

        networkStream.Flush();

        memoryStream.Position = 0;
        var serializer = new Serializer();
        return serializer.DeSerializeMultipleObject(memoryStream);

    }

    public void ReadAsync(object id)
    {
        ReadAsyncCommand = Read((int)id);
    }

    public void Dispose()
    {
        foreach (var tcpClient in clientSockets.Values)
        {
            tcpClient.Close();
        }
    }

    public int Connect(string ip, int port)
    {
        var tcpClient = new TcpClient();
        tcpClient.ReceiveTimeout = int.MaxValue;
        tcpClient.SendTimeout = int.MaxValue;
        tcpClient.Connect(ip, port);
        int key = findKey();
        clientSockets.Add(key, tcpClient);

        return key;
    }

    public void DestroyConnection(int id)
    {
        clientSockets[id].Close();
        clientSockets.Remove(id);
    }

    private int findKey()
    {
        int key = 0;
        while(clientSockets.ContainsKey(key))
        {
            key++;
        }

        return key;
    }
}

我的服务器端代码在这里：

public class TCPListener : IDisposable
{
    private readonly TcpListener _serverSocket;
    private NetworkStream _networkStream;
    private readonly TcpClient _clientSocket = default(TcpClient);

    public TCPListener(int port)
    {
        _serverSocket = new TcpListener(port);
        _serverSocket.Server.ReceiveTimeout = int.MaxValue;
        _serverSocket.Server.SendTimeout = int.MaxValue;
        _serverSocket.Start();
        _clientSocket = _serverSocket.AcceptTcpClient();
    }

    public void Send(IInverterCommand message)
    {
        _networkStream = _clientSocket.GetStream();
        var serialize = new Serializer();
        var stream = serialize.SerializeMultipleObjects(message);
        var _bytes = stream.GetBuffer();

        if (_bytes.Length > _clientSocket.ReceiveBufferSize)
        {
            byte[] bytes = new byte[_clientSocket.ReceiveBufferSize];
            for (int i = 0; i < _bytes.Length; i += _clientSocket.ReceiveBufferSize)
            {
                for (int j = 0; j < _clientSocket.ReceiveBufferSize && i + j != _bytes.Length; ++j)
                {
                    bytes[j] = _bytes[i + j];
                }

                _networkStream.Write(bytes, 0, _clientSocket.ReceiveBufferSize);
            }
        }
        else
        {
            _networkStream.Write(_bytes, 0, _bytes.Length);
        }

        Thread.Sleep(50);

        _networkStream.Flush();
    }

    public IInverterCommand ReadCommand()
    {
        _networkStream = _clientSocket.GetStream();
        var memoryStream = new MemoryStream();

        do
        {
            var buffer = new byte[_clientSocket.ReceiveBufferSize];
            int sizeRead = _networkStream.Read(buffer, 0, buffer.Length);
            memoryStream.Write(buffer, 0, sizeRead);
            Thread.Sleep(50);
        } while (_networkStream.DataAvailable);

        _networkStream.Flush();
        memoryStream.Position = 0;
        var serializer = new Serializer();
        return serializer.DeSerializeMultipleObject(memoryStream);
    }

    public void Dispose()
    {
        _clientSocket.Close();
        _serverSocket.Stop();
    }
}

这里通常是在客户端调用代码：

IInverterCommand command = new SoftwareUpdateInverterCommand();
        tcpClient.Send(command, tcpId);

        var thread = new Thread(tcpClient.ReadAsync);
        thread.Start(tcpId);

        if (!thread.Join(timeout))
        {
            thread.Abort();
            tcpClient.DestroyConnection(tcpId);
            return;
        }

以及服务器端调用代码：

//Recieving CMD on software update
            TcpListener = new TCPListener((int)port);
            var command = TcpListener.ReadCommand();

            //Sending OK back to server
            command.Message = "OK";
            TcpListener.Send(command);

Answer 1

我建议您阅读有关线程同步的主题以及它的含义以及如何使用它。当您在非线程安全的线程之间共享内容时，也会发生什么情况。

您的代码中根本没有线程同步，但是您在没有任何同步的情况下从至少 2 个不同的线程读取和写入字典（不安全）。

仅此一项就可能导致难以追踪的错误以及许多您不愿处理的不同现象。

您真的需要阅读有关如何正确执行线程的信息。