从 Saga 分派命令时未调用事件处理程序

Question

我有一个CommandHandler、EventSourcingHandler 和EventHandler 链，当直接发送命令Foo 时，它们都会按预期调用。如果从Saga 或公共EventHandler 内调度相同的命令，则不会调用相同的EventHandler。

很遗憾，我无法透露实际代码，但这里有一个说明：

未调用的EventHandler 仅将事件作为参数，没有@MetaData 注释参数或类似参数。它负责写入读取模型，@SagaEventHandler 需要继续执行。此时该过程失败，因为尚未更新读取模型，因为尚未调用 EventHandler。

这可能是什么问题？使用 Axon 4.3.1。

配置：

/*
 * Copyright (c) 2020 divine Cloud Services Ltd.
 * All rights reserved
 *
 * All information contained herein is and remains the property of
 * divine Cloud Services Ltd. The intellectual and technical concepts
 * contained herein are proprietary to divine Cloud Services Ltd. and
 * may be protected by trade secret or copyright law. Dissemination
 * of this information or reproduction of this material is strictly prohibited
 * unless prior written permission is obtained from divine Cloud Services Ltd.
 */

package engineering.divine.infrastructure.conf;

import engineering.divine.artefact.model.meta.ArtefactDefinition;
import engineering.divine.artefact.ports.ArtefactDefinitionRepository;
import engineering.divine.infrastructure.persistence.BinarySerialiser;
import engineering.divine.infrastructure.persistence.DefaultArtefactDefinitionRepository;
import engineering.divine.infrastructure.persistence.graph.GraphEventStorageEngine;
import engineering.divine.infrastructure.persistence.graph.database.GraphManager;
import engineering.divine.infrastructure.security.interceptors.AuditCommandDispatchInterceptor;
import engineering.divine.integration.github.GitHubCommandHandler;
import engineering.divine.integration.github.GitHubQueryHandler;
import org.axonframework.commandhandling.CommandBus;
import org.axonframework.commandhandling.CommandMessage;
import org.axonframework.commandhandling.SimpleCommandBus;
import org.axonframework.common.transaction.TransactionManager;
import org.axonframework.config.Configurer;
import org.axonframework.config.EventProcessingConfigurer;
import org.axonframework.eventhandling.PropagatingErrorHandler;
import org.axonframework.eventsourcing.EventCountSnapshotTriggerDefinition;
import org.axonframework.eventsourcing.EventSourcingRepository;
import org.axonframework.eventsourcing.SnapshotTriggerDefinition;
import org.axonframework.eventsourcing.Snapshotter;
import org.axonframework.eventsourcing.eventstore.EmbeddedEventStore;
import org.axonframework.eventsourcing.eventstore.EventStorageEngine;
import org.axonframework.eventsourcing.eventstore.EventStore;
import org.axonframework.messaging.interceptors.BeanValidationInterceptor;
import org.axonframework.modelling.command.Repository;
import org.axonframework.queryhandling.QueryBus;
import org.axonframework.queryhandling.QueryMessage;
import org.axonframework.queryhandling.SimpleQueryBus;
import org.axonframework.spring.eventsourcing.SpringAggregateSnapshotterFactoryBean;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import java.util.concurrent.Executors;

@SuppressWarnings("unchecked")
@Configuration
public class AxonConfig implements ApplicationContextAware {

    private ApplicationContext applicationContext;

    @Value("${engineering.divine.snapshot.threshold:30}")
    private int snapshotThreshold;

    @Bean
    public CommandBus commandBus(TransactionManager transactionManager,
                                 AuditCommandDispatchInterceptor auditCommandDispatchInterceptor,
                                 BeanValidationInterceptor<CommandMessage<?>> validationInterceptor) {

        SimpleCommandBus commandBus = SimpleCommandBus.builder()
                .transactionManager(transactionManager)
                .build();

        commandBus.registerDispatchInterceptor(auditCommandDispatchInterceptor);
        commandBus.registerDispatchInterceptor(validationInterceptor);

        return commandBus;
    }

    @Bean
    public QueryBus queryBus(TransactionManager transactionManager,
                             BeanValidationInterceptor<QueryMessage<?, ?>> validationInterceptor) {
        SimpleQueryBus queryBus = SimpleQueryBus.builder()
                .transactionManager(transactionManager)
                .build();

        queryBus.registerDispatchInterceptor(validationInterceptor);

        return queryBus;
    }

    @Bean
    @Primary
    EventStore eventStore(GraphManager graphManager, BinarySerialiser serialiser) {
        EventStorageEngine storageEngine = GraphEventStorageEngine
                .builder()
                .graphManager(graphManager)
                .serialiser(serialiser)
                .build();

        return EmbeddedEventStore.builder()
                .storageEngine(storageEngine)
                .build();
    }

    @Bean
    Snapshotter snapshotter(EventStore eventStore) {
        SpringAggregateSnapshotterFactoryBean factoryBean = new SpringAggregateSnapshotterFactoryBean();
        factoryBean.setApplicationContext(applicationContext);
        factoryBean.setEventStore(eventStore);
        factoryBean.setExecutor(Executors.newSingleThreadExecutor());
        return factoryBean.getObject();
    }

    @Bean
    SnapshotTriggerDefinition snapshotTriggerDefinition(Snapshotter snapshotter) {
        return new EventCountSnapshotTriggerDefinition(snapshotter, snapshotThreshold);
    }

