C# 等待 ThreadPool 中的所有线程终止

Question

我有一个主线程和许多其他后台线程。

这些后台线程的主要用途是查询数据（许多查询来自网络，这就是我创建多个线程的原因：以避免用户界面滞后）。

当涉及到在主线程（用户界面）中导出数据时，我需要等到所有其他线程完成。

我的代码是：

//...code to open save file dialog...

//this loop is to wait for all the threads finish their query
//QueryThread.threadCount is the count of the background threads
while (QueryThread.threadCount != 0)
{
    Thread.CurrentThread.Join(1000);
    Console.WriteLine(QueryThread.threadCount);
}

//...code to export data...

如果我注释掉 while 循环，程序将顺利运行，但我导出的一些数据将有可能显示一些“不需要的”材料，因为某些后台线程尚未完成其工作。

然而，上面的 while 循环是无限的，threadCount 永远不会改变，这意味着在“Join()”方法期间，没有后台线程在运行。

为什么后台线程被阻塞以及如何解决该问题？

多谢！

Answer 1

您正在当前线程上调用 Join 方法，但该方法不会很有意义。您应该在工作线程上调用它：

foreach (Thread thread in workerThreads)
{
    thread.Join(1000);
}

不幸的是，这种方式违背了使用线程的目的，因为它将阻止调用，直到所有其他线程完成为止。

的 RunWorkerCompleted 事件BackgroundWorker 可用于通知后台任务完成并在表单上执行更新。

Answer 2

您的实现是错误的，您不应该使用 Join 作为线程之间的同步原语。

您应该做的是实现生产者-消费者模式。这将允许您让线程等待执行工作，然后当您将其放入队列时，线程会活跃起来执行该工作。

但是，我要做的更改是，从 UI 线程中，不要将数据直接添加到生产者和消费者共享的队列中。相反，制作该数据的副本，然后将该放入队列中。

有关如何在 .NET 中实现生产者-消费者模式的更多信息，我建议您阅读标题为“如何：同步生产者线程和消费者线程（C# 编程指南）"

Answer 3

我想你想研究一下信号。为您的线程提供信号（ManualResetEvent/AutoResetEvent）。完成后，在工作线程中 Set() 关联的信号句柄。在主线程中执行“WaitAll(signal1, signal2,signal3)”来等待工作线程完成。

希望这有帮助，

Answer 4

我忍不住自己尝试一些。我确信还有改进的空间，但我认为它展示了如何处理一些多线程问题，包括原始问题。

Form.cs

namespace STAFormWithThreadPoolSync
{
    internal delegate void WorkerEvent(WorkerEventInfo info);

    public partial class Form1 : Form
    {
        // We'll create a state object for each worker process
        List<WorkerState> workerStates = new List<WorkerState>();

        public Form1()
        {
            InitializeComponent();
        }

        // Executed in the main thread
        private void button1_Click(object sender, EventArgs e)
        {
            workersList.Items.Clear();

            // Read the amount of thread we should start from the form
            int threadCountToUse = (int)ThreadCount.Value;

            WorkerEvent woEvent = new WorkerEvent(this.workerEventOccured);

            // Start up all threads
            for (int counter = 0; counter < threadCountToUse; ++counter)
            {
                // An object we can pass values into for the worker process to use.
                WorkerState workerState = new WorkerState();

                workerState.OnStarted += woEvent;
                workerState.OnFinished += woEvent;

                // Register for the signal (and store its registered wait handle in the stateObj, which we also pass into the parameters!)
                workerState.registeredWaitHandle = ThreadPool.RegisterWaitForSingleObject(workerState.finishSignal, this.ItemHasFinished, workerState, -1, true); 

                // Store the state object for later use.
                workerStates.Add(workerState);
            }

            WorkersProgress.Minimum = 0;
            WorkersProgress.Maximum = workerStates.Count;

            workerStates.ForEach(workerState =>
                {
                    // Fire of the worker thread (with the state object)
                    ThreadPool.QueueUserWorkItem(this.ProcessItem, workerState);
                }
            );



            button1.Enabled = false;
            CurrentResult.Value = 0;
            CurrentResultLabel.Text = "Current value";
            ProgressTimer.Start();
        }

