事件溯源

示例：跟踪船舶（C#）

这是一个非常简单的事件溯源示例，用于了解基本思想。在这个例子中，我故意保持极度简单，作为起点——然后我将使用其他例子来探索一些更复杂的问题。

领域模型是一个简单的模型，它包含运载货物并在港口之间移动的船舶。

有四种类型的事件会影响模型

• 到达：船舶到达港口
• 离开：船舶离开港口
• 装载：货物装载到船上
• 卸载：货物从船上卸载

让我们举一个简单的例子，将船舶移动到周围。

class Tester...

  Ship kr;
  Port sfo, la, yyv;
  Cargo refact;
  EventProcessor eProc;

  [SetUp]
  public void SetUp() {
    eProc = new EventProcessor(); 
    refact = new Cargo ("Refactoring");
    kr = new Ship("King Roy");
    sfo = new Port("San Francisco", Country.US);
    la = new Port("Los Angeles", Country.US);
    yyv = new Port("Vancouver", Country.CANADA) ;
  }

[Test]
public void ArrivalSetsShipsLocation() {
  ArrivalEvent ev = new ArrivalEvent(new DateTime(2005,11,1), sfo, kr);
  eProc.Process(ev);
  Assert.AreEqual(sfo, kr.Port);
}
[Test]
public void DeparturePutsShipOutToSea() {
  eProc.Process(new ArrivalEvent(new DateTime(2005,10,1), la, kr));
  eProc.Process(new ArrivalEvent(new DateTime(2005,11,1), sfo, kr));
  eProc.Process(new DepartureEvent(new DateTime(2005,11,1), sfo, kr));
  Assert.AreEqual(Port.AT_SEA, kr.Port);    
}

为了使这些测试能够正常工作，我们只需要到达和离开事件。事件处理器非常简单。

class EventProcessor...

  IList log = new ArrayList();  
  public void Process(DomainEvent e) {
    e.Process();
    log.Add(e);
  }

每个事件都有一个处理方法。

class DomainEvent...

  DateTime _recorded, _occurred;
  internal DomainEvent (DateTime occurred) {
    this._occurred = occurred;
    this._recorded = DateTime.Now;
  }
  abstract internal void Process();

到达事件只是捕获数据，并有一个处理方法，该方法只是将事件转发到相应的域对象。

class DepartureEvent...

  Port _port;
  Ship _ship;
  internal Port Port  {get { return _port; }}
  internal Ship Ship  {get { return _ship; }}
  internal DepartureEvent(DateTime time, Port port, Ship ship) : base (time)  {
    this._port = port;
    this._ship = ship;
  }
  internal override void Process() {
    Ship.HandleDeparture(this);
  }

因此，在这里事件只执行处理选择逻辑。处理域逻辑由船舶完成。

class Ship...

  public Port Port;
  public void HandleDeparture(DepartureEvent ev) {
    Port = Port.AT_SEA;
  }

离开事件只是将船舶的港口设置为特殊情况。您会注意到我将事件传递到域对象中。这里有一个选择，即事件是否应该只传递域对象处理所需的數據，还是传递事件本身。通过传递事件，事件不需要知道域逻辑需要哪些数据。如果事件后来获得了更多数据，则无需更新签名。传递事件的缺点是域逻辑现在意识到事件本身。

这个简单的测试只是展示了基本的事件处理是如何工作的。现在我将展示一些域逻辑，看看它是如何工作的。我们的船只运输书籍，以满足我拥有整个航运公司将我的书籍运往世界各地的幻想。正如您所知，通过加拿大运送书籍非常危险，因为这样做会使书籍有被大量“eh”污染的风险。我见过一些书，几乎每句话都有一个 eh（较长的句子可以得到两个或三个）。

因此，我的完美书籍运输系统能够检测货物是否经过加拿大。

class Tester...

  [Test]
  public void VisitingCanadaMarksCargo() {
    eProc.Process(new LoadEvent(new DateTime(2005,11,1), refact, kr));
    eProc.Process(new ArrivalEvent(new DateTime(2005,11,2), yyv, kr));
    eProc.Process(new DepartureEvent(new DateTime(2005,11,3), yyv, kr ));
    eProc.Process(new ArrivalEvent(new DateTime(2005,11,4), sfo, kr));
    eProc.Process(new UnloadEvent(new DateTime(2005,11,5), refact, kr));
    Assert.IsTrue(refact.HasBeenInCanada);
  }

由于货物可能在船舶之间移动和卸载，因此货物有责任知道它是否暴露于这些北方的危险之中。幸运的是，风险只发生在实际停泊在港口时，只是在水域中是安全的。因此，我们的到达事件必须跟踪这一点。

class ArrivalEvent...

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

处理程序再次是 Ship 对象。

