并行模型

示例：运输时间查询（C#）

我将此示例基于我为事件溯源开发的关于运输、货物和港口的简单示例。因此，在深入研究此示例之前，您应该熟悉该示例。域逻辑代码几乎相同。

图 1：我们运输示例的域对象。

由于并行模型的大部分复杂性都与处理临时并行模型和活动数据库有关，因此我在此示例中引入了数据库。我使用流行的 NHibernate 对象关系映射器来映射到数据库。我不会详细介绍映射，这并不十分有趣。相反，我将重点介绍它如何被存储库包装，以及如何在它和并行模型的存储库之间切换。

与基本示例一样，域模型的所有更改都由事件处理。当船舶抵达港口时，这将通过到达事件记录下来。

class ArrivalEvent...

  string _port;
  int _ship;  
  internal ArrivalEvent (DateTime occurred, string port, int ship) : base (occurred) {
    this._port = port;
    this._ship = ship;
  } 
  internal Port Port {get {return Port.Find(_port);}}
  internal Ship Ship {get {return Ship.Find(_ship);}}
  
  internal override void Process() {
    Ship.HandleArrival(this);
  }

与基本示例不同的是，在这种情况下，港口和船舶由简单的值作为标识符表示 - 在这种情况下对应于数据库中的主键，尽管任何键都可以。但是，属性使用实际的域对象。为了确保我们始终获得正确的域对象，我们使用在域对象类上定义的查找方法从数据库中提取它们。

class Port...

  public static Port Find(string s) {
    return (Port) Registry.DB.Find(typeof (Port), s);
  }

查找方法反过来委托给存储库。在我们的基本情况下，此存储库是使用 NHibernate 映射域对象的数据库。因此，存储库代码使用 NHibernate API 从数据库（或 NHibernate 的缓存）中提取对象。

class DB...

  public object Find(Type type, object key) {
    return _hibernateSession.Load(type, key);
  }

NHibernate 与大多数基于数据映射器的数据源一样，使用工作单元来跟踪它操作的对象。因此，您永远不会告诉对象将自己保存到数据库，而是提交工作单元，它会找出哪些对象在内存中发生了更改以及如何将它们写入。

对于此示例，我将在事件处理器中进行此事务包装，事件处理器的处理器现在看起来像这样。

class EventProcessor...

  public void Process(DomainEvent e) {
    Registry.DB.BeginTransaction();
    try {
      e.Process();
      InsertToLog(e);
      Registry.DB.Commit();
    } catch (Exception ex) {
      Registry.DB.Rollback();
      LogFailedEvent(e, ex);
    }
  }

这绝不是将工作单元包装的最佳位置，这将取决于应用程序的性质。但这对于我的示例来说已经足够了，并且也提出了问题 - 我们如何轻松地切换出持久存储库来处理临时查询？

让我们通过一个测试用例来了解这一点

class Tester...

  [Test] 
  public void TemporalQueryForShipsLocation() {
    eProc.Process(new ArrivalEvent(new DateTime(2005,11,2), la, kr));
    eProc.Process(new DepartureEvent(new DateTime(2005,11,5), la, kr ));
    eProc.Process(new ArrivalEvent(new DateTime(2005,11,6), sfo, kr));
    Assert.AreEqual(sfo, Ship.Find(kr).Port.Code);
    eProc.SetToEnd(new DateTime(2005,11,2));
    Assert.AreEqual(la, Ship.Find(kr).Port.Code);
  }

这里至关重要的方法是SetToEnd，它将我们的存储库更改为使用内存中存储库，并重新处理事件日志，以便事件播放到该天的最后一个事件。这为 11 月 2 日创建了我们的并行模型。

class EventProcessor...

  IList log;
  public void SetToEnd(DateTime date) {
    SynchronizeLog();
    IRepository temporalRepository = new MemoryDB();
    Registry.enableAlternateRepository(temporalRepository);
    foreach (DomainEvent e in log) {
      if (e.Occurred > date) return;
      e.Process();
    }
  }

为了运行时间查询，处理器完全脱离数据库。在这种情况下，处理器保留事件日志的副本。在脱离数据库之前，它会将其日志与持久日志同步，以便它拥有写入数据库的所有事件的完整记录。

日志同步后，我们创建一个完全在内存中的新存储库。这可以由嵌入式内存数据库支持，这将允许您继续使用 SQL。它也可以是满足存储库接口的手写内容。由于这是一个简单的示例，我只使用了一堆哈希表。

class MemoryDB...

  public object Find(Type type, object key) {
    object result = this[type][key];
    if (null == result) 
      throw new ApplicationException ("unable to find: " + key.ToString());
    return result;
  }
  private IDictionary store = new Hashtable();
  private IDictionary this[Type index] {get {return (IDictionary)store[index];}}

内存中存储库需要初始化为与数据库在处理任何事件之前相同的初始状态 - 在这种情况下，保存港口和船舶的参考数据。

