此版本仍在开发中，尚不被认为是稳定的。对于最新的稳定版本，请使用 Spring Integration 6.5.1！spring-doc.cadn.net.cn

MongoDb 支持

2.1 版引入了对 MongoDB 的支持：“高性能、开源、面向文档的数据库”。spring-doc.cadn.net.cn

您需要将此依赖项包含在您的项目中：spring-doc.cadn.net.cn

Mavenspring-doc.cadn.net.cn
Gradlespring-doc.cadn.net.cn

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mongodb</artifactId>
    <version>6.5.2-SNAPSHOT</version>
</dependency>

compile "org.springframework.integration:spring-integration-mongodb:6.5.2-SNAPSHOT"

要下载、安装和运行 MongoDB，请参阅 MongoDB 文档。spring-doc.cadn.net.cn

连接到 MongoDb

阻塞还是反应？

从版本 5.3 开始，Spring Integration 提供了对响应式 MongoDB 驱动程序的支持，以在访问 MongoDB 时启用非阻塞 I/O。要启用响应式支持，请将 MongoDB 响应式流驱动程序添加到您的依赖项中：spring-doc.cadn.net.cn

Mavenspring-doc.cadn.net.cn
Gradlespring-doc.cadn.net.cn

<dependency>
    <groupId>org.mongodb</groupId>
    <artifactId>mongodb-driver-reactivestreams</artifactId>
</dependency>

compile "org.mongodb:mongodb-driver-reactivestreams"

对于常规同步客户端，您需要将其各自的驱动程序添加到依赖项中：spring-doc.cadn.net.cn

Mavenspring-doc.cadn.net.cn
Gradlespring-doc.cadn.net.cn

<dependency>
    <groupId>org.mongodb</groupId>
    <artifactId>mongodb-driver-sync</artifactId>
</dependency>

compile "org.mongodb:mongodb-driver-sync"

他们俩都是optional在框架中，以获得更好的最终用户选择支持。spring-doc.cadn.net.cn

要开始与 MongoDB 交互，您首先需要连接到它。 Spring Integration 建立在另一个 Spring 项目 Spring Data MongoDB 提供的支持之上。它提供了名为MongoDatabaseFactory和ReactiveMongoDatabaseFactory，这简化了与 MongoDB 客户端 API 的集成。spring-doc.cadn.net.cn

Spring Data 默认提供阻塞 MongoDB 驱动程序，但您可以通过包含上述依赖项来选择响应式使用。

用`MongoDatabaseFactory`

要连接到 MongoDB，您可以使用MongoDatabaseFactory接口。spring-doc.cadn.net.cn

以下示例演示如何使用SimpleMongoClientDatabaseFactory:spring-doc.cadn.net.cn

Javaspring-doc.cadn.net.cn
XMLspring-doc.cadn.net.cn

MongoDatabaseFactory mongoDbFactory =
        new SimpleMongoClientDatabaseFactory(com.mongodb.client.MongoClients.create(), "test");

<bean id="mongoDbFactory" class="o.s.data.mongodb.core.SimpleMongoClientDatabaseFactory">
    <constructor-arg>
        <bean class="com.mongodb.client.MongoClients" factory-method="create"/>
    </constructor-arg>
    <constructor-arg value="test"/>
</bean>

SimpleMongoClientDatabaseFactory采用两个参数：MongoClient实例和String指定数据库的名称。如果您需要配置属性，例如host,port和其他构造函数，您可以使用基础MongoClients类。有关如何配置MongoDB的更多信息，请参阅Spring-Data-MongoDB参考。spring-doc.cadn.net.cn

用`ReactiveMongoDatabaseFactory`

要使用响应式驱动程序连接到 MongoDB，您可以使用ReactiveMongoDatabaseFactory接口。spring-doc.cadn.net.cn

以下示例演示如何使用SimpleReactiveMongoDatabaseFactory:spring-doc.cadn.net.cn

Javaspring-doc.cadn.net.cn
XMLspring-doc.cadn.net.cn

ReactiveMongoDatabaseFactory mongoDbFactory =
        new SimpleReactiveMongoDatabaseFactory(com.mongodb.reactivestreams.client.MongoClients.create(), "test");

