用Spring集成实现MongoDB入站流

Question

我们将有一个Mongo集合包含多个工作单元。我的想法是，文档将有一个状态字段，其中有四个选项：未处理、处理、完成、失败。Spring Integration将被配置为从这个db读取并处理存储在那里的消息。

入站Mongo DSL流将根据未处理的值从集合中读取：

MongoDbMessageSource messageSource = new MongoDbMessageSource(mongo, new LiteralExpression("{'status' : 'UNPROCESSED'}"));
return IntegrationFlows.from(messageSource)...

这里有一个问题：如果我有几台工作机器从同一个数据库中读取数据，我希望防止它们对同一行未处理的数据进行操作，因为我的poller使用maxmessagesperpoll的保守值，或者消息处理需要一段时间。

似乎正确的方法是使用TransactionSynchronizationFactory定义一个ProcessBeforeCommit阶段将状态更新为PROCESSING，并定义一个ProcessAfterCommit阶段将状态更新为DONE或Failed。但是，在查看Pollers和TransactionManagers的API时，我并不清楚添加这一点的机制。有XML中的例子，但我看不到使用DSL的例子。

我还希望确保ProcessBeforeCommit发生在数据库读取时，而不是在处理之后······是吗？此外，如果这不是从Mongo集合中读取解决方案的最佳设计方法，请随时建议更好的架构。

Answer 1

不，processBeforeCommit和processafterCommit是非常接近的回调。他们肯定会发生在你的过程的最后。让我们假设您有一个方法，如下所示：

@Transactional
void foo() {}

当您调用这样的方法时，事务在进入方法体之前就开始了。当我们在方法体执行后退出它时，将执行beforeCommit回调。它可能会失败，因为在我们的过程中有一个外部连接（DB？）可能会丢失。只有在没有问题的情况下，我们才继续执行提交后的。

可以通过AbstractMessageSourceAdvision实现：https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-channels-section.html#condition-pollers来完成您所要求的内容。因此，在afterreceive()实现中，您可以将文档更新为processing，甚至决定返回null，而不是消息：仅仅因为它在数据库中的状态已经是processing。这样的建议可以被注入到pollerspec:

/**
 * Specify AOP {@link Advice}s for the {@code pollingTask}.
 * @param advice the {@link Advice}s to use.
 * @return the spec.
 */
public PollerSpec advice(Advice... advice) {

/**
 * Specify the {@link TransactionSynchronizationFactory} to attach a
 * {@link org.springframework.transaction.support.TransactionSynchronization}
 * to the transaction around {@code poll} operation.
 * @param transactionSynchronizationFactory the TransactionSynchronizationFactory to use.
 * @return the spec.
 */
public PollerSpec transactionSynchronizationFactory(