利用System.Thread.Timer优化Active Object模式执行效率[zz]

http://zhenyulu.cnblogs.com/articles/27432.html
源代码下载：未经改造的ActiveObject；利用Timer对象改造后的ActiveObject

关于ActiveOjbect模式，第一次是在Robert C. Martin的《敏捷软件开发－原则、模式与实践》一书中看到的，后来找到了ActiveObject的出处：Lavender的《Active Object An Object Behavioral Pattern for Concurrent Programming》，才发现Active Object模式比我想象中的还要复杂，功能还要强大。它模拟实现了一种异步的、多线程的控制模式，为许多工业系统提供了一个简单的多任务核心。

《敏捷软件开发》中给出了一个Java实现的例子，从一个侧面表现了ActiveObject的使用方法。该例子通过ActiveObject引擎对命令队列中的命令进行循环处理，实现了命令的延时调用。其关键算法可以描述如下：

1、构造一延时命令，该延时命令将包含一任务并记录开始时间
2、将该延时命令放入队列。
3、从队列中取出一命令，若是延时命令，则判断是否到时，若尚未到时，则转向第2步。若到时，则释放出包含的任务到队列中。
4、判断是否是循环延时任务，如果是，则重新记录开始时间，并转向第 2 步。
5、转向第 3 步。

我将其改为了C#代码，关键代码如下：

public   void  Execute()
{
  DateTime currentTime = DateTime.Now;
  TimeSpan ts = currentTime - startTime;
  if(!started)
  {
    started = true;
    startTime = currentTime;
    ActiveObjectEngine.addCommand(this);
  }
  else if(ts.TotalMilliseconds < sleepTime)
  {
    ActiveObjectEngine.addCommand(this);
  }
  else
  {
    ActiveObjectEngine.addCommand(InnerTask);
    if(repeat)
    {
      this.startTime = DateTime.Now;
      ActiveObjectEngine.addCommand(this);
    }
  }
}

下面是用Together生成的UML图：

整个C#源代码可以从这里下载（.net 1.1下调试通过）

在该代码的应用中发现，一旦使用上该模式，CPU的占用率就达到了100％。也难怪，在任务比较轻的情况下，ActiveObject引擎绝大多数的时间都是在做一些无用功：从队列取出Command，看看是否到时，没有的话再把它放回到队列中。这些无用功占用了大量的CPU时间。即使创建一个新线程运行引擎，将线程的优先级别设置为ThreadPriority.BelowNormal，也避免不了CPU占用100％的问题。只不过此时用户不会感受到有任何迟滞而已（因为线程的优先级是BelowNormal）。

至此，能否更为有效的使用ActiveObject模式成了项目需要解决的一个关键问题。Windows中的Timer控件给了我一些启发：Timer控件可以定时触发事件，但从来不占用大量的CPU时间。如果简单的周期性任务，使用Timer非常简便（例如：每秒更新一下窗口上时钟的显示），但面对动态增加的，变化的周期性任务，使用Timer控件就显得有些麻烦（例如：每秒更新时间显示、每1.7秒读取一个数据、每2.3秒发送一条消息....），我们很容易想象到通过动态添加删除Timer控件来达到上述目的所面临的麻烦。如果能够将ActiveObject模式与Timer的优点结合起来，就可以构建一个高效、灵活的ActiveObject实现方案。下面是我的实现方案：

1、构造一Timer对象
2、遍历队列，执行已经到时的任务并计算距离最近一次事件触发还需多少时间
3、根据步骤 2 的结果设置Timer对象的下一次触发时间
4、Timer对象触发事件 2

通过将整个过程置于一个单独线程中执行，并在ActiveObject引擎空闲的时候，用WaitOne方法归还CPU占用以达到降低CPU占用率的问题。关键代码如下：

private   static   bool  Stoped  =   true ;
private   static   readonly  TimeSpan MaxTimeSpan  =  TimeSpan.FromMilliseconds( int .MaxValue);

private   static  Thread thread;
private   static  ArrayList itsCommands  =   new  ArrayList();
private   static  TimerCallback timerDelegate  =   new  TimerCallback(ActiveObjectEngine.CallOnTime);

private   static  ManualResetEvent manualEvent  =   new  ManualResetEvent( false );
private   static  Timer stateTimer  =   new  Timer(timerDelegate, manualEvent, System.Threading.Timeout.Infinite, System.Threading.Timeout.Infinite);