<bean id="mongoDbFactory" class="o.s.data.mongodb.core.SimpleReactiveMongoDatabaseFactory">
    <constructor-arg>
        <bean class="com.mongodb.reactivestreams.client.MongoClients" factory-method="create"/>
    </constructor-arg>
    <constructor-arg value="test"/>
</bean>

MongoDB 消息存储

如《企业集成模式 （EIP）》一书中所述，消息存储允许您持久化消息。在处理能够缓冲消息（QueueChannel,aggregator,resequencer，等。如果可靠性是一个问题。在 Spring Integration 中，MessageStore策略还为声明检查模式提供了基础，这在 EIP 中也有描述。spring-doc.cadn.net.cn

Spring Integration 的 MongoDB 模块提供了MongoDbMessageStore，这是MessageStore策略（主要由声明检查模式使用）和MessageGroupStore策略（主要由聚合器和重排序器模式使用）。spring-doc.cadn.net.cn

以下示例配置MongoDbMessageStore使用QueueChannel和aggregator:spring-doc.cadn.net.cn

<bean id="mongoDbMessageStore" class="o.s.i.mongodb.store.MongoDbMessageStore">
    <constructor-arg ref="mongoDbFactory"/>
</bean>

<int:channel id="somePersistentQueueChannel">
    <int:queue message-store="mongoDbMessageStore"/>
<int:channel>

<int:aggregator input-channel="inputChannel" output-channel="outputChannel"
         message-store="mongoDbMessageStore"/>

前面的示例是一个简单的 bean 配置，它需要一个MongoDbFactory作为构造函数参数。spring-doc.cadn.net.cn

这MongoDbMessageStore展开Message作为具有所有嵌套属性的 Mongo 文档，使用 Spring Data Mongo 映射机制。当您需要访问payload或headers用于审计或分析，例如，针对存储的消息。spring-doc.cadn.net.cn

这MongoDbMessageStore使用自定义MappingMongoConverter实现存储Message实例作为 MongoDB 文档，并且属性（payload和header值）Message.

从 5.1.6 版开始，MongoDbMessageStore可配置自定义转换器，这些转换器可传播到内部MappingMongoConverter实现。看MongoDbMessageStore.setCustomConverters(Object… customConverters)JavaDocs 了解更多信息。spring-doc.cadn.net.cn

Spring Integration 3.0 引入了ConfigurableMongoDbMessageStore. 它实现了MessageStore和MessageGroupStore接口。此类可以作为构造函数参数接收MongoTemplate，例如，您可以使用它配置自定义WriteConcern. 另一个构造函数需要MappingMongoConverter和MongoDbFactory，这允许您为Message实例及其属性。请注意，默认情况下，ConfigurableMongoDbMessageStore使用标准的 Java 序列化进行写入和读取Message与MongoDB之间的实例（请参阅MongoDbMessageBytesConverter）并依赖于其他属性的默认值MongoTemplate. 它构建了一个MongoTemplate从提供的MongoDbFactory和MappingMongoConverter. 由ConfigurableMongoDbMessageStore是configurableStoreMessages. 建议在消息包含复杂数据类型时使用此实现来创建可靠且灵活的解决方案。spring-doc.cadn.net.cn

从版本 6.0.8 开始，AbstractConfigurableMongoDbMessageStore提供一个setCreateIndexes(boolean)（默认为true）选项，可用于禁用自动索引创建。以下示例显示了如何声明 bean 并禁用自动索引创建：spring-doc.cadn.net.cn

@Bean
public MongoDbChannelMessageStore mongoDbChannelMessageStore(MongoDatabaseFactory databaseFactory) {
    MongoDbChannelMessageStore mongoDbChannelMessageStore = new MongoDbChannelMessageStore(databaseFactory);
    mongoDbChannelMessageStore.setCreateIndexes(false);
    return mongoDbChannelMessageStore;
}

MongoDB 通道消息存储

