问题描述
我有一个usersHttpService:
accept(code: string,user: UserInfo) {
return this.http.post(`${this.env.MY URL}/mon/accepting/${code}/accept`,user);
}
,CommandHandler
和EventSourcingHandler
链,当直接分派命令EventHandler
时,它们都按预期被调用。如果从Foo
或公用EventHandler
中分派同一命令,则不会导致调用同一Saga
。
不幸的是,我无法透露实际的代码,但这是发生的情况的说明:
未调用的EventHandler
仅将事件作为参数,没有带注释的EventHandler
或类似参数。它负责写入读取模型,@MetaData
需要继续执行该读取模型。此时,此过程失败,因为尚未更新读取模型,因为尚未调用@SagaEventHandler
。
这里可能是什么问题?使用Axon 4.3.1。
配置:
EventHandler
简短的通话链:
- 通过HTTP请求接收到初始命令。
/*
* 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;
}
}
- 命令A产生一个新的
@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
-
@Aggregate public class MyFooAggregate { @CommandHandler public MyFooAggregate(CommandA cmd,MetaData MetaData) { apply(new FooCreated(),MetaData); } ... }
接获@Saga
事件并发出新命令。没有针对sagas的独特配置类。
FooCreated
- 不是集合的专用命令处理程序负责命令B。它使用接口
@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); } }
,实现为该接口调度另一个命令。该处理程序已在配置中注册(请参见MyProvider
)。
AxonConfig
-
@Component public class CommandHandler { MyProvider provider; @CommandHandler public void dob(CommandB cmdB) { ... provider.provideMeWithSomething(); ... } }
的实现最终调度了命令C。
MyProvider
- 命令C产生一个新的
@Component public class MyDefaultProvider implements MyProvider { @Override public Object provideMeWithSomething() { CommandC cmdC = new CommandC(); commandGateway.sendAndWait(cmdC); ... } }
。创建此聚合所调度的事件是@Aggregate
中正在处理的事件,但找不到到达@EventSourcingHandler
的事件。
@EventHandler
- 同一类
@Aggregate public class MyBaraggregate { @CommandHandler public MyBaraggregate(CommandC cmdC) { apply(new BarCreated()); } ... }
中的@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
确实会被调用,只有在这种特定情况下才有此特定实例。
解决方法
首先,我在考虑您使用的EventProcessor
的类型,试图找出问题所在。如配置所示(感谢添加),您正在使用SubscribingEventProcessor
。
从本质上讲,这意味着发布命令(在本例中为Saga)的同一线程(至少应该)调用了@EventHandler
注释方法,因为这就是SubscribingEventProcessor
的工作方式。尤其是如果您正在执行CommandGateway#sendAndWait(Object)
,这意味着锁定Saga的线程还将锁定聚集以处理命令,并将负责向SubscribingEventProcessor
的事件提供相同的事件。所说的事件处理程序(未被调用)。
因此,第二,您如何调度该命令?不能共享确切的代码很好,但是sample / pseudocode应该可行吧?
第三,传奇如何配置?只是具有@Saga
构造型的常规Spring自动配置,还是有独特的配置类?
说实话,我仍然很难理解为什么没有调用负责SubscribingEventProcessor
注释功能的@EventHandler
。您可以确认是否未调用属于此SubscribingEventProcessor
的所有事件处理程序,或者仅仅是该事件处理类吗?