多线程套接字 - 可能的并发访问

Question

我可能是错的，但在我看来，在这段代码中，一个名为 _buffer 的全局变量被多个线程分配给堆上的一个新对象，所以如果一个线程试图在一个在之前的函数中写入函数之后，但同时另一个线程已将此变量 _buffer 分配给堆上的另一个对象，我会得到错误的数据。这是真的发生还是我错了？如果是，我该如何解决？

public class SocketServer
{
    Socket _serverSocket;
    List<Socket> _clientSocket = new List<Socket>();
    byte[] _buffer;

    public SocketServer()
    {
        _serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
    }

    public void Bind(int Port)
    {
        Console.WriteLine("Setting up server...");
        _serverSocket.Bind(new IPEndPoint(IPAddress.Any, Port));
    }

    public void Listen(int BackLog)
    {
        _serverSocket.Listen(BackLog);
    }

    public void Accept()
    {
        _serverSocket.BeginAccept(AcceptCallback, null);
    }

    private void AcceptCallback(IAsyncResult AR)
    {
        Socket socket = _serverSocket.EndAccept(AR);
        _clientSocket.Add(socket);
        Console.WriteLine("Client Connected");
        _buffer = new byte[1024];
        socket.BeginReceive(_buffer, 0, _buffer.Length, SocketFlags.None, ReceiveCallback, socket);
        Accept();
    }

    private void ReceiveCallback(IAsyncResult AR)
    {
        Socket socket = AR.AsyncState as Socket;
        int bufferSize = socket.EndReceive(AR);

        string text = Encoding.ASCII.GetString(_buffer, 0, bufferSize);
        Console.WriteLine("Text Received: {0}", text);

        string response = string.Empty;

        if (text.ToLower() != "get time")
            response = $"\"{text}\" is a Invalid Request";
        else
            response = DateTime.Now.ToLongTimeString();

        byte[] data = Encoding.ASCII.GetBytes(response);
        socket.BeginSend(data, 0, data.Length, SocketFlags.None, SendCallback, socket);

        _buffer = new byte[1024];
        socket.BeginReceive(_buffer, 0, _buffer.Length, SocketFlags.None, ReceiveCallback, socket);
    }

    private void SendCallback(IAsyncResult AR)
    {
        (AR.AsyncState as Socket).EndSend(AR);
    }
}

Answer 1

_buffer 不是线程安全的。我会使用像 ConcurrentBag 这样的并发集合，而不是平面字节数组。这将为您保证线程安全。 如果要将 _buffer 保留为数组，则必须使用适当的锁（例如使用 lock 关键字）以确保多个线程不会同时尝试访问 _buffer。 更多关于 ConcurrentBag：https://msdn.microsoft.com/en-us/library/dd381779(v=vs.110).aspx

Answer 2

当涉及多个线程时，有一个data race，并且其中至少有一个是写者。

Socket 是线程安全的，但 SocketServer 不是。您在使用它之前立即写信给_buffer。这绝对是多线程场景中的数据竞争。您需要围绕每次访问共享状态的锁定机制。

如果您在传递之前立即覆盖 _buffer 字段，则没有意义。如果您需要使用一个缓冲区，请在初始化时分配一次。为了避免改变太多，你可以这样实现它：

class SocketServer
{
    class Transaction
    {
        public readonly byte[] Data;
        public readonly Socket Socket;

        public Transaction(byte[] data, Socket socket)
        {
            Data = data;
            Socket = socket;
        }
    }

    private readonly object _syncObj = new object();
    private readonly List<Transaction> _received = new List<Transaction>();
    //...

    //...
    private void AcceptCallback(IAsyncResult AR)
    {
        //...
        byte[] buffer = new byte[1024];
        socket.BeginReceive(
            buffer, 0, buffer.Length, SocketFlags.None,
            ReceiveCallback, new Transaction(buffer, socket));
        //...
    }
    private void ReceiveCallback(IAsyncResult AR)
    {
        Transaction trans = (Transaction)AR.AsyncState;
        Socket socket = trans.Socket;
        int bufferSize = socket.EndReceive(AR);
        lock (_syncObj) {
            _received.Add(trans);
        }
        //...
        byte[] buffer = new byte[1024];
        socket.BeginReceive(
            buffer, 0, buffer.Length, SocketFlags.None, 
            ReceiveCallback, new Transaction(buffer, socket));
    }
    //...

    // Call this to get all the received data. 
    // This will block ReceiveCallback until it completes.
    public byte[] GetReceivedData()
    {
        int totalSize = 0;
        lock (_syncObj) {
            for (int i = 0; i < _received.Length; i++) {
                totalSize += _received[i].Data.Length;
            }

            byte[] totalData = new byte[totalSize];
            int offset = 0;
            for (int i = 0; i < _received.Length; i++) {
                byte[] blockData = _received[i].Data;
                Buffer.BlockCopy(blockData, 0, totalData, offset, blockData.Length);
                offset += blockData.Length;
            }
            _received.Clear();
            return totalData;
        }
    }
}

或者，您可以创建IList<ArraySegment<byte>> 的线程安全实现并使用适当的overloads，但这超出了此答案的范围。

在不相关的注释中，您的命名约定不一致。您对字段使用下划线驼峰式大小写，参数混合使用大写和帕斯卡大小写，局部变量使用驼峰式大小写。使用您想要的任何约定，但请保持一致。我建议关注general guidelines。