我很感兴趣了解您对以下规则下哪种线程架构更好的想法:
一个线程必须在应用程序的生命周期内运行,在队列中没有要执行的任务时处于睡眠/等待模式。
线程必须具有BelowNormal优先级(这消除了使用ThreadPool的可能性)。
线程必须在完成任务后向主线程提供反馈。
线程将监视Queue<T>以获取要执行的更多作业。
我正在使用.NET Framework 4.0。
请让我知道您的想法 :)
我很感兴趣了解您对以下规则下哪种线程架构更好的想法:
一个线程必须在应用程序的生命周期内运行,在队列中没有要执行的任务时处于睡眠/等待模式。
线程必须具有BelowNormal优先级(这消除了使用ThreadPool的可能性)。
线程必须在完成任务后向主线程提供反馈。
线程将监视Queue<T>以获取要执行的更多作业。
我正在使用.NET Framework 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);
}
}
- 线程必须具有BelowNormal优先级(这消除了使用ThreadPool的可能性)。
这似乎是使用TPL和ThreadPool的主要难点。您确定您没有高估较低优先级的有用性吗?
您将不得不投入大量的工作才能想出一些始终比TPL弱得多(且测试/可靠性都较差)的东西。
建议您重新考虑此事。
个人而言,我通常会自己编写代码,因为我喜欢更严格的控制。
我在媒体浏览器中使用这个:
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.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);
}
}
通过阅读上述条件
一些问题
1- 是否有其他线程将作业填充到队列< T >中?
如果答案是肯定的,那么可以在此处使用生产者/消费者设计模式,我不了解 .net 4.0,但这种设计可以在 .net 3.5 中实现。
请参见此处的示例。