c# - 需要由线程处理的作业队列

Question

我有一些工作(作业)在队列中(所以有几个)，我希望每个作业都由一个线程处理。

我在看 Rx，但这不是我想要的，然后遇到了并行任务库。

由于我的工作将在网络应用程序中完成，我不希望客户端等待每项工作完成，所以我做了以下工作:

    public void FromWebClientRequest(int[] ids);
    {
        // I will get the objects for the ids from a repository using a container (UNITY)


       ThreadPool.QueueUserWorkItem(delegate
                                         {
                                             DoSomeWorkInParallel(ids, container);
                                         });
    }

    private static void DoSomeWorkInParallel(int[] ids, container)
    {
        Parallel.ForEach(ids, id=>
                                        {
                                            Some work will be done here...
                                            var respository = container.Resolve...
                                        });


       // Here all the work will be done.
       container.Resolve<ILogger>().Log("finished all work");
    }

我会在网络请求上调用上面的代码，这样客户端就不必等待了。

这是执行此操作的正确方法吗？

TIA

Answer 1

从 MSDN 文档中我看到 Unitys IContainer Resolve 方法不是线程安全的(或者它没有被编写)。这意味着您需要在线程循环之外执行此操作。编辑:更改为 Task。

public void FromWebClientRequest(int[] ids);
{
   IRepoType repoType = container.Resolve<IRepoType>();
   ILogger logger = container.Resolve<ILogger>();
   // remove LongRunning if your operations are not blocking (Ie. read file or download file  long running queries etc)
   // prefer fairness is here to try to complete first the requests that came first, so client are more likely to be able to be served "first come, first served" in case of high CPU use with lot of requests
   Task.Factory.StartNew(() => DoSomeWorkInParallel(ids, repoType, logger), TaskCreationOptions.LongRunning | TaskCreationOptions.PreferFairness);
}

private static void DoSomeWorkInParallel(int[] ids, IRepoType repository, ILogger logger)
{
    // if there are blocking operations inside this loop you ought to convert it to tasks with LongRunning
    // why this? to force more threads as usually would be used to run the loop, and try to saturate cpu use, which would be doing nothing most of the time
    // beware of doing this if you work on a non clustered database, since you can saturate it and have a bottleneck there, you should try and see how it handles your workload
    Parallel.ForEach(ids, id=>{
                  // Some work will be done here...
                  // use repository
             });
   logger.Log("finished all work");
}

此外，如 fiver 所述，如果您拥有 .Net 4，那么 Tasks 就是您的不二之选。

为什么要执行任务(评论中的问题):

如果您的方法 fromClientRequest 会被疯狂地频繁触发，您将填满线程池，并且整体系统性能可能不如具有细粒度的 .Net 4。这是任务进入游戏的地方。每个任务都不是自己的线程，而是新的 .Net 4 线程池创建了足够的线程来最大限度地提高系统性能，您无需担心会有多少 cpu 和多少线程上下文切换。

ThreadPool 的一些 MSDN 引用:

When all thread pool threads have been assigned to tasks, the thread pool does not immediately begin creating new idle threads. To avoid unnecessarily allocating stack space for threads, it creates new idle threads at intervals. The interval is currently half a second, although it could change in future versions of the .NET Framework.

The thread pool has a default size of 250 worker threads per available processor

Unnecessarily increasing the number of idle threads can also cause performance problems. Stack space must be allocated for each thread. If too many tasks start at the same time, all of them might appear to be slow. Finding the right balance is a performance-tuning issue.

通过使用 Tasks，您可以解决这些问题。

另一个好处是您可以细化要运行的操作类型。如果您的任务确实运行阻塞操作，这一点很重要。在这种情况下，需要同时分配更多线程，因为它们大部分时间都在等待。 ThreadPool 无法自动实现此目的:

Task.Factory.StartNew(() => DoSomeWork(), TaskCreationOptions.LongRunning);

当然，您可以在不求助于 ManualResetEvent 的情况下按需完成它:

var task = Task.Factory.StartNew(() => DoSomeWork());
task.Wait();

除此之外，如果您不希望出现异常或阻塞，则不必更改 Parallel.ForEach，因为它是 .Net 4 任务并行库的一部分，并且(通常)在 .Net 上运行良好并经过优化。 Net 4 池作为任务。

但是，如果您转到任务而不是并行 for，请从调用者任务中删除 LongRunning，因为 Parallel.For 是阻塞操作而启动任务(使用 fiver 循环)不是。但是这样你就失去了有点先到先得的优化，或者你必须在更多的任务(全部通过 ids 产生)上进行，这可能会给出不太正确的行为。另一种选择是在 DoSomeWorkInParallel 结束时等待所有任务。