问题 在.NET中是否存在用于长时间运行线程的线程调度程序?


我们的场景是网络扫描仪。

它连接到一组主机并使用低优先级后台线程并行扫描它们一段时间。

我希望能够安排大量工作,但只有任何给定的说十个或同时扫描的任何数量的主机。即使我创建自己的线程,许多回调和其他异步善良使用ThreadPool,我最终耗尽资源。我应该看一下MonoTorrent ......

如果我使用ThreadPool,我可以将我的应用程序限制为某个数字,这些数字将足以让应用程序的其余部分顺利运行吗?

是否有一个线程池,我可以初始化为n长寿线程?

[编辑] 似乎没有人注意到我对一些回复做了一些评论,所以我会在这里添加一些东西。

  • 线程应该可以取消 优雅而有力。
  • 线程应具有低优先级,使GUI响应。
  • 线程长时间运行,但是按顺序(分钟)而不是订单(天)。

针对给定目标主机的工作基本上是:

  For each test
    Probe target (work is done mostly on the target end of an SSH connection)
    Compare probe result to expected result (work is done on engine machine)
  Prepare results for host

有人可以解释为什么使用SmartThreadPool标记有消极的用处吗?


1514
2018-04-22 13:15


起源

Jeff Sternal的答案看起来很合适,但是你可能最好不要创建自己的BackWorker System.Threading.Thread 实例。但是,您还提到即使在您自己的线程中,由于异步操作,您也会耗尽线程池资源。也许您应该解释为什么要在专门用于运行这些操作的线程中执行异步操作。 - Michael Petito
经过一番搜索,我遇到了另一个ThreadPool smartthreadpool.codeplex.com 哪个可能会到位。我在工作线程中使用asynch东西来处理输出和错误。随着应用程序的运行,用户将获得反馈,以便他们了解每项任务的进度。 - LogicMagic
应用程序使用GTK#和MVC模式编写。有一个GUI,当某些事件触发时,GUI线程被编组并进行更新。扫描不是简单的ping,应用程序是经过身份验证的扫描程序,可以查看目标的配置。扫描应该花费O(分钟)时间来完成。每个主机。由于显而易见的原因,每个主机报告在主机线程上的引擎计算机上完成。每个连接创建一个过程,这可能是昂贵的,但与几分钟相比无关紧要。 - LogicMagic
只是为了迂腐,C#根本没有任何线程调度。实际上,它没有线程。当然,.NET有这两者。 - John Saunders


答案:


在.NET 4中,您已经集成了 任务并行库。当您创建新任务(新线程抽象)时,您可以指定任务 长跑。我们已经取得了很好的经验(长期是几天而不是几分钟或几小时)。

您也可以在.NET 2中使用它,但它实际上是一个扩展,检查 这里

在VS2010中 调试并行应用程序 基于任务(不是线程)已经 从根本上改善了。建议尽可能使用Tasks而不是原始线程。因为它允许您以更加面向对象的友好方式处理并行性。

UPDATE
未指定为的任务 长跑,排队到线程池(或任何其他调度程序)。
但是如果指定了任务 长跑,它只是创造了一个 独立线程没有线程池


8
2018-04-22 13:22



TPL不在ThreadPool之上吗? - LogicMagic
TPL的目的是在线程和调度不同任务时添加一个有意义的抽象层。 TPL的美妙之处在于你可以指定一个 调度。调度程序 能够 实际上是线程池。但你也可以定义自己的。欲了解更多信息: bit.ly/aW4Lq4 和 bit.ly/9VAkbf - ntziolis


CLR ThreadPool 不适合执行长时间运行的任务:它用于执行短任务,其中创建线程的成本几乎与执行方法本身一样高。 (或者至少占执行该方法所花费的时间的很大一部分。)正如您所见,.NET本身消耗线程池线程,您不能为自己保留它们的块,以免冒着运行时的风险。

调度,限制和取消工作是另一回事。没有其他内置的.NET工作队列线程池,所以你将自己滚动(管理 线程 要么 BackgroundWorkers 你自己)或找到一个先前存在的(Ami Bar的SmartThreadPool 看起来很有希望,虽然我自己没有用它)。


5
2018-04-22 13:18