        // event is run on the callers thread, so carefull accessing our controls on our form.
        internal void workerEventOccured(WorkerEventInfo info)
        {
            if (this.workersList.InvokeRequired)
            {
                WorkerEvent workerEvent = new WorkerEvent(workerEventOccured);
                this.Invoke(workerEvent, new object[] { info });
            }
            else
            {
                switch (info.eventType)
                {
                    case EventType.WorkerStarted:
                        this.workersList.Items.Add(String.Format("Worker started on thread : {0}", info.workerState.threadId));
                        break;

                    case EventType.WorkerEnded:
                        this.workersList.Items.Add(String.Format("Worker finished on thread : {0}", info.workerState.threadId));
                        break;
                    case EventType.AllWorkersFinished:
                        this.workersList.Items.Add("ALL workers finished");
                        ProgressTimer.Stop();
                        button1.Enabled = true;
                        CurrentResultLabel.Text = "Final value";
                        break;
                }
            }
        }

        // Executed in threadpool thread.
        private void ProcessItem(object state)
        {
            WorkerState workerState = state as WorkerState;
            int threadId = Thread.CurrentThread.ManagedThreadId;
            workerState.threadId = threadId.ToString();

            WorkerEventInfo weInfo = new WorkerEventInfo();
            weInfo.eventType = EventType.WorkerStarted;
            weInfo.workerState = workerState;
            workerState.Started(weInfo);

            // Simulate work for ((threadid / 2) seconds.
            Thread.Sleep((threadId * 500));

            // Set the result in the state object to the threadId; 
            workerState.result = threadId;

            // Signal that this thread is done.
            workerState.finishSignal.Set();
        }

        // Executed in threadpool thread
        private void ItemHasFinished(object state, bool timedOut)
        {
            // get our state object
            WorkerState workerState = state as WorkerState;

            WorkerEventInfo weInfo = new WorkerEventInfo();
            weInfo.eventType = EventType.WorkerEnded;
            weInfo.workerState = workerState;

            workerState.Finished(weInfo);
        }

        private void ProgressTimer_Tick(object sender, EventArgs e)
        {
            List<WorkerState> removeStates = new List<WorkerState>();
            workerStates.ForEach(workerState =>
                {
                    if (workerState.finishSignal.WaitOne(0))
                    {
                        CurrentResult.Value += workerState.result;
                        removeStates.Add(workerState);
                    }
                }
            );

            removeStates.ForEach(workerState =>
                {
                    workerState.registeredWaitHandle.Unregister(workerState.finishSignal);
                    workerStates.Remove(workerState);
                }
            );


            WorkersProgress.Value = workerStates.Count;
            if (workerStates.Count == 0)
            {
                WorkerEventInfo weInfo = new WorkerEventInfo();
                weInfo.eventType = EventType.AllWorkersFinished;
                weInfo.workerState = null;
                this.workerEventOccured(weInfo);
            }

        }
    }

    internal class WorkerState
    {
        internal string threadId = "";
        internal int    result = 0;
        internal RegisteredWaitHandle registeredWaitHandle = null;
        internal AutoResetEvent finishSignal = new AutoResetEvent(false);
        internal event WorkerEvent OnStarted = new WorkerEvent( (info) => {});
        internal event WorkerEvent OnFinished = new WorkerEvent((info) => { });

        internal void Started(WorkerEventInfo info)
        {
            OnStarted(info);
        }

        internal void Finished(WorkerEventInfo info)
        {
            OnFinished(info);
            this.finishSignal.Set();
        }
    }

    internal enum EventType
    {
        WorkerStarted,
        WorkerEnded,
        AllWorkersFinished
    }

    internal class WorkerEventInfo
    {
        internal EventType eventType;
        internal WorkerState workerState;
    }
}

