将事件发布到 Saga 时没有具有实际事务的 EntityManager

Question

我使用的是“axon 4.0.3 + Spring Boot 2 + Spring Data (PostgreSQL)”默认配置。

已将事件发布到 EventStore 并等待它被 @SagaEventHandler 捕获，我收到以下异常：

javax.persistence.TransactionRequiredException: 没有 EntityManager 可用于当前线程的实际事务 - 不能可靠 处理“坚持”调用 org.springframework.orm.jpa.SharedEntityManagerCreator$SharedEntityManagerInvocationHandler.invoke(SharedEntityManagerCreator.java:292) ~[spring-orm-5.1.5.RELEASE.jar:5.1.5.RELEASE] 在 com.sun.proxy.$Proxy104.persist(Unknown Source) ~[na:na] at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184) ~[na:1.8.0_191] 在 java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193) ~[na:1.8.0_191] 在 java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382) ~[na:1.8.0_191] 在 java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481) ~[na:1.8.0_191] 在 java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471) ~[na:1.8.0_191] 在 java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151) ~[na:1.8.0_191] 在 java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174) ~[na:1.8.0_191] 在 java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[na:1.8.0_191] 在 java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418) ~[na:1.8.0_191] 在 org.axonframework.eventsourcing.eventstore.jpa.JpaEventStorageEngine.appendEvents(JpaEventStorageEngine.java:276) ~[axon-eventsourcing-4.0.3.jar:4.0.3] 在 org.axonframework.eventsourcing.eventstore.AbstractEventStorageEngine.appendEvents(AbstractEventStorageEngine.java:98) ~[axon-eventsourcing-4.0.3.jar:4.0.3]

EventStore 需要什么额外配置来处理这种情况？

附言。在方法上添加@Transactional 解决了这个问题，但我不明白为什么这是必要的。

最小代码示例（以下端点 127.0.0.1:8080/1 有效，但另一个端点 127.0.0.1:8080/1 无效）：

@SpringBootApplication
class TestAxonApplication

class UserId(val userId: String = IdentifierFactory.getInstance().generateIdentifier()) : Serializable

class TestCommand(@TargetAggregateIdentifier val userId: UserId)

class TestedEvent(val userId: UserId)

fun main(args: Array<String>) {
    runApplication<TestAxonApplication>(*args)
}

@RestController
@RequestMapping
class Controller(var commandGateway: CommandGateway, var eventStore: EventStore) {

    @GetMapping("/1")
    fun done(): UserId? {
        return commandGateway.sendAndWait<UserId>(TestCommand(UserId()))
    }

    @GetMapping("/2")
    fun failure() {
        eventStore.publish(
                GenericEventMessage.asEventMessage<Void>(
                        TestedEvent(UserId())
                )
        )
    }

}

@Aggregate
class User() {

    @AggregateIdentifier
    private lateinit var userId: UserId

    @CommandHandler
    constructor(cmd: TestCommand) : this() {
        AggregateLifecycle.apply(TestedEvent(cmd.userId))
    }

    @EventHandler
    fun on(event: TestedEvent) {
        this.userId = event.userId
    }

}

@Saga
@ProcessingGroup("mySaga")
class MySaga {

    @StartSaga
    @SagaEventHandler(associationProperty = "userId")
    fun start(event: TestedEvent) {
        println("DONE ${event.userId.userId}")
    }

}

Answer 1

调用之间的区别在于，一个通过命令总线，而另一个跳过该调用并直接发布到事件总线。默认情况下，在命令总线上配置TransactionManager。但是，在事件总线上并非如此。

这意味着您正在发布一个事件，而没有激活事务。 Hibernate 不喜欢这样。

解决方案是将@Transactional 放在您的端点上，以确保在存储事件时事务处于活动状态。

Answer 2

您似乎忘记设置命令总线以允许事务管理。像下面的例子一样添加它，它会起作用：

@Bean
public CommandBus commandBus(TransactionManager transactionManager) {
    return new SimpleCommandBus(transactionManager, NoOpMessageMonitor.INSTANCE);
}

更新

我的错，首先我习惯使用 Axon 3.3（不是最新的），我认为您正在使用事件存储的自定义配置。

Axon spring boot 默认使用 inmemory 事件存储，这就是为什么如果您不定义自定义事件存储，则事务将无法工作。

这是您的配置中缺少的内容：

@Bean
public EventStorageEngine eventStorageEngine(EntityManagerProvider entityManagerProvider, TransactionManager transactionManager) {
    return new JpaEventStorageEngine(entityManagerProvider, transactionManager);
}

@Bean
public PlatformTransactionManager transactionManager(EntityManagerFactory emf) {
    return new JpaTransactionManager(emf);
}

´´´

这是我为其他提案所做的一个简单示例，效果很好。

注意事项

另一个有趣的地方是您必须定义聚合存储库，因为 Axon 会尝试寻找与定义的聚合匹配的存储库。

@Bean
public Repository<User> documentAggregateRepository(EventStore eventStore) {
    return new EventSourcingRepository<>(User.class, eventStore);
}

saga 也是一样，你必须注册它们，否则 saga 永远不会被触发

// remember to call the method sagaName + Configuration
// or you must set up the @Saga configurationBean name pointing this method
@Bean
public SagaConfiguration<MySaga> mySagaConfiguration() {
    return SagaConfiguration.subscribingSagaManager(MySaga.class);
}

在此示例中是事件溯源，但如果您对遗留迁移感兴趣，也许您会在迁移过程中使用 GenericJpaRepository（如果您需要示例，请告诉我）。

HTH。