【问题标题】:Code for a simple thread pool in C# [closed]C# 中简单线程池的代码 [关闭]
【发布时间】:2010-09-30 22:38:35
【问题描述】:

寻找一些简单线程池实现的示例代码 (C#)。

我在 codeproject 上找到了一个,但代码库非常庞大,我不需要所有这些功能。

无论如何,这更多是出于教育目的。

【问题讨论】:

  • 简短的回答是,除非是学习练习,否则您不应该自己动手。如果这是一个学习练习,你将通过自己编写而不是复制别人的代码来学到更多。 :)
  • @Greg:在任何情况下,您都可能需要一个独立于现有标准 ThreadPool 的线程池吗?
  • @Anthony:阅读 Joe Duffy(和其他人)的各种帖子中内置线程池中的内容,我有理由相信,我拼凑在一起的任何线程池都会比已经存在。

标签: c# .net multithreading threadpool


【解决方案1】:

没有必要自己实现,因为使用现有的 .NET 实现并不难。

来自ThreadPool Documentation

using System;
using System.Threading;

public class Fibonacci
{
    public Fibonacci(int n, ManualResetEvent doneEvent)
    {
        _n = n;
        _doneEvent = doneEvent;
    }

    // Wrapper method for use with thread pool.
    public void ThreadPoolCallback(Object threadContext)
    {
        int threadIndex = (int)threadContext;
        Console.WriteLine("thread {0} started...", threadIndex);
        _fibOfN = Calculate(_n);
        Console.WriteLine("thread {0} result calculated...", threadIndex);
        _doneEvent.Set();
    }

    // Recursive method that calculates the Nth Fibonacci number.
    public int Calculate(int n)
    {
        if (n <= 1)
        {
            return n;
        }

        return Calculate(n - 1) + Calculate(n - 2);
    }

    public int N { get { return _n; } }
    private int _n;

    public int FibOfN { get { return _fibOfN; } }
    private int _fibOfN;

    private ManualResetEvent _doneEvent;
}

public class ThreadPoolExample
{
    static void Main()
    {
        const int FibonacciCalculations = 10;

        // One event is used for each Fibonacci object
        ManualResetEvent[] doneEvents = new ManualResetEvent[FibonacciCalculations];
        Fibonacci[] fibArray = new Fibonacci[FibonacciCalculations];
        Random r = new Random();

        // Configure and launch threads using ThreadPool:
        Console.WriteLine("launching {0} tasks...", FibonacciCalculations);
        for (int i = 0; i < FibonacciCalculations; i++)
        {
            doneEvents[i] = new ManualResetEvent(false);
            Fibonacci f = new Fibonacci(r.Next(20,40), doneEvents[i]);
            fibArray[i] = f;
            ThreadPool.QueueUserWorkItem(f.ThreadPoolCallback, i);
        }

        // Wait for all threads in pool to calculation...
        WaitHandle.WaitAll(doneEvents);
        Console.WriteLine("All calculations are complete.");

        // Display the results...
        for (int i= 0; i<FibonacciCalculations; i++)
        {
            Fibonacci f = fibArray[i];
            Console.WriteLine("Fibonacci({0}) = {1}", f.N, f.FibOfN);
        }
    }
}

【讨论】:

  • 线程池有很大的局限性
  • 每个应用程序域一个池,无法尝试中止等待线程等。那里有大量信息stackoverflow.com/questions/145304codeproject.com/KB/threads/smartthreadpool.aspxcodeproject.com/KB/threads/cancellablethreadpool.aspx
  • @Jeffrey:OP 提出的这些限制在哪里?您在 OP 的哪里看到任何证据表明 OP 需要滚动自己的线程池?
  • @GEOCHET 限制不需要被提出来使其存在。此外,OP 明确谈到“教育目的”。为了进行基准测试,我需要一个专用的 ThreadPool,这样没有其他代码会干扰它,并且我可以在不影响我使用的库中的其他现有代码的情况下调整它,这可能会利用 .net 的内置代码。但即使我不需要自定义池,我可能仍然只是好奇如何实现一个。换句话说,我看到了很多人可能对自定义 impl 感兴趣的原因。
  • @EugeneBeresovksy:您可能想实际阅读 OP,阅读答案,然后阅读 cmets。也许还要检查 OP 的历史......
【解决方案2】:

