QueueUserWorkItem 执行需要多长时间? [英] How long should QueueUserWorkItem take to execute?

问题描述

我正在开发的应用程序在处理完数据后(从不超过 1 毫秒)处理来自 Blocking 集合的消费者线程中的数据，它将处理后的数据从 ThreadPool 旋转到一个新线程中，以发送给我的用户并存储在数据库中.

The application I am developing processes data in consumer threads from a Blocking collection when it is done processing the data (never any more than 1ms) it spins that processed data off into a new thread from the ThreadPool to be sent to my users and stored in the database.

我在调用 ThreadPool.QueueUserWorkItem 之前启动秒表并停止它，这是我在 ThreadPool.QueueUserWorkItem 调用的函数中做的第一件事(在此之前的任何代码的计时都小于 1 毫秒).

I start a stopwatch just before I call ThreadPool.QueueUserWorkItem and stop it as one of the first things I do in the function called by ThreadPool.QueueUserWorkItem (any code before this has been clocked at less than 1ms).

这个秒表报告的时间让我有点担心，平均时间是 4 毫秒(没问题)但最大值超过 900 毫秒，使用过的线程最少是 1，最常用的是 ~60，平均是 ~40.

The time this stopwatch is reporting is worrying me a little, the average time is 4ms (no problem there) but the max is upwards of 900ms, the least threads ever used is 1, the most used is ~60 and the average is ~40.

尽管只要平均值保持不变，它应该不是问题，但我很想知道为什么我通常有 900 多个空闲时间时偶尔会等待大约 1 秒的线程.

Although as long as the average stays the same it should not be a problem I would love to know why I get the occasional ~1sec wait for a thread when I normally have 900+ free.

一些示例代码将独立运行，睡眠取代实际处理:

Some example code that will run independently with sleeps replacing actual processing:

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApplication2
{
    class Program
    {
        static int[] AvailableThreads;
        static Stopwatch[] stopwatch;
        static BlockingCollection<int> Queue;
        static ManualResetEvent consumerEvent;
        static ManualResetEvent alldone;
        static bool disposed = false;

        static void Main(string[] args)
        {
            int iterations = 10000;
            stopwatch = new Stopwatch[iterations];
            AvailableThreads = new int[iterations];
            Queue = new BlockingCollection<int>();
            consumerEvent = new ManualResetEvent(false);
            alldone = new ManualResetEvent(false);

            Thread processThread = new Thread(new ThreadStart(ProcessThread));
            processThread.IsBackground = true;
            processThread.Start();

            int MaxThreads = 0;
            int y = 0;

            ThreadPool.GetMaxThreads(out MaxThreads, out y);

            for (int i = 0; i < stopwatch.Length; i++)
            {
                Queue.Add(i);
                consumerEvent.Set();
            }

            alldone.Reset();
            alldone.WaitOne();

            long av = 0;
            long max = 0;

            int threadsAv = 0;
            int threadsMax = 0;

            for (int i = 0; i < stopwatch.Length; i++)
            {
                long ms = stopwatch[i].ElapsedMilliseconds;
                av += ms;
                threadsAv += AvailableThreads[i];
                if (max < ms) max = ms;
                if (threadsMax < AvailableThreads[i]) threadsMax = AvailableThreads[i];
            }

            if(av != 0) av = av / stopwatch.Length;

            if (threadsAv != 0) threadsAv = threadsAv / stopwatch.Length;

            Console.WriteLine("Average Time: {0}, Max Time: {1}, Max Thread: {2}, Average Available Threads: {3}, Max Available Threads: {4}", av, max, MaxThreads, threadsAv, threadsMax);

            Console.ReadLine();

            disposed = true;
        }

        static void ProcessThread()
        {
            while (!disposed)
            {
                foreach (int i in Queue.GetConsumingEnumerable())
                {
                    if (disposed) return;
                    // Proccess a bit of data here .....
                    stopwatch[i] = new Stopwatch();
                    stopwatch[i].Start();
                    int y = 0;
                    ThreadPool.GetAvailableThreads(out AvailableThreads[i], out y);
                    ThreadPool.QueueUserWorkItem(TransmitThread, i);
                }

                consumerEvent.Reset();
                consumerEvent.WaitOne();
            }
        }

        static void TransmitThread(object data)
        {
            int i = (int)data;
            stopwatch[i].Stop();
            //Fake some work.

            Thread.Sleep(1);

            if (i == stopwatch.Length - 1) alldone.Set();
        }
    }
}

示例输出:

Average Time: 581, Max Time: 1126, Max Thread: 1023, Average Available Threads: 1015, Max Available Threads: 1023

有人可以对此提供一些见解吗?

Can anyone provide some insight on this?

推荐答案

我可以确认问题出在从超过线程池上的最小值，增加这个数字显着提高性能.

I can confirm that the issue is with the throttling incurred from exceeding the min value on the thread pool, increasing this number significantly improves performance.

您可以增加将立即创建的最小线程数.这样做的唯一负面后果是，如果您的工作负载长时间不需要这些线程，它可能会导致更高的内存使用量.这是您问题的有效解决方案.

You can increase the minimum number of threads that will be created immediately. The only negative consequence this has is that it might result in higher memory usage in case your workload does not need those threads for a long time. This is a valid solution to your problem.

另一方面:为什么要运行数百甚至数千个 (?) 线程?！这当然是受支持的，而且并非不可靠.但它非常不寻常，并暗示了架构问题.考虑使用异步 IO 和异步等待.您不必使一切异步.只是那些花费大部分时间阻塞的地方.可能有少数地方会导致 99% 的阻塞.使这些异步，您的线程数就会下降到正常水平.

On the other hand: Why do you have hundreds or even thousands (?) of threads running?! This is certainly supported and not unreliable. But it is very unusual and hints at architectural problems. Consider using async IO and async waiting. You don't have to make everything async. Just those places where most of the time blocking is spent. Probably there are a handful of places that cause 99% of the blocking. Make those async and your thread counts are down to normal levels.

这篇关于QueueUserWorkItem 执行需要多长时间?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！