4.0 版本引入了新的MongoDbChannelMessageStore. 这是一个优化的MessageGroupStore用于QueueChannel实例。跟priorityEnabled = true，您可以在<int:priority-queue>实例，实现持久化消息的优先级顺序轮询。优先级 MongoDB 文档字段从IntegrationMessageHeaderAccessor.PRIORITY (priority）消息头。spring-doc.cadn.net.cn

此外，所有 MongoDBMessageStore实例现在有一个sequence字段MessageGroup文件。这sequencevalue 是$inc作简单sequence来自同一集合的文档，该集合是按需创建的。这sequence字段用于poll作以提供先进先出（FIFO）消息顺序（如果已配置）在优先级内存储时。spring-doc.cadn.net.cn

我们不建议使用相同的MongoDbChannelMessageStorebean 表示优先级和非优先级，因为priorityEnabled选项适用于整个商店。然而，同样collection两者都可用于MongoDbChannelMessageStore类型，因为来自存储的消息轮询是排序的并使用索引。要配置该场景，您可以从另一个消息存储 Bean 扩展一个消息存储 Bean，如以下示例所示：

<bean id="channelStore" class="o.s.i.mongodb.store.MongoDbChannelMessageStore">
    <constructor-arg name="mongoDbFactory" ref="mongoDbFactory"/>
</bean>

<int:channel id="queueChannel">
    <int:queue message-store="store"/>
</int:channel>

<bean id="priorityStore" parent="channelStore">
    <property name="priorityEnabled" value="true"/>
</bean>

<int:channel id="priorityChannel">
    <int:priority-queue message-store="priorityStore"/>
</int:channel>

使用 AbstractConfigurableMongoDbMessageStore 禁用自动索引创建

从版本 6.0.8 开始，AbstractConfigurableMongoDbMessageStore实现一个setCreateIndex(boolean)可用于禁用或启用（默认）自动索引创建。以下示例显示如何声明 bean 并禁用自动索引创建：spring-doc.cadn.net.cn

@Bean
public AbstractConfigurableMongoDbMessageStore mongoDbChannelMessageStore(MongoDatabaseFactory databaseFactory)
{
    AbstractConfigurableMongoDbMessageStore mongoDbChannelMessageStore = new MongoDbChannelMessageStore(databaseFactory);
    mongoDbChannelMessageStore.setCreateIndex(false);

    return mongoDbChannelMessageStore;
}

MongoDB 元数据存储

Spring Integration 4.2 引入了基于 MongoDB 的新MetadataStore（请参阅元数据存储）实现。您可以使用MongoDbMetadataStore以在应用程序重新启动时维护元数据状态。您可以使用这个新的MetadataStore使用适配器实现，例如：spring-doc.cadn.net.cn

要指示这些适配器使用新的MongoDbMetadataStore，声明一个 Spring bean，其 bean 名称为metadataStore. 源入站通道适配器自动选取并使用声明的MongoDbMetadataStore. 以下示例显示了如何声明名称为metadataStore:spring-doc.cadn.net.cn

@Bean
public MetadataStore metadataStore(MongoDbFactory factory) {
    return new MongoDbMetadataStore(factory, "integrationMetadataStore");
}

这MongoDbMetadataStore还实现ConcurrentMetadataStore，使其在多个应用程序实例之间可靠地共享，其中只允许一个实例存储或修改键的值。所有这些作都是原子的，这要归功于 MongoDB 的保证。spring-doc.cadn.net.cn

MongoDB 入站通道适配器

MongoDB入站通道适配器是一个轮询消费者，它从MongoDB读取数据并将其作为Message有效载荷。以下示例显示了如何配置 MongoDB 入站通道适配器：spring-doc.cadn.net.cn

<int-mongodb:inbound-channel-adapter id="mongoInboundAdapter"
       channel="replyChannel"
       query="{'name' : 'Bob'}"
       entity-class="java.lang.Object"
       auto-startup="false">
		<int:poller fixed-rate="100"/>
</int-mongodb:inbound-channel-adapter>

如前面的配置所示，您可以使用inbound-channel-adapter元素并为各种属性提供值，例如：spring-doc.cadn.net.cn

query：JSON 查询（请参阅 MongoDB 查询)spring-doc.cadn.net.cn
query-expression：计算为 JSON 查询字符串的 SpEL 表达式（作为query属性）或添加到o.s.data.mongodb.core.query.Query. 与query属性。spring-doc.cadn.net.cn
entity-class：有效负载对象的类型。如果未提供，则com.mongodb.DBObject被返回。spring-doc.cadn.net.cn
collection-name或collection-name-expression：标识要使用的 MongoDB 集合的名称。spring-doc.cadn.net.cn
mongodb-factory：对o.s.data.mongodb.MongoDbFactoryspring-doc.cadn.net.cn
mongo-template：对o.s.data.mongodb.core.MongoTemplatespring-doc.cadn.net.cn
所有其他入站适配器中通用的其他属性（例如“channel”）。spring-doc.cadn.net.cn

