深入线程，实现自定义的SynchronizationContext

(转载自:http://www.cnblogs.com/Kevin-moon/archive/2009/02/09/1384627.html)

首先说下关于一些线程的基础东西，这样有利于更好的理解。

---

 

1、STA、MTA和Apartments
    在16 bit Windows的时代，只有一个线程存在，所以一切都是那么的简单，可是，随着时代的进步，出现了Win32，这时出现了混乱，为了维持次序，Apartments出现了（Apartments是COM为了简化对象对多线程的支持而推出的一套机制，用于指定线程和COM对象的多线程特性，并且对不同特性的套间之间的调用提供同步支持，保证不同多线程特性的对象之间可以互相正确调用而不会引入同步问题），线程的执行必须在Apartments中，也就是线程对COM对象执行多线程调用，还是单线程调用。STA表示只支持单线程调用，MTA表示支持多线程调用。
    在一个进程中可以有多个STA单元，一个MTA单元。一个STA单元中只能有一个线程，而MTA单元中允许有多个线程。
    对于STA和MTA，我们把它用于标记线程和COM对象，这里需要注意的是如果COM对象标记为STA，那么该COM对象只能被STA的线程访问，我们可以在注册表中通过ThreadingModel来设定

属性值	含义
Main (缺省值)	主STA，也就是第一个创建的STA
Apartment	STA
Both	STA或者MTA都可以
Free	MTA
Neutral	NTA

    在Apartments之间，如果要跨Apartments之间进行通讯，那么访问的不是对象本身，而是对象的代理，如果是在Apartments内的通讯，访问的就是对象本身。通过下图更清楚点：

    在NET中，除非将[STAThreadAttribute]应用于入口点过程，否则主应用程序线程会被初始化为 MTA。我们创建的线程和ThreadPool都是MTA的，如果想要去创建STA线程，需要通过Thread的SetApartmentState来实现，并且要在线程启动之前设置，否则不允许执行(后面的自定义SynchronizationContext就要用到这个)。
    同时，这里有个需要我们注意的，NET中的主线程就是UI线程（STA），在入口处创建。所有的控件都是运行在这个线程上，也只能在这个线程上操控,这点我们可以通过了解Control的BeginInvoke方法来认识，BeginInvoke是实现异步机制的，也就是会创建一个线程（MTA），这时如果你想在BeginInvoke的委托方法中去修改UI控件，你就必须通过invoke来实现，因为invoke把修改的动作带回了UI线程。

2、WaitHandle类
   在多线程的时代，共享资源是大家都争抢的东西，为了不出现混乱的局面，NET给了我们WaitHandle。
   不过单一的WaitHandle有点单调，于是NET又给了我们Mutex、Semaphore、EventWaitHandler。
         Mutex 是同步基元，它只向一个线程授予对共享资源的独占访问权。如果一个线程获取了互斥体，则要获取该互斥体的第二个线程将被挂起，直到第一个线程释放该互斥体。线程通过WaitOne获取互斥体的访问权，通过ReleaseMutex来释放。
   Semaphore：通过信号量限制线程对共享资源的访问，信号量的计数在每次线程进入信号量时减小，在线程释放信号量时增加。当计数为零时，后面的请求将被阻塞，直到有其他线程释放信号量。当所有的线程都已释放信号量时，计数达到创建信号量时所指定的最大值。线程通过WaitOne进入信号量，通过Release退出信号量。(后面的自定义SynchronizationContext就要用到这个)。
    Mutex和Semaphore都可以分为局部和全局的，局部的只能在进程中使用，而全局的可以跨进程使用，对于全局的我们只需要在构造的时候指定一个名称就可以了。
    EventWaitHandler：允许线程通过发信号互相通信。通常，一个或多个线程在 EventWaitHandle 上阻止，直到一个未阻止的线程调用 Set方法，以释放一个或多个被阻止的线程。继承它的类有AutoResetEvent和ManualRestEvent。(后面自定义SynchronizationContext就要用到这个)。

呵呵，这些东西都理解了嘛！了解这些后，看下面的东西会变的简单的多。

StaSynchronizationContext(自定义的SynchronizationContext)

---

   StaSynchronizationContext是STA线程的上下文环境，我们可以向该上下文环境发送消息，该上下文环境中的STA线程负责监听，如果发现有消息进入，则执行消息。 
   对于这个需求，核心就是如何实现监听？！

 
1、队列BlockingQueue
     因为需要向该上下文环境发送消息，所以首先我们必须先建立一个结构来接收消息，同时该结构必须要线程安全、保持先进先出的顺序，并且还需要建立一个机制，如果在该结构中没有任何消息，那么这时调用Dequeue，该线程要一直等待，直到有消息进入或者手工释放该线程。

internal interface IQueueReader<t> : IDisposable
{
  T Dequeue();
  void ReleaseReader();
}
internal interface IQueueWriter<t> : IDisposable
{
  void Enqueue(T data);
}

internal class BlockingQueue<t> : IQueueReader<t>, 
                                     IQueueWriter<t>, IDisposable
{
  private Queue<t> mQueue = new Queue<t>();
  // 初始化信号量，并且设置信号量的计数为0
  private Semaphore mSemaphore = new Semaphore(0, int.MaxValue);
  // 创建EventWaitHandler，并设置为非终止状态
  private ManualResetEvent mKillThread = new ManualResetEvent(false);
  private WaitHandle[] mWaitHandles;

  public BlockingQueue()
  {
    mWaitHandles = new WaitHandle[2] { mSemaphore, mKillThread };
  }
  public void Enqueue(T data)
  {
    lock (mQueue) mQueue.Enqueue(data);
    
    //释放信号量，意味着其它线程可以访问该队列
    mSemaphore.Release();
  }

  public T Dequeue()
  {
    //执行该方法的线程处于等待状态，直到mWaitHandles接受到信号
    WaitHandle.WaitAny(mWaitHandles);
    lock (mQueue)
    {
      if (mQueue.Count > 0)
         return mQueue.Dequeue();
    }
    return default(T);
  }

  public void ReleaseReader()
  {
    //释放被阻止的线程
    mKillThread.Set();
  }

  void IDisposable.Dispose()
  {
    if (mSemaphore != null)
    {
      mSemaphore.Close();
      mQueue.Clear();
      mSemaphore = null;
    }
  }
}

2、SendOrPostCallbackItem
     完成了消息结构，接下来就来实现消息对象了，它包含执行动作的委托实例、执行动作的方式(Send和Post)、事件信号。
     向上下文环境发送消息可以有异步(Post)和同步(Send)两种方式，消息对象中的执行动作的方式就对应这两种方式，事件信号也是为了实现这个机制所必须的属性，具体的看后面的类

internal enum ExecutionType
{
   Post,
   Send
}

internal class SendOrPostCallbackItem
{
   object mState;
   private ExecutionType mExeType;
   SendOrPostCallback mMethod;
   ManualResetEvent mAsyncWaitHandle = new ManualResetEvent(false);
   Exception mException = null;

   internal SendOrPostCallbackItem(SendOrPostCallback callback,
      object state, ExecutionType type)
   {
      mMethod = callback;
      mState = state;
      mExeType = type;
   }

   internal Exception Exception
   {
      get { return mException; }
   }

   internal bool ExecutedWithException
   {
      get { return mException != null; }
   }

   internal void Execute()
   {
      if (mExeType == ExecutionType.Send)
         Send();
      else
         Post();
   }

   internal void Send()
   {
      try
      {
         mMethod(mState);
      }
      catch (Exception e)
      {
         mException = e;
      }
      finally
      {
         mAsyncWaitHandle.Set();
      }
   }

   internal void Post()
   {
      mMethod(mState);
   }

   internal WaitHandle ExecutionCompleteWaitHandle
   {
      get { return mAsyncWaitHandle; }
   }
}

3、上下文环境中的STA线程类
     该类负责的就是监听队列中是否有消息进入，如果有消息进入，则执行消息动作，否则一直等待，同时负责对STA线程的管理（启动，终止....）
     特别说明的地方：
     1、设置STA线程是通过"mStaThread.SetApartmentState(ApartmentState.STA)"实现的。
     2、让该线程一直执行监听的方法是"While(true)"，只有当调用Stop的时候，该监听才停止。

internal class StaThread
{
   private Thread mStaThread;
   private IQueueReader<sendorpostcallbackitem> mQueueConsumer;
   private ManualResetEvent mStopEvent = new ManualResetEvent(false);

   internal StaThread(IQueueReader<sendorpostcallbackitem> reader)
   {
      mQueueConsumer = reader;
      mStaThread = new Thread(Run);
      mStaThread.Name = "STA Worker Thread";
      mStaThread.SetApartmentState(ApartmentState.STA);
   }

   internal void Start()
   {
      mStaThread.Start();
   }

   internal void Join()
   {
      mStaThread.Join();
   }

   private void Run()
   {
      while (true)
      {
         bool stop = mStopEvent.WaitOne(0);
         if (stop)
         {
            break;
         }

        SendOrPostCallbackItem workItem = mQueueConsumer.Dequeue();
        if (workItem != null)
            workItem.Execute();
      }
   }

   internal void Stop()
   {
      mStopEvent.Set();
      mQueueConsumer.ReleaseReader();
      mStaThread.Join();
      mQueueConsumer.Dispose();
   }
}

4、上下文的环境类StaSynchronizationContext
     该类继承于System.Threading.SynchronizationContext，类中包含了StaThread对象和队列。
     需要特别说明的地方：
     1、上下文环境中的STA线程是随着该上下文环境对象的创建而启动的
     2、Send方法会等待SendOrPostCallback委托执行完成后，才运行下面的代码，也就是说这是同步的执行
          通过"item.ExecutionCompleteWaitHandle.WaitOne()"这句话来实现的
     3、Post方法和Send方法相反，是异步执行的

public class StaSynchronizationContext : SynchronizationContext, IDisposable
{
   private BlockingQueue<sendorpostcallbackitem > mQueue;
   private StaThread mStaThread;
   public StaSynchronizationContext()
      : base()
   {
      mQueue = new BlockingQueue<sendorpostcallbackitem>();
      mStaThread = new StaThread(mQueue);
      mStaThread.Start();
   }

   public override void Send(SendOrPostCallback d, object state)
   {
      SendOrPostCallbackItem item = new SendOrPostCallbackItem(d, state, 
                                        ExecutionType.Send);
      mQueue.Enqueue(item);
      item.ExecutionCompleteWaitHandle.WaitOne();

      if (item.ExecutedWithException)
         throw item.Exception;
   }

   public override void Post(SendOrPostCallback d, object state)
   {
      SendOrPostCallbackItem item = new SendOrPostCallbackItem(d, state, 
                                        ExecutionType.Post);
      mQueue.Enqueue(item);
   }

   public void Dispose()
   {
      mStaThread.Stop();
   }

   public override SynchronizationContext CreateCopy()
   {
      return this;
   }
}

OK，到这里StaSynchronizationContext就完成了，接下来我们做个DEMO来运行下吧！

public class Params
{
  public string Output {get; set;}
  public int CallCounter { get; set; }
  public int OriginalThread { get; set; }
}

class Program
{
  private static int mCount = 0;
  private static StaSynchronizationContext mStaSyncContext = null;
  static void Main(string[] args)
  {
     mStaSyncContext = new StaSynchronizationContext();
     for (int i = 0; i < 100; i++)
     {
        ThreadPool.QueueUserWorkItem(NonStaThread);
     }
     Console.WriteLine("Processing");
     Console.WriteLine("Press any key to dispose SyncContext");
     Console.ReadLine();
     mStaSyncContext.Dispose();
  }

  private static void NonStaThread(object state)
  {
     int id = Thread.CurrentThread.ManagedThreadId;

     for (int i = 0; i < 10; i++)
     {
        var param = new Params { OriginalThread = id, CallCounter = i };
        mStaSyncContext.Send(RunOnStaThread, param);
        Debug.Assert(param.Output == "Processed", "Unexpected behavior by STA thread");
     }
  }

  private static void RunOnStaThread(object state)
  {
     mCount++;
     Console.WriteLine(mCount);
     int id = Thread.CurrentThread.ManagedThreadId;
     var args = (Params)state;
     Trace.WriteLine("STA id " + id + " original thread " + 
                     args.OriginalThread + " call count " + args.CallCounter);
     args.Output = "Processed";

  }
}

运行结果：

STA id 11 original thread 7 call count 0
STA id 11 original thread 12 call count 0
STA id 11 original thread 7 call count 1
STA id 11 original thread 12 call count 1
STA id 11 original thread 7 call count 2
STA id 11 original thread 12 call count 2
STA id 11 original thread 7 call count 3
STA id 11 original thread 12 call count 3
STA id 11 original thread 7 call count 4
STA id 11 original thread 12 call count 4
STA id 11 original thread 7 call count 5
STA id 11 original thread 12 call count 5
STA id 11 original thread 7 call count 0
STA id 11 original thread 12 call count 0
STA id 11 original thread 7 call count 1
STA id 11 original thread 12 call count 1
STA id 11 original thread 7 call count 2
STA id 11 original thread 12 call count 2
STA id 11 original thread 7 call count 3
STA id 11 original thread 12 call count 3
STA id 11 original thread 7 call count 4
STA id 11 original thread 12 call count 4
STA id 11 original thread 7 call count 5
STA id 11 original thread 12 call count 5

最后说明一下：
     该StaSynchronizationContext中的线程并不一定要设置为STA，MTA也可以，在这里把线程设定为STA的目的：可以在该线程上，除了处理NET中的托管对象外，还可以处理一些非托管的对象（COM）。

写完了！
StaSynchronizationContext原代码：SynchronizationContext.rar