class Ship...

  IList cargo;
  public void HandleArrival (ArrivalEvent ev) {
    Port = ev.Port;
    foreach (Cargo c in cargo) c.HandleArrival(ev);
  }

船舶不负责跟踪加拿大的访问，因此它将到达通知传递给货物。

class Cargo...

  public bool HasBeenInCanada = false;
  public void HandleArrival(ArrivalEvent ev) {
    if (Country.CANADA == ev.Port.Country) 
      HasBeenInCanada = true;
  }

考虑事件处理方法包含域逻辑的另一种情况 - 当域逻辑变得更加复杂时，它需要了解大量域模型。在这种方法中，域对象将事件传递给相关对象，以便它们可以处理事件以执行它们需要做的事情。

示例：更新外部系统 (C#)

事件溯源的一个重要功能是您可以根据需要重新处理事件。但是，如果处理事件会导致与外部系统的交互，那么这是一件坏事。最好的结果是他们会厌倦你所有的垃圾邮件事件。

一种简单的解决方法是确保您的系统通过网关调用外部系统，这些网关可以配置为确保除非您正在“真正”处理事件，否则不会发送任何消息。

我将用一个简单的例子来说明这一点，使用船舶和港口（这次没有货物）。假设每当船舶进入港口时，它必须通知当地海关当局。我们可以在事件处理域逻辑中实现这一点。

class Port...

  public void HandleArrival (ArrivalEvent ev) {
    ev.Ship.Port = this;
    Registry.CustomsNotificationGateway.Notify(ev.Occurred, ev.Ship, ev.Port);
  }

请注意，此代码只是在网关对象上调用通知，它不关心这是真正的处理还是某种重放。这里的一般原则是域逻辑不应该关心事件运行的上下文。

网关负责确定是否实际发送消息。由于这种情况非常简单，它只是通过链接到事件处理器并检查处理器是否处于活动状态来实现这一点。

class CustomsEventGateway...

  EventProcessor processor;
  public void Notify (DateTime arrivalDate, Ship ship, Port port) {
    if (processor.isActive) 
      SendToCustoms(BuildArrivalMessage(arrivalDate, ship, port));
  }

事件处理器在进行常规处理时只是使自己处于活动状态。

class EventProcessor...

  public void Process(DomainEvent e) {
    isActive = true;
    e.Process();
    isActive = false;
    log.Add(e);
  }

虽然这种情况非常简单，但基本原理是相同的。网关决定是否发送外部消息，而不是域逻辑。网关根据它们收集的有关处理上下文的的信息来决定这一点。在这种情况下，来自处理器的简单布尔状态就足够了。

示例：反转事件 (C#)

在这里，我们将以运输示例为例，看看如何反转事件。我们需要反转的关键是确保我们可以准确地计算出由于事件而发生状态更改的任何对象的先前状态。

存储此先前数据的最佳位置是在事件本身，这与示例中将事件传递给域对象的做法非常吻合。由于域对象拥有事件，因此它们可以轻松地在事件上存储信息。

加载事件提供了一个简单的示例。事件包含以下源数据。

class LoadEvent...

  int _shipCode;
  string _cargoCode;
  internal LoadEvent(DateTime occurred, string cargo, int ship) : base(occurred){
    this._shipCode = ship;
    this._cargoCode = cargo;
  }
  internal Ship Ship {get { return Ship.Find(_shipCode); }}
  internal Cargo Cargo {get { return Cargo.Find(_cargoCode); }}

处理被传递给货物对象，货物对象需要存储货物的先前位置的港口。

class LoadEvent...

  internal override void Process() {
    Cargo.HandleLoad(this);
  }

internal Port priorPort;

class Cargo...

  internal void HandleLoad(LoadEvent ev) {
    ev.priorPort = _port;
    _port = null;
    _ship = ev.Ship;
    _ship.HandleLoad(ev);
  }

为了反转事件，我们添加了一个反转方法，该方法镜像处理方法，在域对象上调用反转方法。

class LoadEvent...

  internal override void Reverse() {
    Cargo.ReverseLoad(this);
  }

class Cargo...

  public void ReverseLoad(LoadEvent ev) {
    _ship.ReverseLoad(ev);
    _ship = null;
    _port = ev.priorPort;
  }

在这种情况下，事件承担了一堆可变的先前数据，这些数据是其处理数据的一部分。在这种情况下，简单的字段就足够了。其他情况可能需要更复杂的数据结构。当货物处理到达事件时，它会跟踪它是否在加拿大 eh。它可以使用一个简单的布尔字段来做到这一点。为了在事件上存储先前值，需要比简单字段更多的东西，因为许多货物可能会受到到达的影响。因此，在这种情况下，我使用一个以货物为索引的映射。

class Cargo...

  public void HandleArrival(ArrivalEvent ev) {
    ev.priorCargoInCanada[this] = _hasBeenInCanada;
    if ("CA" == ev.Port.Country) 
      _hasBeenInCanada = true;
  }
  private bool _hasBeenInCanada = false;
  public bool HasBeenInCanada {get { return _hasBeenInCanada;}}

class ArrivalEvent...

  internal Port priorPort;
  internal IDictionary priorCargoInCanada = new Hashtable();

然后反转

class Cargo...

  public void ReverseArrival(ArrivalEvent ev) {
    _hasBeenInCanada = (bool) ev.priorCargoInCanada[this];
  }

这个例子很好地说明了事件上的源数据和错误处理如何影响我们进行反转的方式。对于加载事件，我们需要存储货物加载时的港口。如果事件在其源数据中包含了这一点，我们就不需要这样做。一些额外的源数据消除了添加先前数据的必要性。但这并不适用于所有地方：到达事件无法获取其货物的先前加拿大状态。

被认为是正确处理的内容也会产生影响。在这个系统中，我们有船舶的到达事件和离开事件。假设一切正常，我们应该始终获得这些交错的事件。因此，船舶可以通过将其港口字段设置为 Port.OUT_TO_SEA 来反转到达事件。当我们连续获得两个到达事件时会发生什么？为了反转这一点，我们需要在船舶上存储先前的港口。我们的另一种选择是将没有离开的第二个到达声明为错误，我们不需要进行此存储。

示例：外部查询 (C#)

即使对于基本的事件溯源，外部查询也很尴尬，因为如果您想重建应用程序状态，您需要使用过去进行的外部查询响应来完成此操作。

让我们想象一下，船舶必须在进入港口时确定其货物的价值。此估值由外部服务完成。如果船舶在 11 月 3 日进入港口，我立即处理事件，我将获得该货物在 11 月 3 日的价值。如果我在 12 月 5 日重建我的应用程序状态，我将希望获得相同的货物价值，即使它的价值在此期间发生了变化。

对于这个例子，我将展示一种处理方法，通过记录外部查询并使用它们在重新处理事件时提供值。在许多方面，它与您在外部更新中使用的方法类似 - 将交互转换为系统边界的事件，并使用事件记录来记住发生了什么。

使用我在这些示例中其他地方使用的类似于事件处理的风格，我们将事件传递给域对象进行处理。在这种情况下，货物启动对外部系统的调用并保存值。（我们假设我们将对这个值做一些有用的事情，但这与这个例子无关。）

class Cargo...

  public void HandleArrival(ArrivalEvent ev) {
    declaredValue = Registry.PricingGateway.GetPrice(this);
  }
  private Money declaredValue;

按照我的习惯，我通过一个我控制的网关封装所有外部系统访问。基本网关只会将调用转换为外部交互所需的任何内容。为了支持重放事件，我只是用一个类来包装它，该类记录这些查询。

图 6：使用日志记录网关包装网关以支持从事件正确重建。

日志记录网关检查每个调用以查看它是否有一个与之匹配的旧请求，如果没有，它将发出一个新请求。

class LoggedPricingGateway...

  public Money GetPrice(Cargo cargo) {
    GetPriceRequest oldReq = oldRequest(cargo);
    if (null != oldReq) return (Money) oldReq.Result; 
    else return newRequest(cargo);
  }

如果请求是新的，它会将请求转换为事件对象，调用外部查询，并将其存储在日志中。

class LoggedPricingGateway...

  private Money newRequest(Cargo cargo) {
    GetPriceRequest request = new GetPriceRequest(cargo);
    request.Result = gateway.GetPrice(cargo);
    log.Store(request);
    return (Money) request.Result;
  }

private class GetPriceRequest : QueryEvent {
  private Cargo cargo;
  public GetPriceRequest(Cargo cargo) : base() {
    this.cargo = cargo;
  }

class QueryEvent...

  DomainEvent _eventBeingProcessed;
  Object result;
  public QueryEvent() {
    _eventBeingProcessed = Registry.EventProcessor.CurrentEvent;
  }
  public object Result {
    get { return result; }
    set { result = value; }
  }
  public DomainEvent EventBeingProcessed {
    get { return _eventBeingProcessed; }
  }
}

因此，为了找到旧的请求，它会搜索其日志。

class LoggedPricingGateway...

  private GetPriceRequest oldRequest(Cargo cargo) {
    IList candidates =
      log.FindBy(EventProcessor.CurrentEvent, typeof (GetPriceRequest));
    foreach (GetPriceRequest request in candidates) {
      if (request.Cargo.RegistrationCode == cargo.RegistrationCode)
        return request;
    }
    return null;
  }

查询日志是通用的，因此我们可以使用已处理的域事件和请求类型发出查询以获取一些项目。这给了我们一个需要在网关特定方式进行进一步检查的小集合。

请求日志需要以与域事件日志相同的方式持久化，因为它需要重建应用程序状态。