Form.Designer.cs

namespace STAFormWithThreadPoolSync
{
    partial class Form1
    {
        /// 
        /// Required designer variable.
        /// 
        private System.ComponentModel.IContainer components = null;

        /// 
        /// Clean up any resources being used.
        /// 
        /// true if managed resources should be disposed; otherwise, false.
        protected override void Dispose(bool disposing)
        {
            if (disposing && (components != null))
            {
                components.Dispose();
            }
            base.Dispose(disposing);
        }

        #region Windows Form Designer generated code

        /// 
        /// Required method for Designer support - do not modify
        /// the contents of this method with the code editor.
        /// 
        private void InitializeComponent()
        {
            this.components = new System.ComponentModel.Container();
            this.button1 = new System.Windows.Forms.Button();
            this.ThreadCount = new System.Windows.Forms.NumericUpDown();
            this.workersList = new System.Windows.Forms.ListView();
            this.WorkerProcessColumn = new System.Windows.Forms.ColumnHeader();
            this.ProgressTimer = new System.Windows.Forms.Timer(this.components);
            this.WorkersProgress = new System.Windows.Forms.ProgressBar();
            this.CurrentResultLabel = new System.Windows.Forms.Label();
            this.CurrentResult = new System.Windows.Forms.NumericUpDown();
            this.label2 = new System.Windows.Forms.Label();
            ((System.ComponentModel.ISupportInitialize)(this.ThreadCount)).BeginInit();
            ((System.ComponentModel.ISupportInitialize)(this.CurrentResult)).BeginInit();
            this.SuspendLayout();
            // 
            // button1
            // 
            this.button1.Location = new System.Drawing.Point(212, 19);
            this.button1.Name = "button1";
            this.button1.Size = new System.Drawing.Size(93, 23);
            this.button1.TabIndex = 0;
            this.button1.Text = "Start threads";
            this.button1.UseVisualStyleBackColor = true;
            this.button1.Click += new System.EventHandler(this.button1_Click);
            // 
            // ThreadCount
            // 
            this.ThreadCount.Location = new System.Drawing.Point(23, 21);
            this.ThreadCount.Minimum = new decimal(new int[] {
            2,
            0,
            0,
            0});
            this.ThreadCount.Name = "ThreadCount";
            this.ThreadCount.Size = new System.Drawing.Size(183, 20);
            this.ThreadCount.TabIndex = 1;
            this.ThreadCount.Value = new decimal(new int[] {
            4,
            0,
            0,
            0});
            // 
            // workersList
            // 
            this.workersList.Columns.AddRange(new System.Windows.Forms.ColumnHeader[] {
            this.WorkerProcessColumn});
            this.workersList.Location = new System.Drawing.Point(23, 80);
            this.workersList.Name = "workersList";
            this.workersList.Size = new System.Drawing.Size(486, 255);
            this.workersList.TabIndex = 3;
            this.workersList.UseCompatibleStateImageBehavior = false;
            this.workersList.View = System.Windows.Forms.View.Details;
            // 
            // WorkerProcessColumn
            // 
            this.WorkerProcessColumn.Text = "Worker process";
            this.WorkerProcessColumn.Width = 482;
            // 
            // ProgressTimer
            // 
            this.ProgressTimer.Interval = 200;
            this.ProgressTimer.Tick += new System.EventHandler(this.ProgressTimer_Tick);
            // 
            // WorkersProgress
            // 
            this.WorkersProgress.Location = new System.Drawing.Point(112, 341);
            this.WorkersProgress.Name = "WorkersProgress";
            this.WorkersProgress.Size = new System.Drawing.Size(397, 24);
            this.WorkersProgress.TabIndex = 4;
            // 
            // CurrentResultLabel
            // 
            this.CurrentResultLabel.AutoSize = true;
            this.CurrentResultLabel.Location = new System.Drawing.Point(578, 266);
            this.CurrentResultLabel.Name = "CurrentResultLabel";
            this.CurrentResultLabel.Size = new System.Drawing.Size(74, 13);
            this.CurrentResultLabel.TabIndex = 5;
            this.CurrentResultLabel.Text = "Current Result";
            // 
            // CurrentResult
            // 
            this.CurrentResult.Location = new System.Drawing.Point(581, 282);
            this.CurrentResult.Maximum = new decimal(new int[] {
            -1593835520,
            466537709,
            54210,
            0});
            this.CurrentResult.Name = "CurrentResult";
            this.CurrentResult.ReadOnly = true;
            this.CurrentResult.Size = new System.Drawing.Size(169, 20);
            this.CurrentResult.TabIndex = 6;
            // 
            // label2
            // 
            this.label2.AutoSize = true;
            this.label2.Location = new System.Drawing.Point(25, 352);
            this.label2.Name = "label2";
            this.label2.Size = new System.Drawing.Size(81, 13);
            this.label2.TabIndex = 7;
            this.label2.Text = "processing load";
            // 
            // Form1
            // 
            this.AutoScaleDimensions = new System.Drawing.SizeF(6F, 13F);
            this.AutoScaleMode = System.Windows.Forms.AutoScaleMode.Font;
            this.ClientSize = new System.Drawing.Size(762, 377);
            this.Controls.Add(this.label2);
            this.Controls.Add(this.CurrentResult);
            this.Controls.Add(this.CurrentResultLabel);
            this.Controls.Add(this.WorkersProgress);
            this.Controls.Add(this.workersList);
            this.Controls.Add(this.ThreadCount);
            this.Controls.Add(this.button1);
            this.Name = "Form1";
            this.Text = "Form1";
            ((System.ComponentModel.ISupportInitialize)(this.ThreadCount)).EndInit();
            ((System.ComponentModel.ISupportInitialize)(this.CurrentResult)).EndInit();
            this.ResumeLayout(false);
            this.PerformLayout();

        }

