事件溯源 - 复杂的聚合设计

Question

我有以下代码规范示例，它在 Photoshop 中对图像进行建模。

图片以PhotoshopImage 给出。每个图像都有Layers，它是一个包含图像构成的所有层的对象，在我的例子中它只包含两层 - 第一个是实体层（DefaultLayer 的实例），第二个是透明层（实例NotifiableLayer)。每当DefaultLayer 更新时，我们还必须更新NotifiableLayer，它正在监听DefaultLayer（即下方）上的更改，以便它可以自我更新（例如当您更新下面图层上的一些黑色像素时，然后透明不透明度为 50% 且位于该较低层之上的层将显示该像素为灰色）。

实现如下：

public class ES2 {
    public static void main(String[] args) {
        PhotoshopImage image = new PhotoshopImage();

        //draw ine black pixel at position 1,1 in layer 1 (top transparent layer)
        DrawOneBlackPixelCommand command1 = new DrawOneBlackPixelCommand(1,1,new Coordinates(1,1));
        image.drawOneBlackPixel(command1);

        //draw one black pixel at position 0,0 in layer 0 (bottom solid layer)
        //this command will also affect transparent layer 1 via callback
        DrawOneBlackPixelCommand command2 = new DrawOneBlackPixelCommand(1,0,new Coordinates(0,0));
        image.drawOneBlackPixel(command2);

        int[][] imagePixels = image.getImagePixels();

        //[2, 0]
        //[0, 1]
        System.out.println(Arrays.toString(imagePixels[0]));
        System.out.println(Arrays.toString(imagePixels[1]));
    }
}

record DrawOneBlackPixelCommand(
    int imageId,
    int layerType,
    Coordinates pixelCoordinates
){}
record Coordinates(int x, int y){}

class PhotoshopImage{
    Integer imageId = 1;
    String imageName = "someName";
    LocalDateTime dateTime = LocalDateTime.now();
    Layers layers;

    PhotoshopImage(){
        layers = new Layers();
    }

    void drawOneBlackPixel(DrawOneBlackPixelCommand command){
        if(LocalDateTime.now().isBefore(dateTime)){
            throw new DrawingPixelTimeExpiredException();
        }
        layers.drawOneBlackPixel(command.layerType(), command.pixelCoordinates());
    }

    int[][] getImagePixels(){
        return layers.getVisibleLayerPixels();
    }

    class DrawingPixelTimeExpiredException extends RuntimeException{}
}

class Layers{
    Set<NotifiableLayer> notifiableLayerObservers = new HashSet<>();
    NavigableMap<Integer, Layer> layers = new TreeMap<>();

    Layers(){
        DefaultLayer solid = new DefaultLayer();
        NotifiableLayer transparent = new NotifiableLayer();
        layers.put(0, solid);
        layers.put(1, transparent);
        notifiableLayerObservers.add(transparent);
    }

    void drawOneBlackPixel(int layerType, Coordinates pixelCoordinates){
        if(!layers.containsKey(layerType)){
            throw new LayerDoesNotExistException();
        }
        Layer change = layers.get(layerType);
        change.drawOneBlackPixel(pixelCoordinates);
        notifiableLayerObservers.forEach(l -> l.notifyLayer(change, pixelCoordinates));
    }

    public int[][] getVisibleLayerPixels() {
        return layers.lastEntry().getValue().getLayerPixels();
    }

    class LayerDoesNotExistException extends RuntimeException{}
}

interface Layer{
    void drawOneBlackPixel(Coordinates coordinates);
    int[][] getLayerPixels();
}

class DefaultLayer implements Layer{
    int[][] pixels = new int[][]{{0,0},{0,0}};

    @Override
    public void drawOneBlackPixel(Coordinates c) {
        pixels[c.x()][c.y()] = 1;
    }

    @Override
    public int[][] getLayerPixels() {
        return pixels;
    }
}

class NotifiableLayer implements Layer{
    int[][] pixels = new int[][]{{0,0},{0,0}};

    void notifyLayer(Layer changed, Coordinates c){
        //if it is not this layer, then it is layer below (solid layer)
        if(changed!=this){
            int pixelInLayerBelow = changed.getLayerPixels()[c.x()][c.y()];
            syncPixelWithLayerBelow(pixelInLayerBelow, c);
        }
    }