后台工作人员被认为是在保持UI响应的同时安排长时间运行的任务。但是,它们不是应用程序逻辑线程的正确选择。 - ntziolis
@ntziolis - 我同意,虽然我们没有足够的关于应用程序的信息来完全排除它:'网络扫描仪'可能意味着 一个GUI应用程序。 - Jeff Sternal
这是真的,也许创作者可以添加一些额外的信息? - ntziolis
池的原因是计划所有工作并设置应在任何给定时间运行的并发线程总数的最大值。我也忘了提到用户应该能够中止线程。 - LogicMagic


在您的特定情况下,最佳选项不是线程或线程池或后台工作程序,而是框架提供的异步编程模型(BeginXXX,EndXXX)。

使用的好处 异步模型 是每当有数据要读取时,TcpIp堆栈都会使用回调,并且回调会自动在线程池的线程上运行。

使用 异步模型,您可以控制每个启动时间间隔的请求数量,如果您愿意,您可以在正常优先级线程上处理请求时启动来自较低优先级线程的所有请求,这意味着数据包将在内部保留尽可能少的内容网络堆栈的Tcp队列。

异步客户端套接字示例 - MSDN

附:对于多个并发和长时间运行的作业,它们不进行大量计算但主要等待IO(网络,磁盘等),更好的选择始终是使用回调机制而不是线程。


1
2018-04-22 13:49





我会创建自己的线程管理器。在下面的简单示例中,Queue用于保存等待线程,而Dictionary用于保存活动线程,由ManagedThreadId键入。线程完成后,它会从活动字典中删除自己,并通过回调启动另一个线程。

您可以从UI更改最大运行线程限制,并且可以将额外信息传递给ThreadDone回调以监视性能等。如果线程失败,例如网络超时,则可以重新插入队列。向Supervisor添加额外的控制方法,用于暂停,停止等。

using System;
using System.Collections.Generic;
using System.Threading;

namespace ConsoleApplication1
{
    public delegate void CallbackDelegate(int idArg);

    class Program
    {
        static void Main(string[] args)
        {
            new Supervisor().Run();
            Console.WriteLine("Done");
            Console.ReadKey();
        }
    }

    class Supervisor
    {
        Queue<System.Threading.Thread> waitingThreads = new Queue<System.Threading.Thread>();
        Dictionary<int, System.Threading.Thread> activeThreads = new Dictionary<int, System.Threading.Thread>();
        int maxRunningThreads = 10;
        object locker = new object();
        volatile bool done;

        public void Run()
        {
            // queue up some threads
            for (int i = 0; i < 50; i++)
            {
                Thread newThread = new Thread(new Worker(ThreadDone).DoWork);
                newThread.IsBackground = true;
                waitingThreads.Enqueue(newThread);
            }
            LaunchWaitingThreads();
            while (!done) Thread.Sleep(200);
        }

        // keep starting waiting threads until we max out
        void LaunchWaitingThreads()
        {
            lock (locker)
            {
                while ((activeThreads.Count < maxRunningThreads) && (waitingThreads.Count > 0))
                {
                    Thread nextThread = waitingThreads.Dequeue();
                    activeThreads.Add(nextThread.ManagedThreadId, nextThread);
                    nextThread.Start();
                    Console.WriteLine("Thread " + nextThread.ManagedThreadId.ToString() + " launched");
                }
                done = (activeThreads.Count == 0) && (waitingThreads.Count == 0);
            }
        }

        // this is called by each thread when it's done
        void ThreadDone(int threadIdArg)
        {
            lock (locker)
            {
                // remove thread from active pool
                activeThreads.Remove(threadIdArg);
            }
            Console.WriteLine("Thread " + threadIdArg.ToString() + " finished");
            LaunchWaitingThreads(); // this could instead be put in the wait loop at the end of Run()
        }
    }

    class Worker
    {
        CallbackDelegate callback;
        public Worker(CallbackDelegate callbackArg)
        {
            callback = callbackArg;
        }

        public void DoWork()
        {
            System.Threading.Thread.Sleep(new Random().Next(100, 1000));
            callback(System.Threading.Thread.CurrentThread.ManagedThreadId);
        }
    }
}

1
2018-04-22 19:24





使用内置线程池。它具有良好的能力。

或者你可以看看 这里有智能线程池实现 或 扩展线程池 限制最大工作线程数。


-2
2018-04-22 13:19