不能同时设置两者mongo-template和mongodb-factory.

前面的示例相对简单且静态，因为它具有query并使用collection. 有时，您可能需要根据某些条件在运行时更改这些值。为此，请使用其-expression等效（query-expression和collection-name-expression），其中提供的表达式可以是任何有效的 SpEL 表达式。spring-doc.cadn.net.cn

此外，您可能希望对从 MongoDB 读取的成功处理的数据进行一些后处理。例如;您可能希望在处理文档后移动或删除文档。您可以使用 Spring Integration 2.2 添加的事务同步功能来做到这一点，如以下示例所示：spring-doc.cadn.net.cn

<int-mongodb:inbound-channel-adapter id="mongoInboundAdapter"
    channel="replyChannel"
    query-expression="new BasicQuery('{''name'' : ''Bob''}').limit(100)"
    entity-class="java.lang.Object"
    auto-startup="false">
        <int:poller fixed-rate="200" max-messages-per-poll="1">
            <int:transactional synchronization-factory="syncFactory"/>
        </int:poller>
</int-mongodb:inbound-channel-adapter>

<int:transaction-synchronization-factory id="syncFactory">
    <int:after-commit
        expression="@documentCleaner.remove(#mongoTemplate, payload, headers.mongo_collectionName)"
        channel="someChannel"/>
</int:transaction-synchronization-factory>

<bean id="documentCleaner" class="thing1.thing2.DocumentCleaner"/>

<bean id="transactionManager" class="o.s.i.transaction.PseudoTransactionManager"/>

以下示例显示了DocumentCleaner在前面的示例中引用：spring-doc.cadn.net.cn

public class DocumentCleaner {
    public void remove(MongoOperations mongoOperations, Object target, String collectionName) {
        if (target instanceof List<?> documents){
            for (Object document : documents) {
                mongoOperations.remove(new BasicQuery(JSON.serialize(document)), collectionName);
            }
        }
    }
}

您可以使用transactional元素。此元素可以引用真实的事务管理器（例如，如果流的其他部分调用 JDBC）。如果您没有“真实”事务，则可以使用o.s.i.transaction.PseudoTransactionManager，这是 Spring 的PlatformTransactionManager并在没有实际事务时启用 Mongo 适配器的事务同步功能。spring-doc.cadn.net.cn

这样做不会使 MongoDB 本身成为事务性的。它允许在成功（提交）之前或之后或失败（回滚）之后执行作同步。

一旦轮询器是事务性的，您就可以设置o.s.i.transaction.TransactionSynchronizationFactory在transactional元素。一个TransactionSynchronizationFactory创建TransactionSynchronization. 为了您的方便，我们公开了一个默认的基于 SpELTransactionSynchronizationFactory，允许您配置 SpEL 表达式，并使其执行与事务协调（同步）。支持提交前、提交后和回滚后事件的表达式，以及发送评估结果（如果有）的每个事件的通道。对于每个子元素，您可以指定expression和channel属性。如果只有channel属性存在，则接收到的消息将作为特定同步方案的一部分发送到那里。如果只有expression属性存在，并且表达式的结果是非空值，则生成一条消息，其结果为有效负载并发送到默认通道（NullChannel）并出现在日志中（在DEBUG水平）。如果希望评估结果转到特定通道，请添加channel属性。如果表达式的结果为 null 或 void，则不会生成任何消息。spring-doc.cadn.net.cn

有关事务同步的详细信息，请参阅事务同步。spring-doc.cadn.net.cn