    private void syncPixelWithLayerBelow(int pixelBelow, Coordinates c){
        pixels[c.x()][c.y()] = pixelBelow + 1;
    }

    @Override
    public void drawOneBlackPixel(Coordinates c) {
        pixels[c.x()][c.y()] = 1;
    }

    @Override
    public int[][] getLayerPixels() {
        return pixels;
    }
}

现在，这被实现为可变状态对象（即 - 它不使用事件源）。无论我阅读什么有关事件溯源的手册，它都仅基于一些超级简单的示例。

就我而言-我不知道如何创建事件OneBlackPixelDrawnEvent（一种方法是在下面的更新答案中，但对于 ES 带来的好处来说它看起来太复杂了）-这应该是代码中这两个操作的结果，以及如何应用这些事件 - 应该在PhotoshopImage 中应用，还是每个层都应该负责更新其部分状态？如何将这些事件从 PhotoshopImage 聚合转发到 Layers 并进一步向下？

更新 - 使用事件溯源的一种实施方式示例

import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

public class ES2 {
    public static void main(String[] args) {
        PhotoshopImage image = new PhotoshopImage();

        //draw ine black pixel at position 1,1 in layer 1 (top transparent layer)
        DrawOneBlackPixelCommand command1 = new DrawOneBlackPixelCommand(1,1,new Coordinates(1,1));
        List<Event> events1 = image.drawOneBlackPixel(command1);

        //[OneBlackPixelDrawnEvent[layerType=1, pixelCoordinates=Coordinates[x=1, y=1], pixelValue=1]]
        System.out.println(events1);

        //draw one black pixel at position 0,0 in layer 0 (bottom solid layer)
        //this command will also affect transparent layer 1 via callback
        DrawOneBlackPixelCommand command2 = new DrawOneBlackPixelCommand(1,0,new Coordinates(0,0));
        List<Event> events2 = image.drawOneBlackPixel(command2);

        //[OneBlackPixelDrawnEvent[layerType=0, pixelCoordinates=Coordinates[x=0, y=0], pixelValue=1], LayerSyncedEvent[layerType=1, pixelCoordinates=Coordinates[x=0, y=0], pixelValue=2]]
        System.out.println(events2);

        int[][] imagePixels = image.getImagePixels();

        //[2, 0]
        //[0, 1]
        System.out.println(Arrays.toString(imagePixels[0]));
        System.out.println(Arrays.toString(imagePixels[1]));
    }
}

interface Event{}
record DrawOneBlackPixelCommand(
    int imageId,
    int layerType,
    Coordinates pixelCoordinates
){}
record Coordinates(int x, int y){}

record OneBlackPixelDrawnEvent(
        Integer layerType,
        Coordinates pixelCoordinates,
        Integer pixelValue
) implements Event{}

class PhotoshopImage{
    Integer imageId = 1;
    String imageName = "someName";
    LocalDateTime dateTime = LocalDateTime.now();
    Layers layers;

    PhotoshopImage(){
        layers = new Layers();
    }

    List<Event> drawOneBlackPixel(DrawOneBlackPixelCommand command){
        if(LocalDateTime.now().isBefore(dateTime)){
            throw new DrawingPixelTimeExpiredException();
        }
        List<Event> events = layers.drawOneBlackPixel(command.layerType(), command.pixelCoordinates());
        apply(events);  //Only here we can update state of this aggregate, so it is not updated twice
        return events;
    }

    void apply(List<Event> events){
        layers.apply(events);
    }

    int[][] getImagePixels(){
        return layers.getVisibleLayerPixels();
    }

    class DrawingPixelTimeExpiredException extends RuntimeException{}
}

class Layers{
    Map<Integer, NotifiableLayer> notifiableLayerObservers = new HashMap<>();
    NavigableMap<Integer, Layer> layers = new TreeMap<>();

    Layers(){
        DefaultLayer solid = new DefaultLayer();
        NotifiableLayer transparent = new NotifiableLayer();
        layers.put(0, solid);
        layers.put(1, transparent);
        notifiableLayerObservers.put(1, transparent);
    }