    @Bean
    ArtefactDefinitionRepository artefactDefinitionRepository(EventStore eventStore) {
        Repository<ArtefactDefinition> delegate = EventSourcingRepository
                .builder(ArtefactDefinition.class)
                .eventStore(eventStore)
                .build();

        return new DefaultArtefactDefinitionRepository(delegate);
    }

    @Bean
    BeanValidationInterceptor<CommandMessage<?>> beanValidationCommandInterceptor() {
        return new BeanValidationInterceptor();
    }

    @Bean
    BeanValidationInterceptor<QueryMessage<?, ?>> beanValidationQueryInterceptor() {
        return new BeanValidationInterceptor();
    }

    @Autowired
    public void configure(EventProcessingConfigurer configurer) {
        configurer.usingSubscribingEventProcessors();
        configurer.registerDefaultListenerInvocationErrorHandler(configuration -> PropagatingErrorHandler.instance());
    }

    @Autowired
    public void configure(Configurer configurer, GitHubCommandHandler commandHandler, GitHubQueryHandler queryHandler) {
        configurer.registerCommandHandler(c -> commandHandler);
        configurer.registerQueryHandler(c -> queryHandler);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}

缩写调用链：

1. 初始命令通过 HTTP 请求接收。

@RestController
public class Controller {
  ...
  @RequestMapping("/doA")
  public CompletableFuture<Void> doA(@RequestBody CommandA cmdA) {
    ...
    // Do something else before cmdA
    CommandX cmdX = new CommandX();
    commandGateway.sendAndWait(cmdX);
    ...
    commandGateway.sendAndWait(cmdA);
  }
}

1. 命令 A 生成一个新的 @Aggregate。

@Aggregate
public class MyFooAggregate {
  @CommandHandler
  public MyFooAggregate(CommandA cmd, MetaData metaData) {
    apply(new FooCreated(), metaData);
  }
  ...
}

1. @Saga 接收到FooCreated 事件并发出新命令。 sagas 没有不同的配置类。

@Saga
public class MySaga {

  // (end will not be the same as start in the future)
  @StartSaga
  @EndSaga
  @SagaEventHandler(associationProperty = "id")
  public void handleFooCreated(FooCreated event, @MetaDataValue("valueX") String valueX, CommandGateway commandGateway) {
    CommandB cmdB = new CommandB();
    ...
    Map<String, Object> meta = new HashMap<>();
    meta.put("valueX", valueX);

    CommandMessage<CommandB> cmdMsg = GenericCommandMessage.asCommandMessage(cmdB);
    cmdMsg.withMetaData(meta);

    // because I don't need to wait for the result, I'll just "send" it
    commandGateway.send(cmdMsg);
  }
}

4. 一个不是聚合的专用命令处理程序负责命令 B。它使用接口MyProvider，实现为该接口调度另一个命令。处理程序已在配置中注册（请参阅AxonConfig）。

@Component
public class CommandHandler {

  MyProvider provider;

  @CommandHandler
  public void doB(CommandB cmdB) {
    ...
    provider.provideMeWithSomething();
    ...
  }
}

5. MyProvider的实现最终调度命令C。

@Component
public class MyDefaultProvider implements MyProvider {

  @Override
  public Object provideMeWithSomething() {
    CommandC cmdC = new CommandC();
    commandGateway.sendAndWait(cmdC);
    ...
  }  

}

6. 命令 C 生成一个新的@Aggregate。从该聚合的创建中分派的事件是在 @EventSourcingHandler 中处理的事件，但未找到到达 @EventHandler 的方式。

@Aggregate
public class MyBarAggregate {
  @CommandHandler
  public MyBarAggregate(CommandC cmdC) {
    apply(new BarCreated());
  }
  ...
}

1. 同一个类中的@EventSourcingHandler MyBarAggregate 被调用，但不是负责编写读取模型的常见@EventHandler。

@Aggregate
public class MyBarAggregate {
  ...
  @EventSourcingHandler
  public handleBarCreated(BarCreated event) {
    // DOES get called
    ...
  }
}

@Component
public class EventHandler {
  @EventHandler
  public handleBarCreated(BarCreated event) {
    // DOES NOT get called
    ...
  }
}

然而，当CommandC 被直接调度（例如从@RestController 中）时，两个事件处理程序都会被调用。其他@EventHandlers 确实会被调用，只有这个特定实例不会在这种特定情况下。

Answer 1

首先，我试图设想问题出在哪里，考虑您正在使用的EventProcessor 的类型。如您的配置所示（感谢添加），您使用的是SubscribingEventProcessor。

这实质上意味着发布命令的同一线程（在本例中为 Saga）正在（至少应该）调用 @EventHandler 注释方法，因为这正是 SubscribingEventProcessor 的工作方式。特别是如果您正在执行CommandGateway#sendAndWait(Object)，这意味着锁定您的 Saga 的线程也会锁定一个聚合以处理命令，并将负责向所述事件处理程序的SubscribingEventProcessor 提供相同的事件（没有被调用）。

那么，其次，你是如何调度命令的？很好，您不能共享确切的代码，但是示例/伪代码应该可行，对吗？ 第三，Saga是如何配置的？只是带有 @Saga 构造型的常规 Spring 自动配置，还是有一个独特的配置类？

说实话，我还是很难理解为什么说SubscribingEventProcessor负责@EventHandler注解的函数没有被调用。您能否确认是否未调用属于此SubscribingEventProcessor 一部分的所有 事件处理程序，还是仅此一个事件处理类？