从 5.5 版本开始，MongoDbMessageSource可以配置为updateExpression，其计算结果必须为String使用 MongoDbupdate语法或转换为org.springframework.data.mongodb.core.query.Update实例。它可以用作上述后处理过程的替代方法，它修改从集合中提取的实体，因此它们不会在下一个轮询周期中再次从集合中拉取（假设更新更改了查询中使用的某些值）。还是建议使用事务来实现执行隔离和数据一致性，当多个实例的MongoDbMessageSource对于集群中使用的相同集合。spring-doc.cadn.net.cn

MongoDB 更改流入站通道适配器

从 5.3 版开始，spring-integration-mongodb模块引入了MongoDbChangeStreamMessageProducer- 反应性MessageProducerSupportSpring Data 的实现ReactiveMongoOperations.changeStream(String, ChangeStreamOptions, Class)应用程序接口。此组件生成一个Flux的消息，带有body之ChangeStreamEvent默认情况下作为有效负载和一些与流相关的标头（请参阅MongoHeaders). 建议MongoDbChangeStreamMessageProducer与FluxMessageChannel作为outputChannel用于下游的按需订阅和事件消费。spring-doc.cadn.net.cn

此通道适配器的 Java DSL 配置可能如下所示：spring-doc.cadn.net.cn

@Bean
IntegrationFlow changeStreamFlow(ReactiveMongoOperations mongoTemplate) {
    return IntegrationFlow.from(
            MongoDb.changeStreamInboundChannelAdapter(mongoTemplate)
                    .domainType(Person.class)
                    .collection("person")
                    .extractBody(false))
            .channel(MessageChannels.flux())
            .get();
}

当MongoDbChangeStreamMessageProducer停止，或者订阅在下游取消，或者 MongoDb 更改流生成OperationType.INVALIDATE这Publisher已完成。通道适配器可以重新启动，并且新的Publisher的源数据被创建，并在MessageProducerSupport.subscribeToPublisher(Publisher<? extends Message<?>>). 如果需要从其他地方使用更改流事件，则可以在启动之间重新配置此通道适配器以获得新选项。spring-doc.cadn.net.cn

请参阅 Spring Data MongoDb 文档中有关更改流支持的更多信息。spring-doc.cadn.net.cn

MongoDB 出站通道适配器

MongoDB 出站通道适配器允许您将消息有效负载写入 MongoDB 文档存储，如以下示例所示：spring-doc.cadn.net.cn

<int-mongodb:outbound-channel-adapter id="fullConfigWithCollectionExpression"
	collection-name="myCollection"
	mongo-converter="mongoConverter"
	mongodb-factory="mongoDbFactory" />

如前面的配置所示，您可以使用outbound-channel-adapter元素，为各种属性提供值，例如：spring-doc.cadn.net.cn

collection-name或collection-name-expression：标识要使用的 MongoDb 集合的名称。spring-doc.cadn.net.cn
mongo-converter：对o.s.data.mongodb.core.convert.MongoConverter这有助于将原始 Java 对象转换为 JSON 文档表示。spring-doc.cadn.net.cn
mongodb-factory：对o.s.data.mongodb.MongoDbFactory.spring-doc.cadn.net.cn
mongo-template：对o.s.data.mongodb.core.MongoTemplate. 注意：您不能同时设置 mongo-template 和 mongodb-factory。spring-doc.cadn.net.cn
所有入站适配器中通用的其他属性（例如“channel”）。spring-doc.cadn.net.cn

前面的示例相对简单且静态，因为它具有collection-name. 有时，您可能需要根据某些条件在运行时更改此值。为此，请使用collection-name-expression，其中提供的表达式是任何有效的 SpEL 表达式。spring-doc.cadn.net.cn

MongoDB 出站网关

5.0 版本引入了 MongoDB 出站网关。它允许您通过向数据库的请求通道发送消息来查询数据库。然后，网关将响应发送到回复通道。可以使用消息有效负载和标头来指定查询和集合名称，如以下示例所示：spring-doc.cadn.net.cn

@SpringBootApplication
public class MongoDbJavaApplication {

