从Saga发出命令时未调用事件处理程序

Question

我有一个CommandHandler、EventSourcingHandler和EventHandler链，当命令Foo被直接发送时，它们都会按预期被调用。如果同一个命令是从一个EventHandler或公共Saga中发出的，那么它不会导致调用相同的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);
  }
}

@Aggregate.

1. 命令生成一个新的
2. 命令

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

1. A @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);
  }
}

1. 不是一个专用的命令处理程序，它负责命令B。它使用一个接口MyProvider，实现为该接口分发另一个命令。处理程序已在配置中注册(请参阅AxonConfig).

)。

@Component
public class CommandHandler {

  MyProvider provider;

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

1. MyProvider最终调度命令C.

的实现

@Component
public class MyDefaultProvider implements MyProvider {

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

}

1. 命令C生成一个新的@Aggregate。创建此聚合时发送的事件是在@EventHandler.

中处理的事件，但没有找到到@EventSourcingHandler的方法。

@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内部)。其他的@EventHandler确实会被调用，只有这个特定的实例不会在这个特定的情况下被调用。

Answer 1

首先，我试图设想问题的所在，考虑您正在使用的EventProcessor类型。如您的配置所示(谢谢您的添加)，您正在使用SubscribingEventProcessor。

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

那么，第二，您是如何发送命令的？很好，您不能共享确切的代码，但是示例/伪代码应该是可行的，对吗？第三，Saga是如何配置的？只是使用@Saga原型的常规Spring自动配置，还是存在一个不同的配置类？

老实说，我仍然很难理解为什么没有调用负责@EventHandler注释函数的@EventHandler。您能否确认是否没有调用所有属于该SubscribingEventProcessor的事件处理程序，或者是否只是这一个事件处理类？