c# - 即使 ThreadPool.GetAvailableThreads() 表示丰富，任务也会受到调度缓慢的影响吗？

问题描述

我们有一个在生产中运行的服务器，它接收数据，放入queueA. 另一个线程一次从中获取一个项目queueA，对其进行处理并将其放入queueB.

此设置实际上是镜像的，因此两个镜像服务器接收完全相同的数据并以相同的方式处理它们，在主动-主动冗余设置中。

大约一年一次，其中一台服务器的消息大量堆积在queueA.

所以问题似乎是queueA从queueB. 除了（完全没有必要）使用 Task 库（如下面的精简版和简化版所示）之外，没有太多事情要做。

public IAsyncResult BeginReceive()
{
  Task<object> task = new Task<object>(_ =>
  {
    object message;
    if (!queueA.TryDequeue(out message))
    {
      if (queueA.IsEmpty)
        waitQueueA.WaitOne(10); // waitQueueA is properly signaled whenever an item is put into queueA
    }
    return message;
  }, null, TaskCreationOptions.PreferFairness);

  task.ContinueWith((t) =>
  {
    object receivedMessage = t.Result;
    if (receivedMessage != null)
    {
      lock (bLock) // bLock is only used by this piece of code
      {
        queueB.Enqueue(receivedMessage);
      }
    }
    else
    {
      Thread.Sleep(1);
    }
    BeginReceive(OnReceive, channel);
  });

  task.Start();
  return task;
}

public object EndReceive(IAsyncResult result)
{
  Task<object> task = (Task<object>) result;
  return task.Result;
}

忽略代码的许多特性（就我个人而言，我会为此创建一个单独的专用线程，并在一个大while (true) { }循环中执行上述所有操作，而不Task涉及任何 s），这种情况可能会使其性能如此糟糕，以至于循环旋转以 15 次迭代/秒的性能为主，最高低于 50 次迭代/秒？我们ThreadPool.GetAvailableThreads()每 5 秒记录一次，它表明整个过程中有数千个可用线程。

这段代码大部分时间都运行良好（足够），但是当它失败时，它似乎从程序开始到结束都失败了，大约一个小时左右，当内存耗尽queueA并且它的项目。所以看起来程序可以进入某种无法恢复的时髦状态。

这是一张每秒处理的物品数量的图表，对于好机器和坏机器，横轴是时间轴。（请注意纵轴是对数）

图表显示，例如，“坏”服务器的上限在大多数情况下约为 17/18 项/秒，而“好”服务器能够执行 3700 项/秒，具体取决于新项目的速率收到并进入queueA。

由于我对任务库的复杂性不太熟悉，我想知道在某些随机情况下PreferFairness，异步（因此不是真的）“递归”调用的组合BeginReceive是否会导致这个问题。还有其他想法如何解决这个问题吗？

注意我已经删除了几个try {} catch { error.log(); }结构来简化它。没有记录错误，所以我相信代码不会引发异常。而且queueA它不会单调地增长，它有时会缩小一点，所以这个循环似乎是活的，虽然很慢。

标签： c#.nettask-parallel-library

c# - 即使 ThreadPool.GetAvailableThreads() 表示丰富，任务也会受到调度缓慢的影响吗？

问题描述

解决方案

推荐阅读