.net 什么决定了TaskFactory派生作业的线程数?

6jjcrrmo  于 2023-01-06  发布在  .NET
关注(0)|答案(3)|浏览(125)

下面的代码:

var factory = new TaskFactory();
for (int i = 0; i < 100; i++)
{
    var i1 = i;
    factory.StartNew(() => foo(i1));
}

static void foo(int i)
{
    Thread.Sleep(1000);
    Console.WriteLine($"foo{i} - on thread {Thread.CurrentThread.ManagedThreadId}");
}

我可以看到它一次只做4个线程(基于观察)。我的问题:
1.什么决定了一次使用的线程数?
1.我怎样才能找回这个号码?
1.这个号码怎么改?
pidoss.我的盒子有4个核。
P.P.S.我需要有一个特定数量的任务(不能再多了),这些任务由TPL并发处理,并以下面的代码结束:

private static int count = 0;   // keep track of how many concurrent tasks are running

private static void SemaphoreImplementation()
{
    var s = new Semaphore(20, 20);  // allow 20 tasks at a time

    for (int i = 0; i < 1000; i++)
    {
        var i1 = i;

        Task.Factory.StartNew(() =>
        {
            try
            {                        
                s.WaitOne();
                Interlocked.Increment(ref count);

                foo(i1);
            }
            finally
            {
                s.Release();
                Interlocked.Decrement(ref count);
            }
        }, TaskCreationOptions.LongRunning);
    }
}

static void foo(int i)
{
    Thread.Sleep(100);
    Console.WriteLine($"foo{i:00} - on thread " + 
            $"{Thread.CurrentThread.ManagedThreadId:00}. Executing concurently: {count}");
}
uinbv5nw

uinbv5nw1#

当您在.NET中使用Task时,您是在告诉TPL安排一项工作(通过TaskScheduler)在ThreadPool上执行。请注意,工作将在其最早的时机以及调度程序认为合适的任何方式进行调度。这意味着TaskScheduler将决定使用多少线程来运行n数量的任务,以及在哪个线程上执行哪个任务。
TPL经过了很好的调整,在执行任务时会不断调整算法。因此,在大多数情况下,它会尽量减少争用。这意味着如果您运行100个任务,而只有4个内核(您可以使用Environment.ProcessorCount获得),在任何给定时间执行4个以上线程都没有意义,否则它将需要执行更多的上下文切换。现在,有时您需要显式覆盖此行为。假设您需要等待某种IO完成,这是完全不同的情况
总之,请相信TPL。但是如果您坚持为每个任务生成一个线程(并不总是一个好主意!),您可以用途:

Task.Factory.StartNew(
    () => /* your piece of work */, 
    TaskCreationOptions.LongRunning);

这将告诉 DefaultTaskscheduler为该工作显式生成一个新线程。
你也可以使用你自己的Scheduler并把它传递给TaskFactory,你可以找到一大堆SchedulersHERE
请注意,另一种替代方法是使用**PLINQ,默认情况下,它会分析您的查询并决定并行化是否会产生任何好处,同样,在阻塞IO的情况下,如果您确信启动多个线程将导致更好的执行,则可以使用WithExecutionMode(ParallelExecutionMode.ForceParallelism)强制并行化,然后可以使用WithDegreeOfParallelism**,提示要使用多少线程但是记住不保证您会得到那么多线程,如MSDN所述:
设置查询中使用的并行度。并行度是将用于处理查询的并发执行任务的最大数目
最后,我强烈推荐阅读关于ThreadingTPLTHIS系列文章。

rdlzhqv9

rdlzhqv92#

如果你增加任务的数量,比如说1000000,你会看到更多的线程随着时间的推移而产生,TPL倾向于每500ms注入一个。
TPL线程池不理解IO绑定的工作负载(睡眠就是IO)。在这些情况下,依赖TPL来选择正确的并行度不是一个好主意。TPL完全没有头绪,根据对吞吐量的模糊猜测注入更多的线程。这也是为了避免死锁。
这里,TPL策略显然没有用,因为添加的线程越多,获得的吞吐量就越大。在这种人为的情况下,每个线程每秒可以处理一个项目。TPL对此一无所知。将线程数限制为内核数是没有意义的。
什么决定了一次使用的线程数?
几乎没有文献记载的第三方物流启发式。他们经常出错。特别是在这种情况下,他们会随着时间的推移产生 * 无限数量的线程 *。使用任务管理器自己看看。让它运行一个小时,你会有1000个线程。
我怎样才能找回这个号码?我怎样才能更改这个号码?
你可以检索 * 一些 * 这些数字,但这不是正确的方式去。如果你需要一个有保证的DOP,你可以使用AsParallel().WithDegreeOfParallelism(...)或自定义任务调度程序。你也可以手动启动LongRunning任务。不要弄乱进程全局设置。

sxpgvts3

sxpgvts33#

我建议使用SemaphoreSlim,因为它不使用Windows内核(因此可以在Linux C#微服务中使用),并且还有一个count属性SemaphoreSlim.CurrentCount,因此您不需要Interlocked.IncrementInterlocked.Decrement

private static void SemaphoreImplementation()
{
    // Note that I am not passing 20,20 as arguments: You don't want to start with 20 initial number of entries 
    var semaphoreSlim = new SemaphoreSlim(0, 20);  // allow 20 tasks at a time

    for (int i = 0; i < 1000; i++)
    {
        var i1 = i;

        Task.Factory.StartNew(() =>
        {
            try
            {                        
                semaphoreSlim.Wait();

                foo(i1, semaphoreSlim.CurrentCount);
            }
            finally
            {
                semaphoreSlim.Release();
            }
        }, TaskCreationOptions.LongRunning);
    }
}

static void foo(int i, int semaphoreSlimCurrentCount)
{
    Thread.Sleep(100);
    Console.WriteLine($"foo{i:00} - on thread " + 
            $"{Thread.CurrentThread.ManagedThreadId:00}. Executing concurently: {semaphoreSlimCurrentCount}");
}

相关问题