    List<Event> drawOneBlackPixel(int layerType, Coordinates pixelCoordinates){
        if(!layers.containsKey(layerType)){
            throw new LayerDoesNotExistException();
        }
        Layer change = layers.get(layerType);
        OneBlackPixelDrawnEvent event = change.drawOneBlackPixel(pixelCoordinates);
        //Here, I have to add layerType, since it is a missing info on event!
        OneBlackPixelDrawnEvent updatedEvent = new OneBlackPixelDrawnEvent(layerType, event.pixelCoordinates(), event.pixelValue());
        List<LayerSyncedEvent> syncedEvents = notifiableLayerObservers.entrySet().stream()
                .map(en ->
                    en.getValue()
                            .notifyLayer(change, updatedEvent)
                            //Here we have to re-pack event, since it is missing some info that can be
                            //filled only on this level
                            .map(e -> new LayerSyncedEvent(en.getKey(), e.pixelCoordinates(), e.pixelValue()))
                )
                .flatMap(Optional::stream)
                .collect(Collectors.toList());
        List<Event> results = new ArrayList<>();
        results.add(updatedEvent);
        results.addAll(syncedEvents);
        //apply(results); we still cannot apply here, since applying in aggregate root would apply twice!
        return results;
    }

    public void apply(List<Event> events){
        for(Event e : events){
            if(e instanceof LayerSyncedEvent ev){
                layers.get(ev.layerType()).apply(ev);
            }
            if(e instanceof OneBlackPixelDrawnEvent ev){
                layers.get(ev.layerType()).apply(ev);
            }
        }
    }

    public int[][] getVisibleLayerPixels() {
        return layers.lastEntry().getValue().getLayerPixels();
    }

    class LayerDoesNotExistException extends RuntimeException{}
}

interface Layer{
    OneBlackPixelDrawnEvent drawOneBlackPixel(Coordinates coordinates);
    int[][] getLayerPixels();
    <T extends Event> void apply(T e);
}

class DefaultLayer implements Layer{
    int[][] pixels = new int[][]{{0,0},{0,0}};

    @Override
    public OneBlackPixelDrawnEvent drawOneBlackPixel(Coordinates c) {
        OneBlackPixelDrawnEvent event = new OneBlackPixelDrawnEvent(null, c, 1);
        //apply(event); ! Since applying in aggregate root - cannot apply here!
        return event;
    }

    @Override
    public int[][] getLayerPixels() {
        return pixels;
    }

    @Override
    public <T extends Event> void apply(T e) {
        if(e instanceof OneBlackPixelDrawnEvent ev){
            Coordinates c = ev.pixelCoordinates();
            pixels[c.x()][c.y()] = ev.pixelValue();
        }
    }
}

record LayerSyncedEvent(
        Integer layerType,
        Coordinates pixelCoordinates,
        Integer pixelValue
) implements Event{}

class NotifiableLayer implements Layer{
    int[][] pixels = new int[][]{{0,0},{0,0}};

    Optional<LayerSyncedEvent> notifyLayer(Layer changed, OneBlackPixelDrawnEvent event){
        //if it is not this layer, then it is layer below (solid layer)
        if(changed!=this){
            Coordinates c = event.pixelCoordinates();
            //Since layer is not updated anymore in-place, we have to take changes from event!
            //int pixelInLayerBelow = changed.getLayerPixels()[c.x()][c.y()];
            int pixelInLayerBelow = event.pixelValue();
            return Optional.of(syncPixelWithLayerBelow(pixelInLayerBelow, c));
        }
        return Optional.empty();
    }

    private LayerSyncedEvent syncPixelWithLayerBelow(int pixelBelow, Coordinates c){
        LayerSyncedEvent event = new LayerSyncedEvent(null, c, pixelBelow + 1);
        //apply(event); ! Since applying in aggregate root - cannot apply here!
        return event;
    }

    @Override
    public OneBlackPixelDrawnEvent drawOneBlackPixel(Coordinates c) {
        OneBlackPixelDrawnEvent event = new OneBlackPixelDrawnEvent(null, c, 1);
        //apply(event); ! Since applying in aggregate root - cannot apply here!
        return event;
    }

    @Override
    public int[][] getLayerPixels() {
        return pixels;
    }