这是我能想到的用于教育目的的最简单、幼稚的线程池实现(C# / .NET 3.5)。它没有以任何方式使用 .NET 的线程池实现。

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

namespace SimpleThreadPool
{
    public sealed class Pool : IDisposable
    {
        public Pool(int size)
        {
            this._workers = new LinkedList<Thread>();
            for (var i = 0; i < size; ++i)
            {
                var worker = new Thread(this.Worker) { Name = string.Concat("Worker ", i) };
                worker.Start();
                this._workers.AddLast(worker);
            }
        }

        public void Dispose()
        {
            var waitForThreads = false;
            lock (this._tasks)
            {
                if (!this._disposed)
                {
                    GC.SuppressFinalize(this);

                    this._disallowAdd = true; // wait for all tasks to finish processing while not allowing any more new tasks
                    while (this._tasks.Count > 0)
                    {
                        Monitor.Wait(this._tasks);
                    }

                    this._disposed = true;
                    Monitor.PulseAll(this._tasks); // wake all workers (none of them will be active at this point; disposed flag will cause then to finish so that we can join them)
                    waitForThreads = true;
                }
            }
            if (waitForThreads)
            {
                foreach (var worker in this._workers)
                {
                    worker.Join();
                }
            }
        }

        public void QueueTask(Action task)
        {
            lock (this._tasks)
            {
                if (this._disallowAdd) { throw new InvalidOperationException("This Pool instance is in the process of being disposed, can't add anymore"); }
                if (this._disposed) { throw new ObjectDisposedException("This Pool instance has already been disposed"); }
                this._tasks.AddLast(task);
                Monitor.PulseAll(this._tasks); // pulse because tasks count changed
            }
        }

        private void Worker()
        {
            Action task = null;
            while (true) // loop until threadpool is disposed
            {
                lock (this._tasks) // finding a task needs to be atomic
                {
                    while (true) // wait for our turn in _workers queue and an available task
                    {
                        if (this._disposed)
                        {
                            return;
                        }
                        if (null != this._workers.First && object.ReferenceEquals(Thread.CurrentThread, this._workers.First.Value) && this._tasks.Count > 0) // we can only claim a task if its our turn (this worker thread is the first entry in _worker queue) and there is a task available
                        {
                            task = this._tasks.First.Value;
                            this._tasks.RemoveFirst();
                            this._workers.RemoveFirst();
                            Monitor.PulseAll(this._tasks); // pulse because current (First) worker changed (so that next available sleeping worker will pick up its task)
                            break; // we found a task to process, break out from the above 'while (true)' loop
                        }
                        Monitor.Wait(this._tasks); // go to sleep, either not our turn or no task to process
                    }
                }

                task(); // process the found task
                lock(this._tasks)
                {
                    this._workers.AddLast(Thread.CurrentThread);
                }
                task = null;
            }
        }

        private readonly LinkedList<Thread> _workers; // queue of worker threads ready to process actions
        private readonly LinkedList<Action> _tasks = new LinkedList<Action>(); // actions to be processed by worker threads
        private bool _disallowAdd; // set to true when disposing queue but there are still tasks pending
        private bool _disposed; // set to true when disposing queue and no more tasks are pending
    }


    public static class Program
    {
        static void Main()
        {
            using (var pool = new Pool(5))
            {
                var random = new Random();
                Action<int> randomizer = (index =>
                {
                    Console.WriteLine("{0}: Working on index {1}", Thread.CurrentThread.Name, index);
                    Thread.Sleep(random.Next(20, 400));
                    Console.WriteLine("{0}: Ending {1}", Thread.CurrentThread.Name, index);
                });

                for (var i = 0; i < 40; ++i)
                {
                    var i1 = i;
                    pool.QueueTask(() => randomizer(i1));
                }
            }
        }
    }
}

【讨论】:

  • +1 谢谢。我正在使用这个sn-p,但在很长一段时间后,我遇到了一个错误:Unhandled Exception: System.NullReferenceException: Object reference not set to an instance of an object. at System.Collections.Generic.LinkedList'1.InternalInsertNodeBefore(LinkedListNode&gt;' node, LinkedListNode'1 newNode) at System.Collections.Generic.LinkedList'1.AddLast(T value) at Prog。 Pool.Worker()`。知道是什么原因造成的吗?
  • @Legend 不确定问题可能是什么,但如果我不得不猜测,我会说这与_workers 链表在锁定之外被访问的事实有关。如果使用 .NET 4,您可以尝试改用 ConcurrentQueue&lt;Action&gt;
  • +1 谢谢。你说的对。我在这里问了一个问题:stackoverflow.com/questions/16763626/… 看来问题确实是由于缺少锁造成的。谢谢你的时间。我目前正在使用 .NET 3.5,这就像一个魅力。
  • 您的代码很棒,但我注意到 dispose 似乎没有等待线程正确结束。尝试在处理池后在 main 中添加以下行:Console.WriteLine("Thread pool disposed!");Thread.Sleep(2000);。然后在该行之后放置一个断点(只是为了避免控制台关闭)。您会注意到在Thread pool disposed! 之后,工作人员将继续编写他们的操作(如Worker 0: Ending 39
  • 对我来说,问题可能是如果一个 Worker 正在运行一个 Action,它不在 List this._workers 中,因此,Dispose 方法(在 foreach (var worker in this._workers) 中)不会等待它结束
猜你喜欢
  • 2011-05-09
  • 2011-04-28
  • 1970-01-01
  • 2010-11-29
  • 1970-01-01
  • 1970-01-01
  • 2012-10-24
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多