c# - 这是正确的实现吗？

Question

我有一个 Windows 服务需要从数据库中选择作业并进行处理。

在这里，每个作业都是一个扫描过程，大约需要 10 分钟才能完成。

我是任务并行库的新手。我已经通过以下方式实现了示例逻辑:

Queue queue = new Queue();

for (int i = 0; i < 10000; i++)
{
    queue.Enqueue(i);
}

for (int i = 0; i < 100; i++)
{
    Task.Factory.StartNew((Object data ) =>
    {
        var Objdata = (Queue)data;
        Console.WriteLine(Objdata.Dequeue());
        Console.WriteLine(
            "The current thread is " + Thread.CurrentThread.ManagedThreadId);
    }, queue, TaskCreationOptions.LongRunning);
}

Console.ReadLine();

但是，这会创建很多线程。 由于循环重复 100 次，因此它创建了 100 个线程。

创建那么多并行线程是否正确？

有没有办法限制线程数为10(并发级别)？

Answer 1

分配新的线程时要记住的一个重要因素是操作系统必须分配一些逻辑实体才能运行当前线程:

1. 线程内核对象 - 描述线程的对象， 包括线程的上下文、cpu 寄存器等
2. 线程环境 block - 用于异常处理和线程本地 存储
3. 用户模式堆栈 - 1MB 堆栈
4. 内核模式堆栈 - 用于将参数从用户模式传递到内核 模式

除此之外，可能运行的并发线程数量取决于您的机器正在打包的内核数量，以及创建的线程数量大于您机器拥有的内核数量将开始导致上下文切换，从长远来看，这可能会减慢您的工作速度。

所以在漫长的介绍之后，到了好东西。我们实际上想做的是限制运行的线程数量并尽可能地重用它们。

对于这种工作，我会选择 TPL Dataflow它基于 Producer-Consumer 模式。只是可以做什么的一个小例子:

// a BufferBlock is an equivalent of a ConcurrentQueue to buffer your objects
var bufferBlock = new BufferBlock<object>();

// An ActionBlock to process each object and do something with it
var actionBlock = new ActionBlock<object>(obj =>
{
     // Do stuff with the objects from the bufferblock
});

bufferBlock.LinkTo(actionBlock);
bufferBlock.Completion.ContinueWith(t => actionBlock.Complete());

您可以为每个 Block 传递一个 ExecutionDataflowBlockOptions，这可能会限制 Bounded Capacity(BufferBlock 中的对象数量)和 MaxDegreeOfParallelism 告诉 block 您可能需要的最大并发数。

有一个很好的例子here让你开始。