追溯事件

示例：使用回溯的追溯货物（C#）

我一直用一个基于在港口之间运输货物的示例来说明围绕事件溯源的大多数示例。我将继续使用该示例，这次看看如何使用追溯事件。

首先，我们需要对如何进行追溯事件做出一些战略决策。对于这个例子，我将使用通过回溯实时模型来达到分支点的方法。由于示例很简单，我不会使用任何选择性重播，而是直接回溯完整的事件列表。

由于我们使用的是回溯，因此所有事件都必须是可逆的。我不会详细介绍事件处理或逆转，请参阅事件溯源中的示例，了解如何实现。

追溯事件行为由一种特殊类型的事件触发。

class ReplacementEvent...

  private DomainEvent original, replacement;
  public DomainEvent Original {get { return original; }}
  public DomainEvent Replacement {get { return replacement; }}

  public ReplacementEvent(DomainEvent oldEvent, DomainEvent replacement)
    : base(oldEvent.Occurred) {
    this.original = oldEvent;
    this.replacement = replacement;
  }
  internal override void Process() {
    throw new Exception("Replacements should not be processed directly");
  }
  internal override void Reverse() {
    throw new Exception("Cannot reverse replacements");
  }

您会注意到，在这些情况下，我已经阻止了基本过程和逆转方法。在这些示例中，我大多数时候都更喜欢让事件处理自己的处理逻辑。替换不同：它们不包含任何领域逻辑，它们的处理涉及对事件队列的了解以及对该事件队列的操作。我更喜欢让事件处理器单独处理与事件队列的密切交互，因此实际上我在事件队列中构建了所有替换行为。

class EventProcessor...

  public void Process(DomainEvent ev) {
    try {
      if (ev is ReplacementEvent)
        ProcessReplacement((ReplacementEvent) ev);
      else if (OutOfOrder(ev))
        ProcessOutOfOrder(ev);
      else BasicProcessEvent(ev);
    } catch (Exception ex) {
      ev.ProcessingError = ex;
      if (ShouldRethrowExceptions) throw ex;
    }
    InsertToLog(ev);
  }

您会注意到，我在这里犯了面向对象编程的致命错误：基于方法参数类型的显式条件行为。我并不经常这样做，但在这里我这样做是因为我希望事件处理器将对队列的了解保留给自己。（我也可以使用双重分派，但这里的情况似乎并不复杂到需要它。）

您可能还会注意到，我这里没有针对拒绝事件的案例。正如我们将会看到的，我通过将这些事件视为替换事件来处理它们，其中替换事件为 null。

我将从最简单的案例开始。基本 Process（和 reverse）方法只是一个简单的包装器，允许我在需要时添加一些跟踪行为。

class EventProcessor...

  private void BasicProcessEvent(DomainEvent e) {
    traceProcess(e);
    e.Process();
  }
  private void BasicReverseEvent(DomainEvent e) {
    traceReverse(e);
    e.Reverse();
  }

乱序事件是最容易描述的。这些只是我们收到的顺序错误的常规事件，我们不使用替换事件来模拟它们。本质上，这些事件将一个新事件插入流中。处理器通过简单地将其与最后一个事件进行比较来测试乱序事件。

class EventProcessor...

  private bool OutOfOrder(DomainEvent e) {
    if (LogIsEmpty()) return false;
    return (LastEvent().after(e));
  }
  private DomainEvent LastEvent() {
    if (LogIsEmpty()) return null;
    return (DomainEvent) log[log.Count - 1];
  }
  private bool LogIsEmpty() {
    return 0 == log.Count;
  }

为了使这个例子变得非常简单，我只是根据发生的日期对事件进行排序。要真正做到这一点，可能只需将时间点的分辨率提高到更精细的程度。有时，其他因素也会影响排序。

处理乱序案例的概要很简单。

class EventProcessor...

  private void ProcessOutOfOrder(DomainEvent e) {
    RewindTo(e);
    BasicProcessEvent(e);
    ReplayAfter(e);
  }

要回溯，我只需从日志中选择所有晚于乱序事件的事件。同样为了简单起见，我将日志（或至少是它的缓存）保存在内存中。

class EventProcessor...

  private void RewindTo(DomainEvent priorEvent) {
    IList consequences = Consequences(priorEvent);
    for (int i = consequences.Count - 1; i >= 0; i--)
      BasicReverseEvent(((DomainEvent) consequences[i]));
  }
  private IList Consequences(DomainEvent baseEvent) {
    IList result = new ArrayList();
    foreach (DomainEvent candidate in log)
      if (candidate.IsConsequenceOf(baseEvent)) result.Add(candidate);
    return result;
  }

