R2DBC 支持

Spring 集成通过 R2DBC 驱动程序使用对数据库的反应式访问来提供用于接收和发送消息的通道适配器。spring-doc.cadn.net.cn

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

Maven 系列

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-r2dbc</artifactId>
    <version>6.0.9</version>
</dependency>

Gradle

compile "org.springframework.integration:spring-integration-r2dbc:6.0.9"

R2DBC 入站通道适配器

这R2dbcMessageSource是 pollableMessageSource基于R2dbcEntityOperations生成带有Flux或Mono作为根据expectSingleResult选择。对SELECT可以是静态提供的，也可以基于在每个receive()叫。这R2dbcMessageSource.SelectCreator作为 Quote Context 的根对象存在，以允许使用StatementMapper.SelectSpecFluent API 的 API 创建。默认情况下，此通道适配器将记录从 select 映射到LinkedCaseInsensitiveMap实例。它可以进行定制，提供payloadTypeoptions 的 API 中，由EntityRowMapper基于this.r2dbcEntityOperations.getConverter(). 这updateSql是可选的，用于标记数据库中的已读记录，以便从后续轮询中跳过。这UPDATE作可以随BiFunction<DatabaseClient.GenericExecuteSpec, ?, DatabaseClient.GenericExecuteSpec>要将值绑定到UPDATE基于SELECT结果。spring-doc.cadn.net.cn

此 channel adapter 的典型配置可能如下所示：spring-doc.cadn.net.cn

@Bean
@InboundChannelAdapter("fromR2dbcChannel")
public R2dbcMessageSource r2dbcMessageSourceSelectMany() {
    R2dbcMessageSource r2dbcMessageSource = new R2dbcMessageSource(this.r2dbcEntityTemplate,
            "SELECT * FROM person WHERE name='Name'");
    r2dbcMessageSource.setPayloadType(Person.class);
    r2dbcMessageSource.setUpdateSql("UPDATE Person SET name='SomeOtherName' WHERE id = :id");
    r2dbcMessageSource.setBindFunction(
				(DatabaseClient.GenericExecuteSpec bindSpec, Person o) -> bindSpec.bind("id", o.getId()));}
    return r2dbcMessageSource;
}

使用 Java DSL 时，此通道适配器的配置如下所示：spring-doc.cadn.net.cn

@Bean
IntegrationFlow r2dbcDslFlow(R2dbcEntityTemplate r2dbcEntityTemplate) {
    return IntegrationFlow
        .from(R2dbc.inboundChannelAdapter(r2dbcEntityTemplate,
            (selectCreator) ->
                    selectCreator.createSelect("person")
                        .withProjection("*")
                        .withCriteria(Criteria.where("id").is(1)))
                    .expectSingleResult(true)
                    .payloadType(Person.class)
                    .updateSql("UPDATE Person SET id='2' where id = :id")
                    .bindFunction((DatabaseClient.GenericExecuteSpec bindSpec, Person o) ->
                            bindSpec.bind("id", o.getId())),
            e -> e.poller(p -> p.fixedDelay(100)))
        .<Mono<?>>handle((p, h) -> p, e -> e.async(true))
        .channel(MessageChannels.flux())
        .get();
}

R2DBC 出站通道适配器

这R2dbcMessageHandler是一个ReactiveMessageHandlerimplementation 来执行INSERT（默认）、UPDATE或DELETE查询R2dbcEntityOperations. 这R2dbcMessageHandler.Type可以静态配置，也可以通过针对请求消息的 SPEL 表达式进行配置。要执行的查询可以基于tableName,values和criteria表达式选项或（如果tableName未提供），则整个消息有效负载将被视为org.springframework.data.relational.core.mapping.Table实体来执行 SQL。包装org.springframework.data.relational.core.query注册为导入到 SPEL 评估上下文中，以便直接访问CriteriaFluent API，用于UPDATE和DELETE查询。这valuesExpression用于INSERT和UPDATE，并且必须评估为Mapfor column-value pairs 来根据请求消息在目标表中执行更改。spring-doc.cadn.net.cn

此 channel adapter 的典型配置可能如下所示：spring-doc.cadn.net.cn

@Bean
@ServiceActivator(inputChannel = "toR2dbcChannel")
public R2dbcMessageHandler r2dbcMessageHandler(R2dbcEntityTemplate r2dbcEntityTemplate) {
    R2dbcMessageHandler messageHandler = new R2dbcMessageHandler(r2dbcEntityTemplate)
    messageHandler.setValuesExpression(new FunctionExpression<Message<?>>(Message::getPayload));
    messageHandler.setQueryType(R2dbcMessageHandler.Type.UPDATE);
    messageHandler.setCriteriaExpression(
        EXPRESSION_PARSER.parseExpression("T(Criteria).where('id).is(headers.personId)));
    return messageHandler;
}

使用 Java DSL 时，此通道适配器的配置如下所示：spring-doc.cadn.net.cn

.handle(R2dbc.outboundChannelAdapter(r2dbcEntityTemplate)
        .queryType(R2dbcMessageHandler.Type.UPDATE)
        .tableNameExpression("payload.class.simpleName")
        .criteria((message) -> Criteria.where("id").is(message.getHeaders().get("personId")))
        .values("{age:36}"))