    public static void main(String[] args) {
        new SpringApplicationBuilder(MongoDbJavaApplication.class)
            .web(false)
            .run(args);
    }

    @Autowired
    private MongoDbFactory;

    @Autowired
    private MongoConverter;


    @Bean
    public IntegrationFlow gatewaySingleQueryFlow() {
        return f -> f
                .handle(queryOutboundGateway())
                .channel(c -> c.queue("retrieveResults"));
    }

    private MongoDbOutboundGatewaySpec queryOutboundGateway() {
        return MongoDb.outboundGateway(this.mongoDbFactory, this.mongoConverter)
                .query("{name : 'Bob'}")
                .collectionNameFunction(m -> m.getHeaders().get("collection"))
                .expectSingleResult(true)
                .entityClass(Person.class);
    }

}

class MongoDbKotlinApplication {

    fun main(args: Array<String>) = runApplication<MongoDbKotlinApplication>(*args)

    @Autowired
    lateinit var mongoDbFactory: MongoDatabaseFactory

    @Autowired
    lateinit var mongoConverter: MongoConverter

    @Bean
    fun gatewaySingleQueryFlow() =
    integrationFlow {
        handle(queryOutboundGateway())
        channel { queue("retrieveResults") }
    }

    private fun queryOutboundGateway(): MongoDbOutboundGatewaySpec {
        return MongoDb.outboundGateway(this.mongoDbFactory, this.mongoConverter)
            .query("{name : 'Bob'}")
            .collectionNameFunction<Any> { m -> m.headers["collection"] as String }
            .expectSingleResult(true)
            .entityClass(Person::class.java)
    }

}

@SpringBootApplication
public class MongoDbJavaApplication {

    public static void main(String[] args) {
        new SpringApplicationBuilder(MongoDbJavaApplication.class)
            .web(false)
            .run(args);
    }

    @Autowired
    private MongoDbFactory mongoDbFactory;

    @Bean
    @ServiceActivator(inputChannel = "requestChannel")
    public MessageHandler mongoDbOutboundGateway() {
        MongoDbOutboundGateway gateway = new MongoDbOutboundGateway(this.mongoDbFactory);
        gateway.setCollectionNameExpressionString("'myCollection'");
        gateway.setQueryExpressionString("'{''name'' : ''Bob''}'");
        gateway.setExpectSingleResult(true);
        gateway.setEntityClass(Person.class);
        gateway.setOutputChannelName("replyChannel");
        return gateway;
    }

    @Bean
    @ServiceActivator(inputChannel = "replyChannel")
    public MessageHandler handler() {
        return message -> System.out.println(message.getPayload());
    }
}

<int-mongodb:outbound-gateway id="gatewayQuery"
    mongodb-factory="mongoDbFactory"
    mongo-converter="mongoConverter"
    query="{firstName: 'Bob'}"
    collection-name="myCollection"
    request-channel="in"
    reply-channel="out"
    entity-class="org.springframework.integration.mongodb.test.entity$Person"/>

您可以将以下属性与 MongoDB 出站网关一起使用：spring-doc.cadn.net.cn

collection-name或collection-name-expression：标识要使用的 MongoDB 集合的名称。spring-doc.cadn.net.cn
mongo-converter：对o.s.data.mongodb.core.convert.MongoConverter这有助于将原始 Java 对象转换为 JSON 文档表示。spring-doc.cadn.net.cn
mongodb-factory：对o.s.data.mongodb.MongoDbFactory.spring-doc.cadn.net.cn
mongo-template：对o.s.data.mongodb.core.MongoTemplate. 注意：您不能同时设置两者mongo-template和mongodb-factory.spring-doc.cadn.net.cn
entity-class：要传递给find(..)和findOne(..)MongoTemplate 中的方法。如果未提供此属性，则默认值为org.bson.Document.spring-doc.cadn.net.cn
query或query-expression：指定 MongoDB 查询。有关更多查询示例，请参阅 MongoDB 文档。spring-doc.cadn.net.cn
collection-callback：对org.springframework.data.mongodb.core.CollectionCallback. 优选的实例o.s.i.mongodb.outbound.MessageCollectionCallback从 5.0.11 开始，使用请求消息上下文。有关更多信息，请参阅其 Javadocs。注意：你不能同时拥有两者collection-callback以及任何查询属性。spring-doc.cadn.net.cn