class DomainEvent...

  public bool IsConsequenceOf(DomainEvent other) {
    return (!ShouldIgnoreOnReplay && this.after(other));
  }

如您所见，并非所有事件在回溯期间都被选中进行重新处理，即使我没有使用任何选择性重播。本质上，我不会重新处理错误事件或已被拒绝的事件。

再次向前重播事件很简单

class EventProcessor...

  private void ReplayAfter(DomainEvent ev) {
    foreach (DomainEvent e in Consequences(ev)) BasicProcessEvent(e);
  }

现在让我们继续讨论替换。

class EventProcessor...

  private void ProcessReplacement(ReplacementEvent e) {
    if (e.Original.ShouldIgnoreOnReplay)
      throw new ProcessingException("Cannot replace event twice");
    else if (null == e.Replacement) 
      ProcessRejection(e);
    else if (e.HasPriorReplacement) 
      ProcessPriorReplacement(e);
    else 
      ProcessPriorOriginal(e);
    
  }

class ReplacementEvent...

  public bool HasPriorReplacement {
    get {
      if (null == replacement) return false;
      else return original.after(replacement);
    }
  }

这里有几个案例需要处理。如果原始事件已经被标记为忽略，这意味着存在处理错误，因为我们不应该拒绝已经被拒绝的事件。完成此操作后，我们有三个主要案例：替换是拒绝，替换事件早于原始事件，以及原始事件早于替换事件。

让我们从拒绝案例开始，它由一个 null 替换事件指示。在这里，我们回溯到被拒绝的事件，拒绝它，逆转它，然后向前重播。

class EventProcessor...

  private void ProcessRejection(ReplacementEvent e) {
    RewindTo(e.Original);
    BasicReverseEvent(e.Original);
    e.Original.Reject();
    ReplayAfter(e.Original);
  }

class DomainEvent...

  public bool after (DomainEvent other) {
    return this.CompareTo(other) > 0;
  }
  public void Reject() {
    _isRejected = true;
  }
  private bool _isRejected;
  public virtual bool ShouldIgnoreOnReplay {
    get {
      if (WasProcessingError) return true;
      return _isRejected;
    }
  }

拒绝会标记事件，使其不再被处理或回溯。

如果我们有一个先前的替换，我们回溯到替换，拒绝原始事件，处理替换，然后向前重播。

class EventProcessor...

  private void ProcessPriorReplacement(ReplacementEvent e) {
    RewindTo(e.Replacement);
    e.Original.Reject();
    BasicProcessEvent(e.Replacement);
    ReplayAfter(e.Replacement);
  }

对于先前的原始事件，我们回溯到原始事件，逆转并拒绝它，重播到替换，处理它，然后继续向前重播。

class EventProcessor...

  private void ProcessPriorOriginal(ReplacementEvent e) {
    RewindTo(e.Original);
    BasicReverseEvent(e.Original);
    e.Original.Reject();
    ReplayBetween(e.Original, e.Replacement);
    BasicProcessEvent(e.Replacement);
    ReplayAfter(e.Replacement);
  }
  private void ReplayBetween(DomainEvent first, DomainEvent last) {
    IList eventsToReplay = new ArrayList();
    foreach (DomainEvent e in log) {
      if (e.IsConsequenceOf(first) && last.after(e))
        eventsToReplay.Add(e);
    }
    foreach (DomainEvent e in eventsToReplay)
      BasicProcessEvent(e);
  }

我在这里遵循一个约定，即在事件成功处理之前不会将事件添加到日志中。如果日志不在与应用程序状态相同的交易中，这很有用。如果它们都在一个交易中，我可以在到达分支点并向前重播后将事件添加到日志中（这将拾取新事件）。

示例：更新外部系统（C#）

对于简单的事件溯源，在重播期间关闭外部通知就足够了。然而，对于追溯事件，我们需要更进一步，进行检测和修正。为了在回溯/重播期间做到这一点，我们将保存所有生成的事件并进行比较，以确定我们需要修正什么。然后，我们将假设一个简单的自动修正案例，即发送取消消息和新消息以进行更改。

我将继续使用航运示例，这里假设美国海关当局需要在任何货物进入经过加拿大的美国港口时收到通知。

外部通知由 Cargo 对象在处理到达事件时发出。

class Cargo...

  public void HandleArrival(ArrivalEvent ev) {
    ev.priorCargoInCanada[this] = _hasBeenInCanada;
    if ("CA" == ev.Port.Country) 
      _hasBeenInCanada = true;
    if (HasBeenInCanada && "US" == ev.Port.Country) {
      Registry.CustomsGateway.Notify(ev.Occurred, ev.Ship, ev.Port);
      ev.WasNotificationSent = true;
    }
  }
  private bool _hasBeenInCanada = false;
  public bool HasBeenInCanada {get { return _hasBeenInCanada;}}

