我的应用程序中的线程安全建议？

Question

我的 TcpServer 中有以下方法

    private void Receive(ZClient client)
    {
        NetworkStream ns = client.TcpClient.GetStream();
        int bytesRead = 0;
        byte[] message = new byte[client.TcpClient.ReceiveBufferSize];
        while (true)
        {
            bytesRead = ns.Read(message, 0, client.TcpClient.ReceiveBufferSize);
            if (bytesRead == 0)
            {
                break;  // Exit Thread
            }
            else
            {
                Array.Resize<Byte>(ref client.Buffer, client.Buffer.Length + bytesRead);
                Buffer.BlockCopy(message, 0, client.Buffer, client.Buffer.Length - bytesRead, bytesRead);
                // Issue Below
                DisassembleRawMessage(client); // *1 See below
                new Thread(()=>DeQueueMessages(client)).Start();  // *2 See below
                // End Issue
            }
        }
    }

1. DissassembleRawMessage 负责处理我的缓冲区并在我的客户端消息队列中加入单个消息。这里没问题。

2. DeQueueMessages 负责处理之前在 DissassembleRawMessage 方法中放入我的客户端消息队列的对象。

这个问题可能看起来很明显。收到消息后，我启动了一个新线程。在 dequeing 和处理单个消息的过程中，队列可能添加了更多的项目，因为在中间可能已经收到了更多的数据。这导致枚举错误，提示我在枚举过程中不能修改集合的内容。

我不想通过在同一个线程中处理队列来阻止我的接收方法，因为它会破坏允许队列建立请求的目的，同时仍然接收数据。如何继续对对象进行排队，同时处理队列中的对象而不会遇到枚举修改异常？

更新：使消息出队的方法。

while (client.Queue.Count() > 0)
{
   byte[] buffer = client.Queue.Dequeue();
   ...
}

Answer 1

首先，我不会为收到的每条消息都使用新线程。要么启动一个等待新工作而不是退出的线程，要么使用 ThreadPool。

如果您在枚举期间收到 InvalidOperationException，这意味着您正在迭代列表或其他 IEnumerable。这是否意味着你正在做这样的事情？

void DeQueueMessages(ZClient client)
{
    foreach(Message m in client.messageList)
    {
        ...
    }
    messageList.Clear();
}

如果是这样，切换到使用实际的 Queue 类，并执行此 while 循环：

void DeQueueMessages(ZClient client)
{
    while(client.messageQueue.Count > 0)
    {
        Message m = client.messageQueue.Dequeue();
        ...
    }
}

如果有记忆，标准的 System.Collections.Generic.Queue 类不是线程安全的，所以我会选择 System.Collections.Concurrent.ConcurrentQueue。

Answer 2

异步 ​​I/O 通常是网络 I/O 场景中对额外线程的感知需求的良好解决方案。

如果您不想阻塞读取，请通过 NetworkStream.BeginRead 使用异步 I/O - 不需要额外的线程，除非您希望回调将接收到的数据的长期工作移交给线程池以执行.您可以使用 .Net 4、ConcurrentQueue 或 BlockingCollection 中的 concurrent collections 之一来执行此操作。

如果 NetworkStream 上没有可用数据，则此代码可能会导致紧密循环。示例 here 建议您必须在像这样的循环中调用 Read 之前检查 DataAvailable。如果没有那个检查，这段代码似乎有问题。

Answer 3

将消息复制到当前正在处理的第二个队列并清除传入队列。在复制和清除操作期间，您可能至少需要阻止。

或者，将单个消息处理分派给工作线程，因为它们是出队的。

Answer 4

您可以考虑使用ThreadPool.QueueUserWorkItem方法将要在线程池上处理的消息排队，而不是使用自定义队列。

优点：

• 轻松排队，
• 通过更好的工作线程管理降低开销，
• 跨整个应用程序的线程处理的单一机制。

缺点：

• 无法控制排队机制，
• 无法控制线程分配策略。