作为query和query-expression属性，您可以使用collectionCallback属性作为对MessageCollectionCallback功能接口实现。以下示例指定计数作：spring-doc.cadn.net.cn

private MongoDbOutboundGatewaySpec collectionCallbackOutboundGateway() {
    return MongoDb.outboundGateway(this.mongoDbFactory, this.mongoConverter)
            .collectionCallback((collection, requestMessage) -> collection.count())
            .collectionName("myCollection");
}

MongoDB 响应式通道适配器

从 5.3 版开始，ReactiveMongoDbStoringMessageHandler和ReactiveMongoDbMessageSource提供了实现。它们基于ReactiveMongoOperations来自 Spring Data，并且需要一个org.mongodb:mongodb-driver-reactivestreamsDependency。spring-doc.cadn.net.cn

这ReactiveMongoDbStoringMessageHandler是ReactiveMessageHandler当集成流定义中涉及响应式流组合时，框架中原生支持该功能。请参阅 ReactiveMessageHandler 中的详细信息。spring-doc.cadn.net.cn

从配置的角度来看，与许多其他标准通道适配器没有区别。例如，对于 Java DSL，这样的通道适配器可以像这样使用：spring-doc.cadn.net.cn

@Bean
public IntegrationFlow reactiveMongoDbFlow(ReactiveMongoDatabaseFactory mongoDbFactory) {
    return f -> f
            .channel(MessageChannels.flux())
            .handle(MongoDb.reactiveOutboundChannelAdapter(mongoDbFactory));
}

在此示例中，我们将通过提供的 MongoDb 连接到 MongoDbReactiveMongoDatabaseFactory并将请求消息中的数据存储到默认集合中，并使用data名字。实际作将从内部创建的响应式流组合中按需执行ReactiveStreamsConsumer.spring-doc.cadn.net.cn

这ReactiveMongoDbMessageSource是一个AbstractMessageSource基于提供的ReactiveMongoDatabaseFactory或ReactiveMongoOperations和 MongoDb 查询（或表达式），调用find()或findOne()根据expectSingleResult选项，预期的entityClass类型以转换查询结果。查询执行和结果评估在以下情况下按需执行Publisher (Flux或Mono根据expectSingleResult选项）在生成消息的有效负载中被订阅。框架可以自动订阅这样的有效负载（本质上是flatMap）当分路器和FluxMessageChannel在下游使用。否则，目标应用程序负责订阅下游端点中轮询的发布者。spring-doc.cadn.net.cn

使用 Java DSL，可以这样配置这样的通道适配器：spring-doc.cadn.net.cn

@Bean
public IntegrationFlow reactiveMongoDbFlow(ReactiveMongoDatabaseFactory mongoDbFactory) {
    return IntegrationFlow
            .from(MongoDb.reactiveInboundChannelAdapter(mongoDbFactory, "{'name' : 'Name'}")
                            .entityClass(Person.class),
                    c -> c.poller(Pollers.fixedDelay(1000)))
            .split()
            .channel(c -> c.flux("output"))
            .get();
}

从 5.5 版本开始，ReactiveMongoDbMessageSource可以配置为updateExpression. 它具有与阻止相同的功能MongoDbMessageSource. 请参阅 MongoDB 入站通道适配器和AbstractMongoDbMessageSourceSpecJavaDocs 了解更多信息。spring-doc.cadn.net.cn

MongoDb 支持

连接到 MongoDb

阻塞还是反应？

用MongoDatabaseFactory

用ReactiveMongoDatabaseFactory

MongoDB 消息存储

MongoDB 通道消息存储

使用 AbstractConfigurableMongoDbMessageStore 禁用自动索引创建

MongoDB 元数据存储

MongoDB 入站通道适配器

MongoDB 更改流入站通道适配器

MongoDB 出站通道适配器

MongoDB 出站网关

MongoDB 响应式通道适配器

用`MongoDatabaseFactory`

用`ReactiveMongoDatabaseFactory`