class MemoryDB...

  public void Initialize() {
    store[typeof(Ship)] = new Hashtable();
    store[typeof(Cargo)] = new Hashtable();
    store[typeof(Port)] = new Hashtable();
    Insert(new Port("SFO", "San Francisco", "US"));
    Insert(new Port("LAX", "Los Angeles", "US"));
    Insert(new Port("YVR", "Vancouver", "CA"));
    Insert (new Port("XXX", "out to sea", "XX"));
    Insert(new Ship("King Roy", 1));
  }

设置好临时存储库后，我们告诉注册表开始使用它而不是真正的存储库。

class Registry...

  internal static void enableAlternateRepository(IRepository arg) {
    instance._repository = arg;
    arg.Initialize();
  }

现在，域模型中对注册表的任何调用都将使用内存中注册表。由于此设计将 NHibernate 放在常规数据库存储库中，这意味着我们根本不会在并行模型中使用 hibernate。对象出现在内存中，并由存储库保留在那里。

设置好内存中存储库后，我们从日志中按顺序处理发生在我们目标日期或之前的所有事件。完成后，我们拥有代表世界在请求日期结束时的状态的内存中存储库。我们现在对域对象执行的任何查询都将反映该日期。

要返回到正确的数据库，我们只需将临时内存中存储库换回常规存储库连接。

class Registry...

  internal static void restoreBaseRepository() {
    instance._repository = instance._baseRepository;
  }

示例：将并行模型与基本模型进行比较（C#）

对于并行模型的许多用途，我们一次只使用一个模型。一般处理（包括更新）使用基本模型完成，我们构建并行模型用于历史和假设查询。在任何时候，我们都只有一个并行模型在起作用。这种方案相对简单，避免了困扰许多科幻情节的跨平行宇宙身份问题。

但是，有时混合使用两者是有意义的。如果您还记得运输示例中乏味的幽默，您就会知道图书发行商对加拿大的致命恐惧以及用“eh”污染文本的风险。让我们假设，我们不时会遇到特别恶性的“eh”传染病。任何在具有这种传染病的港口的船舶中的货物都应被标记为高风险。当然，我们不会在当天发现这些事情，因此我们必须弄清楚当天港口里有什么。

这是一个表达这个问题的测试用例。

class Tester...

  [Test]
  public void HighRiskDayFlagsAllCargo() {
    eProc.Process(new RegisterCargoEvent(new DateTime(2005,1,1), "UML Distilled", "UML", "LAX" ));
    eProc.Process(new RegisterCargoEvent(new DateTime(2005,1,1), "Planning XP", "PXP", "LAX" ));
    eProc.Process(new RegisterCargoEvent(new DateTime(2005,1,1), "Analysis Patterns", "AP", "LAX" ));
    eProc.Process(new RegisterCargoEvent(new DateTime(2005,1,1), "P of EAA", "eaa", "LAX" ));
    eProc.Process(new ArrivalEvent(new DateTime(2005,11,2), la, kr));
    eProc.Process(new LoadEvent(new DateTime(2005,5,11),"PXP", 1));
    eProc.Process(new LoadEvent(new DateTime(2005,5,11),"AP", 1));
    eProc.Process(new ArrivalEvent(new DateTime(2005,11,9), yvr, kr));
    eProc.Process(new ArrivalEvent(new DateTime(2005,11,12), la, kr));
    eProc.Process(new ContagionEvent(new DateTime(2005,11,10), yvr));
    Assert.IsTrue(Cargo.Find("PXP").IsHighRisk, "PXP should be high risk");
    Assert.IsTrue(Cargo.Find("AP").IsHighRisk, "AP should be high risk");
    Assert.IsFalse(Cargo.Find("UML").IsHighRisk, "UML should NOT be high risk");
    Assert.IsFalse(Cargo.Find("eaa").IsHighRisk, "UML should NOT be high risk");
    Assert.IsFalse(Cargo.Find(refact).IsHighRisk, "UML should NOT be high risk");
  }

我用一个新的事件来描述传染病。

class ContagionEvent...

  internal class ContagionEvent : DomainEvent
  {
    string _portID;
    public ContagionEvent(DateTime occurred, string port) : base(occurred) {
      this._portID = port;
    }
    Port Port {get {return Port.Find(_portID);}}

internal override void Process() {
  Registry.EventProcessor.SetToEnd(Occurred);
  ArrayList infectedCargos = new ArrayList();
  foreach (Ship s in Port.Ships) infectedCargos.AddRange(s.Cargos);
  Registry.restoreBaseRepository();
  foreach (Cargo c in infectedCargos) {
    Cargo actualCargo = Cargo.Find(c.RegistrationCode);
    actualCargo.IsHighRisk = true;
  }
}

您会注意到，在这种情况下，我在事件的 process 方法中实际上有相当多的行为。这样做的原因是我决定让域模型不知道并行模型。因此，事件必须为传染病的日期创建临时并行模型，运行查询以找出哪些货物受到影响，将世界恢复到基本模型，并将更新传递出去。域模型仍然在每个并行模型中执行逻辑，尽管它并不多。

另一种方法（我会认真考虑）是为将货物标记为高风险创建新的事件。在这种情况下，传染病事件将在临时并行模型中找到受影响的货物，然后创建一个事件将这些货物标记为高风险。第二个事件将在基本状态下运行。当我写下这些内容时，我承认不确定我更喜欢哪种方法。