    @Override
    public <T extends Event> void apply(T e) {
        if(e instanceof LayerSyncedEvent ev){
            Coordinates c = ev.pixelCoordinates();
            pixels[c.x()][c.y()] = ev.pixelValue();
        }
        if(e instanceof OneBlackPixelDrawnEvent ev){
            Coordinates c = ev.pixelCoordinates();
            pixels[c.x()][c.y()] = ev.pixelValue();
        }
    }
}

我刚刚更新了这里的示例，使用一种实现聚合根的方法，以及返回事件的方法。我想这是一种可能的实现方式——但看看现在这有多复杂；即使是这个简单的例子——复杂性也增加了 2 倍。是我做错了什么，还是在事件源系统中这不是那么容易做到的？

Answer 1

尽管值得商榷，但我会争论“photoshopping”是否是您想要实现的领域，并考虑到 DDD、CQRS 和事件溯源等范式。至于 VoicOfUnreason 提到的某些方面，有时工作并没有超出收益；您可能只是选择了一个不可行的域。

无论如何，让我尝试为您的问题和您分享的 sn-ps 提供一些指导。我要强调的第一件事是从您的命令处理程序返回 List<Event> 对象。尽管在国产 DDD/CQRS/ES 系统中是合理的，但这不是您对基于 Axon 框架的应用程序所做的事情（我假设您正在通过 axon 标签使用它）。

命令处理程序应该只共享操作是成功、失败还是新创建实体的标识符。就是这样。

另一个值得分享的指针是命令处理程序的放置。您当前已将其设计为从 PhotoshopImage 开始。然而，命令可以完全针对聚合中的一个确切实体。从定义立场来看，这也很好，因为：

聚合是一组关联对象，它们在数据更改方面充当一个单元。有一个对聚合的引用，称为聚合根。最后，一致性规则适用于聚合的边界内。

因此，整个聚合（在您的示例中）由 PhotoshopImage 和 Layer 实体列表组成。在这种情况下，PhotoshopImage 是您的聚合根。采用“单一引用”参数，这意味着命令将始终流经聚合根，即PhotoshopImage。然而，这并没有使PhotoshopImage 实体成为负责决定处理命令的对象。

从实现的外观来看，如果我正确地遵循了您的描述，则有必要在根中处理操作以将操作委托给所有层。这确实会选择命令处理程序，因为它现在位于。

在事件发布后，您可以大大简化事情。请注意，虽然在这种情况下我基于 Axon 框架，但我认为这是公平的，因为正在使用 axon 标签。现在，发布事件的是PhotoshopImage。我会让您的每个图层都发布它自己的OneBlackPixelDrawnEvent。当您将使用事件溯源时，在聚合边界内发布和处理此类事件将优先于进一步执行命令处理操作。

因此，无需在您的示例中调用notifiableLayerObservers 来正确通知所有层。这应该只是您正在使用的 CQRS/DDD/ES 框架的一部分，因此 Axon 框架将为您提供开箱即用的功能。只需将方法标记为 @EventSourcingHandler，Axon 框架就不会为给定事件调用 所有 事件源处理程序，无论它们是驻留在聚合根中还是 任何的实体。 按照该路线，您可以在每个实体处理（在您的场景中）OneBlackPixelDrawnEvent 时调整正确的状态部分。

如前所述，在这种情况下，我假设您使用的是 Axon 之类的框架。或者，您有正确的实现分层来实现相同的目标。有了这样的设置，您就可以摆脱所有您当前在命令处理函数中执行的自定义路由信息。

最后一点，我正在对一个我不熟悉的域进行假设。如果在使用上述方法时有任何伤害，请务必发表评论，以便我们进一步讨论。同时，希望对您有所帮助！

Answer 2

事件溯源基于系统记录发生在聚合根上的事件的假设。在您的情况下，当更新图层时，包含它的图像会将事件附加到内部集合。 LayerUpdated... 之类的东西，尽管给事件起有意义的名称被认为是一种很好的做法。

当所有操作（又名Commands）都已执行后，系统开始持久化这些事件，并且对于每个事件，它还将广播一个通知。

现在您可以让每个 NotifiableLayer 监听特定通知，或者您可以拥有一个单独的服务来执行此操作并相应地更新所有 NotifiableLayer 实例。我会选择这项服务：我不太喜欢域实体监听通知的想法。