我有兴趣从您那里获得一些关于什么是好的/更好的遵循以下规则的线程架构的想法：
线程必须在应用程序的整个生命周期内运行，如果队列中没有要执行的工作，则处于睡眠/等待模式。
线程的优先级必须低于正常（这消除了使用 ThreadPool 的可能性）。
线程必须在任务完成后将其反馈给主线程。
线程将监视一个队列
我正在使用 .Net Framework 4.0
让我知道你的想法:)
【问题讨论】：
标签： c# .net multithreading .net-4.0
当我需要实现自己的多线程处理时，我通常会使用这样的东西：
public class MyWorker<T> : IDisposable
{
private readonly Queue<T> _taskQueue; // task queue
private readonly object _threadLock = new object();
private Thread _thread; // worker thread
private ManualResetEvent _evExit;
private AutoResetEvent _evNewData;
/// <summary>Override this to process data.</summary>
protected abstract void ProcessData(T data);
/// <summary>Override this to set other thread priority.</summary>
protected virtual ThreadPriority ThreadPriority
{
get { return ThreadPriority.BelowNormal; }
}
protected MyWorker()
{
_taskQueue = new Queue<T>();
_evExit = new ManualResetEvent(false);
_evNewData = new AutoResetEvent(false);
}
~MyWorker()
{
Dispose(false);
}
private void ThreadProc()
{
try
{
var wh = new WaitHandle[] { _evExit, _evNewData };
while(true)
{
T data = default(T);
bool gotData = false;
lock(_taskQueue) // sync
{
if(_taskQueue.Count != 0) // have data?
{
data = _taskQueue.Dequeue();
gotData = true;
}
}
if(!gotData)
{
if(WaitHandle.WaitAny(wh) == 0) return; // demanded stop
continue; //we have data now, grab it
}
ProcessData(data);
if(_evExit.WaitOne(0)) return;
}
}
catch(ThreadInterruptedException)
{
// log warning - this is not normal
}
catch(ThreadAbortException)
{
// log warning - this is not normal
}
}
public void Start()
{
lock(_threadLock)
{
if(_thread != null)
throw new InvalidOperationException("Already running.");
_thread = new Thread(ThreadProc)
{
Name = "Worker Thread",
IsBackground = true,
Priority = ThreadPriority,
};
_thread.Start();
}
}
public void Stop()
{
lock(_threadLock)
{
if(_thread == null)
throw new InvalidOperationException("Is not running.");
_evExit.Set();
if(!_thread.Join(1000))
_thread.Abort();
_thread = null;
}
}
/// <summary>Enqueue data for processing.</summary>
public void EnqueueData(T data)
{
lock(_taskQueue)
{
_taskQueue.Enqueue(data);
_evNewData.Set(); // wake thread if it is sleeping
}
}
/// <summary>Clear all pending data processing requests.</summary>
public void ClearData()
{
lock(_taskQueue)
{
_taskQueue.Clear();
_evNewData.Reset();
}
}
protected virtual void Dispose(bool disposing)
{
lock(_threadLock)
{
if(_thread != null)
{
_evExit.Set();
if(!_thread.Join(1000))
_thread.Abort();
_thread = null;
}
}
_evExit.Close();
_evNewData.Close();
if(disposing)
_taskQueue.Clear();
}
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
}
【讨论】：
- 线程的优先级必须低于正常（这消除了使用 ThreadPool 的可能性）。
这似乎是使用 TPL 和 ThreadPool 的主要障碍。你确定你没有高估低优先级的有用性吗？
您将不得不投入大量工作才能提出一些总是比 TPL 更不强大（并且测试/可靠程度更低）的东西。
我会重新考虑这一点。
【讨论】：
通过阅读以上条件
一些问题
1- 是否有任何其他线程将填充队列中的作业
如果答案是肯定的，那么这里可以使用生产者/消费者设计模式，我不知道 .net 4.0，但这种设计可以在 .net 3.5 中实现。
例如，请参阅here。
【讨论】：
就我个人而言，我通常自己动手，因为我喜欢更严格的控制。
我在媒体浏览器中使用它：
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Diagnostics;
using MediaBrowser.Library.Logging;
namespace MediaBrowser.Library.Threading {
public static class Async {
public const string STARTUP_QUEUE = "Startup Queue";
class ThreadPool {
List<Action> actions = new List<Action>();
List<Thread> threads = new List<Thread>();
string name;
volatile int maxThreads = 1;
public ThreadPool(string name) {
Debug.Assert(name != null);
if (name == null) {
throw new ArgumentException("name should not be null");
}
this.name = name;
}
public void SetMaxThreads(int maxThreads) {
Debug.Assert(maxThreads > 0);
if (maxThreads < 1) {
throw new ArgumentException("maxThreads should be larger than 0");
}
this.maxThreads = maxThreads;
}
public void Queue(Action action, bool urgent) {
Queue(action, urgent, 0);
}
public void Queue(Action action, bool urgent, int delay) {
if (delay > 0) {
Timer t = null;
t = new Timer(_ =>
{
Queue(action, urgent, 0);
t.Dispose();
}, null, delay, Timeout.Infinite);
return;
}
lock (threads) {
// we are spinning up too many threads
// should be fixed
if (maxThreads > threads.Count) {
Thread t = new Thread(new ThreadStart(ThreadProc));
t.IsBackground = true;
// dont affect the UI.
t.Priority = ThreadPriority.Lowest;
t.Name = "Worker thread for " + name;
t.Start();
threads.Add(t);
}
}
lock (actions) {
if (urgent) {
actions.Insert(0, action);
} else {
actions.Add(action);
}
Monitor.Pulse(actions);
}
}
private void ThreadProc() {
while (true) {
lock (threads) {
if (maxThreads < threads.Count) {
threads.Remove(Thread.CurrentThread);
break;
}
}
List<Action> copy;
lock (actions) {
while (actions.Count == 0) {
Monitor.Wait(actions);
}
copy = new List<Action>(actions);
actions.Clear();
}
foreach (var action in copy) {
action();
}
}
}
}
static Dictionary<string, ThreadPool> threadPool = new Dictionary<string, ThreadPool>();
public static Timer Every(int milliseconds, Action action) {
Timer timer = new Timer(_ => action(), null, 0, milliseconds);
return timer;
}
public static void SetMaxThreads(string uniqueId, int threads) {
GetThreadPool(uniqueId).SetMaxThreads(threads);
}
public static void Queue(string uniqueId, Action action) {
Queue(uniqueId, action, null);
}
public static void Queue(string uniqueId, Action action, int delay) {
Queue(uniqueId, action, null,false, delay);
}
public static void Queue(string uniqueId, Action action, Action done) {
Queue(uniqueId, action, done, false);
}
public static void Queue(string uniqueId, Action action, Action done, bool urgent) {
Queue(uniqueId, action, done, urgent, 0);
}
public static void Queue(string uniqueId, Action action, Action done, bool urgent, int delay) {
Debug.Assert(uniqueId != null);
Debug.Assert(action != null);
Action workItem = () =>
{
try {
action();
} catch (ThreadAbortException) { /* dont report on this, its normal */ } catch (Exception ex) {
Debug.Assert(false, "Async thread crashed! This must be fixed. " + ex.ToString());
Logger.ReportException("Async thread crashed! This must be fixed. ", ex);
}
if (done != null) done();
};
GetThreadPool(uniqueId).Queue(workItem, urgent, delay);
}
private static ThreadPool GetThreadPool(string uniqueId) {
ThreadPool currentPool;
lock (threadPool) {
if (!threadPool.TryGetValue(uniqueId, out currentPool)) {
currentPool = new ThreadPool(uniqueId);
threadPool[uniqueId] = currentPool;
}
}
return currentPool;
}
}
}
它有一个相当优雅的 API，我想添加的唯一功能是清理空线程池。
用法：
// Set the threads for custom thread pool
Async.SetMaxThreads("Queue Name", 10);
// Perform an action on the custom threadpool named: "Queue Name", when done call ImDone
Async.Queue("Queue Name", () => DoSomeThing(foo), () => ImDone(foo));
这有一些方便的重载允许您对延迟的操作进行排队，还有一个用于推送跳到队列前面的紧急作业。
【讨论】：
这种情况大声而清晰地尖叫BlockingCollection。创建一个专用线程，以适当设置其优先级来监视队列。当队列中没有项目时，BlockingCollection.Take 方法会自动阻塞。
public class Example
{
private BlockingCollection<WorkItem> m_Queue = new BlockingCollection<WorkItem>();
public event EventHandler<WorkItemEventArgs> WorkItemCompleted;
public Example()
{
var thread = new Thread(
() =>
{
while (true)
{
WorkItem item = m_Queue.Take();
// Add code to process the work item here.
if (WorkItemCompleted != null)
{
WorkItemCompleted(this, new WorkItemEventArgs(item));
}
}
});
thread.IsBackground = true;
thread.Priority = ThreadPriority.BelowNormal;
thread.Start();
}
public void Add(WorkItem item)
{
m_Queue.Add(item);
}
}
【讨论】：
线程池听起来就是这样。其实可以通过设置进程优先级来改变.NET自带线程池的优先级。将进程优先级降低一个档次，将你的 UI 提高一个档次，你应该有一个正常优先级的 UI 和较低优先级的线程池。
【讨论】：