为了逆转此事件，我们在正向播放中保存了是否发送通知的记录，并在发送时再次发送它。

class Cargo...

  public void ReverseArrival(ArrivalEvent ev) {
    _hasBeenInCanada = (bool) ev.priorCargoInCanada[this];
    if (ev.WasNotificationSent) 
      Registry.CustomsGateway.Notify(ev.Occurred, ev.Ship, ev.Port);
  }

我在这里遵循的原则是，领域模型应该不知道事件处理中的重播逻辑。它知道如何逆转每个事件的自身状态，但并不关心与外部系统通信的复杂性。

与外部系统通信的逻辑由网关处理，在本例中是一个对象集群。

图 1：一个对象集群构成了处理追溯性的网关。

在这个例子中，CustomsGatewayFront 是一个普通的网关，它将面向领域的接口（由 ICustomsGateway 定义）转换为实际的消息传递基础设施。我们可以安全地忽略它做了什么 - 我们只是假设如果我们调用它上的方法，它会确保海关办公室收到消息。

有趣的事情发生在前面，由 CustomsGatewayBuffer 驱动，它包装了实际的 CustomsGatewayFront，实现了相同的接口，但增加了处理追溯回溯的能力。

由于大部分追溯回溯是通用的，我可以将通用行为放到另一个类 ReplayBuffer 中，这样 Customs Gateway Buffer 只处理与海关案例相关的部分。回溯缓冲区确实需要与事件处理器进行一些通信，以处理回溯/重播过程中的各个阶段。它还需要某种方法来调整最终不匹配的事件，在本例中，我将此任务分配给海关网关缓冲区。在这两种情况下，回溯缓冲区都通过协作接口进行通信，以保持其通用性。

所以这就是主要人物，现在让我们进入行动。我将从我们正在实时运行的正常情况开始。海关网关缓冲区通过创建一个更新事件并将其发送到回溯缓冲区来实现通知操作。

class CustomsGatewayBuffer...

  public void Notify (DateTime arrivalDate, Ship ship, Port port) {
    CustomsNotificationEvent ev =
        new CustomsNotificationEvent(gateway, arrivalDate, ship, port);
    buffer.Send(ev);
  }
  ICustomsGateway gateway;
  ReplayBuffer buffer;

class CustomsNotificationEvent...

  public CustomsNotificationEvent(
    ICustomsGateway gateway, DateTime arrivalDate, 
    Ship ship, Port port) 
  {
    this.gateway = gateway;
    this.arrivalDate = arrivalDate;
    this.ship = ship;
    this.port = port;
  }

如果回溯缓冲区处于活动状态，它将处理该事件，这将导致对实际网关前端的调用。

class ReplayBuffer...

  internal void Send(IGatewayEvent ev) {
    current.Add(ev);
    if (isActive) 
      ev.Process();
  }

class CustomsNotificationEvent...

  public virtual void Process() {
    gateway.Notify(arrivalDate, ship, port) ;
  }

（我将在稍后讨论回溯缓冲区如何变得活跃以及当前列表是什么。）

所以如您所见，对于领域事件的活动处理，所有发生的只是网关缓冲区创建一个网关事件，将其发送到回溯缓冲区，回溯缓冲区处理它，这会导致将原始调用应用于实际网关。所有这些都是一种极其复杂的简单委托形式。

这种怪异的回报在于处理追溯事件。回溯缓冲区通过可回溯接口连接到事件处理器。

class ReplayBuffer...

  IRewindable eventProcessor;

internal interface IRewindable  {
  event EventProcessor.EventHandler RewindStarted, RewindFinished, ReplayFinished;    
}

此接口定义了三个重要的事件。事件处理器在其处理各种案例的过程中会发出这些事件。以下是乱序案例

class EventProcessor...

  private void ProcessOutOfOrder(DomainEvent e) {
    RewindStarted();
    RewindTo(e);
    RewindFinished();
    BasicProcessEvent(e);
    ReplayAfter(e);
    ReplayFinished();
  }

对于替换，总体开始和结束是通用的，但每个案例都需要特定的放置来指示回溯何时结束。

class EventProcessor...

  private void ProcessReplacement(ReplacementEvent e) {
    RewindStarted();
    if (e.Original.ShouldIgnoreOnReplay)
      throw new ProcessingException("Cannot replace event twice");
    else if (null == e.Replacement) 
      ProcessRejection(e);
    else if (e.HasPriorReplacement) 
      ProcessPriorReplacement(e);
    else 
      ProcessPriorOriginal(e);
    ReplayFinished();
  }