        #endregion

        private System.Windows.Forms.Button button1;
        private System.Windows.Forms.NumericUpDown ThreadCount;
        private System.Windows.Forms.ListView workersList;
        private System.Windows.Forms.ColumnHeader WorkerProcessColumn;
        private System.Windows.Forms.Timer ProgressTimer;
        private System.Windows.Forms.ProgressBar WorkersProgress;
        private System.Windows.Forms.Label CurrentResultLabel;
        private System.Windows.Forms.NumericUpDown CurrentResult;
        private System.Windows.Forms.Label label2;
    }
}

希望这有帮助，

Answer 5

我通过改变生产者-消费者模型的方法解决了这个问题。

谢谢大家。
请查看此链接（由casperOne 上面），但要小心不要遵循微软的实施......

Go 这里 相反会给你一个更好的答案。

当然我做了一些改变，在我的例子中队列的类型是Delegate。

public static class QueryThread
{
    private static SyncEvents _syncEvents = new SyncEvents();
    private static Queue<Delegate> _queryQueue = new Queue<Delegate>();

    static Producer queryProducer;
    static Consumer queryConsumer;

    public static void init()
    {
        queryProducer = new Producer(_queryQueue, _syncEvents);
        queryConsumer = new Consumer(_queryQueue, _syncEvents);

        Thread producerThread = new Thread(queryProducer.ThreadRun);
        Thread consumerThread = new Thread(queryConsumer.ThreadRun);

        producerThread.IsBackground = true;
        consumerThread.IsBackground = true;

        producerThread.Start();
        consumerThread.Start();
    }

    public static void Enqueue(Delegate item)
    {
        queryQueue.Enqueue(item);
    }
}

当主线程需要查询时，通过调用Enqueue(Delegate item)将委托指向进行查询的函数入队。这会将委托添加到生产者的“私有”队列中。

生产者会在适当的场合将自己队列中的项目添加到共享队列中（例如msdn示例中生成随机数并将其放入共享队列中）。

消费者使代表出队并运行它们。

感谢大家的帮助。 =]