线程管理示例代码

【核心代码】

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



 /*
     How to use Thread Classs
     * 
     * ==============
     * public ELMService()
        {
            InitializeComponent();
            etm.ClalThreadPool("EmailThreads", (uint)ApplicationInfo.EmailParsingThreads);
        } 
     * //queue items in thread for processing.
     * etm.QueueUserEmailWorkItem(new WaitCallback(ELMService_UpdateLogHandler), objSupplier);
       public void ELMService_UpdateLogHandler(object objmsg)
     * {
     *  return;
     * }
     * 
     */

namespace WorkflowManagement
{

    public delegate void ThreadActivationCallback();
    public delegate void JobFinishCallback(object state);
    public delegate bool OnErrorCallback(Exception exc);

    public class ThreadManager
    {
        
        public event ThreadStart ThreadStarted;
        public event ThreadStart ThreadFinished;
        public event WaitCallback JobStarted;
        public event WaitCallback JobFinished;
        public event OnErrorCallback OnError;
        public event ThreadStart Idle;
        public uint MaxThreads;
        internal int activeThreads = 0;
        internal Stack workItems;
        internal int threadTimeout = 5000; // 5 seconds
        private Timer timer;
        public string Name;

        public void ClalThreadPool(string name, uint maxThreads)
        {
            Name = name;
            MaxThreads = maxThreads;
            workItems = Stack.Synchronized(new Stack((int)(10 * maxThreads)));
            timer = new Timer(new TimerCallback(OnTimer), null, 0, 2000);
        }

        public bool QueueUserWorkItem(WaitCallback callback, object state)
        {
            if (callback == null)
                return false;
            lock (this)
            {
                workItems.Push(new WorkItem(callback, state));
                if (activeThreads == 0)
                    WorkThread.AddWorker(this);
                return true;
            }

        }

        public void OnTimer(object state)
        {
            lock (this)
            {
                if (workItems.Count == 0)
                    return;

                if (activeThreads < MaxThreads)
                    WorkThread.AddWorker(this);
            }

        }

        internal void ThreadStartedEvent()
        {
            if (ThreadStarted != null)
                ThreadStarted();
        }

        internal void ThreadFinishedEvent()
        {
            if (ThreadFinished != null)
                ThreadFinished();
        }

        internal void JobStartedEvent(object parameter)
        {
            if (JobStarted != null)
                JobStarted(parameter);
        }

        internal void JobFinishedEvent(object parameter)
        {
            if (JobFinished != null)
                JobFinished(parameter);
        }

        internal void IdleEvent()
        {
            if (Idle != null)
                Idle();
        }

        internal bool OnErrorEvent(Exception exc)
        {
            if (OnError != null)
                return OnError(exc);
            return false;

        }

        public void Stop()
        {
            lock (this)
            {
                timer.Change(Timeout.Infinite, Timeout.Infinite);
                workItems.Clear();
                MaxThreads = 0;
            }

        }

    }

    internal class WorkItem
    {
        public object parameter;
        public WaitCallback callback;
        public WorkItem(WaitCallback callback, object state)
        {
            this.parameter = state;
            this.callback = callback;
        }

    }

    internal class WorkThread
    {
        ThreadManager pool;
        internal WorkThread(ThreadManager pool)
        {
            this.pool = pool;
        }

        static public void AddWorker(ThreadManager pool)
        {
            Interlocked.Increment(ref pool.activeThreads);
            WorkThread worker = new WorkThread(pool);
            Thread thread = new Thread(new ThreadStart(worker.Loop));
            thread.SetApartmentState(ApartmentState.STA);
            thread.Start();
        }

        private void Loop()
        {
            try
            {
                pool.ThreadStartedEvent();

                while (true)
                {
                    WorkItem item = null;
                    try
                    {
                        item = (WorkItem)pool.workItems.Pop();
                    }

                    catch
                    {
                        Thread.Sleep(pool.threadTimeout);
                        item = (WorkItem)pool.workItems.Pop();
                        if (item == null)
                            break;
                    }
                    try
                    {
                        pool.JobStartedEvent(item.parameter);
                        /*
                         Write your functions you want to process  
                         * Here we cast the item.parameter into the 
                         * ProcessingSteps and get the type to run
                         * According to the type related functions
                         * will be called.
                         * E.g. if it is CreateProcess then 
                         * CreatePrcess of ProcessEngine will be called.
                         * Same way functions called for each parameter.
                         * 
                        */
                        //item.callback(item.parameter);
                        // Following is my function and it is called by passing parameter.
                        //ProcessingManager.Process((Processingstep)item.parameter);
                        pool.JobFinishedEvent(item.parameter);
                    }

                    catch (Exception exc)
                    {
                        try
                        {
                            EngineLog.UpdateLogLine(exc.ToString());
                            pool.OnErrorEvent(exc);
                        }

                        catch (Exception critical)
                        {
                            // Check for critical exception -- pseudo code
                            //if (critical is CRITICAL)
                            throw critical;
                        }

                    }
                    
                    if (pool.activeThreads > pool.MaxThreads)
                        return;
                }

            }

            catch { }
            finally
            {
                Interlocked.Decrement(ref pool.activeThreads);
            }
            pool.ThreadFinishedEvent();
            if (pool.activeThreads == 0)
                pool.IdleEvent();

        }

    }
}