private void ProcessRejection(ReplacementEvent e) {
  RewindTo(e.Original);
  BasicReverseEvent(e.Original);
  RewindFinished();
  e.Original.Reject();
  ReplayAfter(e.Original);
}

private void ProcessPriorOriginal(ReplacementEvent e) {
  RewindTo(e.Original);
  BasicReverseEvent(e.Original);
  RewindFinished();
  e.Original.Reject();
  ReplayBetween(e.Original, e.Replacement);
  BasicProcessEvent(e.Replacement);
  ReplayAfter(e.Replacement);
}

private void ProcessPriorReplacement(ReplacementEvent e) {
  RewindTo(e.Replacement);
  RewindFinished();
  e.Original.Reject();
  BasicProcessEvent(e.Replacement);
  ReplayAfter(e.Replacement);
}

这些事件为回溯缓冲区提供了正确的时间信息，以正确处理追溯处理。我在这里使用事件，因为任何事件处理器都可能具有任意数量的回溯缓冲区，回溯缓冲区不需要了解这些对象，除了它们可以理解这三个事件之外。

所以现在让我们转到回溯缓冲区，看看它如何处理这些事件。回溯缓冲区是用一堆列表创建的。

class ReplayBuffer...

  internal ReplayBuffer (IRewindable processor, IAdjustable adjuster) {
    this.eventProcessor = processor;
    this.adjuster = adjuster;
    SubscribeToProcessorEvents();
    sent = new ArrayList();
    isActive = true;
    current = sent;
  }
  IList sent,rewound, replayed, current;
  IAdjustable adjuster;
  private bool isActive;
  private void SubscribeToProcessorEvents() {
    eventProcessor.RewindStarted += 
      new EventProcessor.EventHandler(processor_RewindStarted);
    eventProcessor.RewindFinished += 
      new EventProcessor.EventHandler(processor_RewindFinished);
    eventProcessor.ReplayFinished += 
      new EventProcessor.EventHandler(processor_ReplayFinished);
  }

缓冲区创建后处于激活状态，这意味着它将处理发送到它的任何事件。当前列表变量将设置为三个核心列表之一，具体取决于我们处于回溯的哪个阶段 - 激活时它指向已发送列表。这让我们记录了所有实际发送的事件。它还订阅了事件处理器上的所有相关事件。（暂时不用担心调整器，我们将在校正中讨论它。）

在回溯开始时，缓冲区使自身处于非活动状态以避免将事件传递到真实网关，它还交换当前列表变量以指向一个新的列表，以便它可以捕获所有回溯的事件。

class ReplayBuffer...

  private void processor_RewindStarted() {
    isActive = false;
    rewound = new ArrayList();
    current = rewound;
  }

通过这个例子，你可能已经猜到回溯停止时会发生什么，但以防万一……

class ReplayBuffer...

  private void processor_RewindFinished() {
    replayed = new ArrayList();
    current = replayed;
  }

所以现在很清楚，最后一个事件将是最有趣的，因为我们必须在这里计算不匹配列表。

class ReplayBuffer...

  private void processor_ReplayFinished() {
    current = sent;
    DetermineChange();
    isActive = true;
    rewound = null;
    replayed = null;
  }
  private void DetermineChange() {
    IList matchingEvents = new ArrayList();
    foreach (IGatewayEvent ev in replayed) {
      if (rewound.Contains(ev)) matchingEvents.Add(ev);
    }
    foreach (IGatewayEvent ev in matchingEvents) {
      replayed.Remove(ev);
      rewound.Remove(ev);
    }
    adjuster.Adjust(rewound, replayed);
  }

算法实际上很简单，我们只需要找到匹配的事件并将它们从两个列表中删除。完成后，我们有两个不匹配事件列表，我们将它们传递给调整器处理。

类 CustomsGatewayBuffer…

  public void Adjust(IList oldEvents, IList newEvents) {
    foreach (CustomsNotificationEvent e in oldEvents)
      new CustomsCancellationEvent(e).Process();
    foreach (CustomsNotificationEvent e in newEvents)
      e.Process();
  }
class CustomsCancellationEvent {
  private CustomsNotificationEvent original;
  public CustomsCancellationEvent(CustomsNotificationEvent original) {
    this.original = original;
  }
  public void Process() {
    original.Gateway.Cancel(original.ArrivalDate, original.Ship, original.Port);
  }
}

在这种情况下，我让海关网关缓冲区处理调整，方法是简单地为所有最终回溯的事件发送取消通知，并为新事件发送新的通知。在我的幻想世界中，政府机构非常乐于助人。在实践中，每个调整都需要单独考虑，它们很容易变得非常复杂，需要人工干预。但是，事件列表应该极大地帮助理清外部系统需要做什么。