// =====================================
//  由定时器触发的方法
// =====================================
private   static   void  CallOnTime(Object stateInfo)
{
  ((ManualResetEvent)stateInfo).Set();
}

// =====================================
//  启动ActiveObject引擎
// =====================================
public   static   void  Start()
{
  Stoped = false;

  thread = new Thread(new ThreadStart(ActiveObjectEngine.Run));
  thread.IsBackground = true;
  thread.Priority = ThreadPriority.BelowNormal;
  thread.Start();
}

// =====================================
//  线程入口点
// =====================================
private   static   void  Run()
{
  while(!Stoped)
  {
    stateTimer.Change(GetNextRunTime(), MaxTimeSpan);
    manualEvent.Reset();
    manualEvent.WaitOne();
  }
}

// =====================================
//  获取下一个需要触发事件的时间
// =====================================
private   static  TimeSpan GetNextRunTime()
{
  TimeSpan ts, nextRunTime = MaxTimeSpan;

  lock(itsCommands)
  {
    if(itsCommands.Count == 0)
    {
      Stop();
      return TimeSpan.FromMilliseconds(System.Threading.Timeout.Infinite);
    }

    foreach(ICommand cmd in itsCommands)
    {
      ts = cmd.Execute();

      if(!cmd.Active)
        itsCommands.Remove(cmd);
      else if(ts < nextRunTime)
        nextRunTime = ts;
    }
  }
  return nextRunTime;
}

// =====================================
//  向ActiveObject引擎中添加定时任务
// =====================================
public   static   void  addCommand(ICommand c)
{
  lock(itsCommands)
  {
    itsCommands.Add(c);
    stateTimer.Change(GetNextRunTime(), MaxTimeSpan);
  }  
}

// =====================================
//  停止ActiveObject引擎
// =====================================
public   static   void  Stop()
{
  Stoped = true;
  manualEvent.Set();
  stateTimer.Change(System.Threading.Timeout.Infinite, System.Threading.Timeout.Infinite);
  Thread.Sleep(500);
}

改造后的C#源代码可以从这里下载。修改后的ActiveObject模式UML图如下：

经过测试，改造后的ActiveObject引擎运行时CPU占用率急剧下降，在我的机器上平均CPU占用率只有不到2％，这大大提高了系统执行效率。

测试代码如下：

using  System;
using  System.Threading;
using  NUnit.Framework;
using  TimedSchedule;

namespace  TestTimedSchedule
{

  [TestFixture] 
  public class TimedSchedule_Test
  {

    ICommand task1 = new TestCommand("Task 1");
    ICommand task2 = new TestCommand("Task 2");
    ICommand task3 = new TestCommand("Task 3");
    ICommand task4 = new TestCommand("Task 4");
    ICommand task5 = new TestCommand("Task 5");
    ICommand task6 = new TestCommand("Task 6");

    ActiveObjectEngine engine = new ActiveObjectEngine();

    [Test]
    public void TestScheduledCommand() 
    {
      Console.WriteLine(" ============= Test ScheduledCommand =============");
      ScheduledCommand sc1 = new ScheduledCommand(TimeSpan.FromMilliseconds(500), task1, true);
      ScheduledCommand sc2 = new ScheduledCommand(TimeSpan.FromMilliseconds(700), task2, true);
      ScheduledCommand sc3 = new ScheduledCommand(TimeSpan.FromMilliseconds(1100), task3, true);
      ScheduledCommand sc4 = new ScheduledCommand(TimeSpan.FromMilliseconds(900), task4, true);
      ScheduledCommand sc5 = new ScheduledCommand(TimeSpan.FromMilliseconds(1800), task5, true);
      ScheduledCommand sc6 = new ScheduledCommand(TimeSpan.FromMilliseconds(2100), task6, true);

      ActiveObjectEngine.addCommand(sc1);
      ActiveObjectEngine.addCommand(sc2);
      ActiveObjectEngine.addCommand(sc3);
      ActiveObjectEngine.addCommand(sc4);
      ActiveObjectEngine.addCommand(sc5);
      ActiveObjectEngine.addCommand(sc6);

      ActiveObjectEngine.Start();
      Thread.Sleep(5000);
      ActiveObjectEngine.Stop();

      Console.WriteLine(" ======================================== ");

      ActiveObjectEngine.Start();
      Thread.Sleep(5000);
      ActiveObjectEngine.Terminate();
    }
  }
}