Apache Cassandra 支持 (Apache Cassandra Support) _ Spring Integration6.3.2-SNAPSHOT中文文档(官方文档中文翻译)|Spring 教程 —

此版本仍在开发中，尚未被视为稳定版本。对于最新的稳定版本，请使用 Spring Integration 6.3.1！

Spring Integration 提供了通道适配器（从 6.0 版开始），用于对 Apache Cassandra 集群执行数据库操作。它完全基于 Spring Data for Apache Cassandra 项目。

您需要将此依赖项包含在项目中：

Maven
Gradle

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-cassandra</artifactId>
    <version>6.3.2-SNAPSHOT</version>
</dependency>

compile "org.springframework.integration:spring-integration-cassandra:6.3.2-SNAPSHOT"

Cassandra 出站组件

这是一个实现，可以在单向（默认）和请求-回复模式（一个选项）下工作。默认情况下，它是异步的（用于重置），并对提供的 . 可以通过该选项配置操作类型。将模式设置为 ;或或将模式设置为。CassandraMessageHandlerAbstractReplyProducingMessageHandlerproducesReplysetAsync(false)INSERTUPDATEDELETESTATEMENTReactiveCassandraOperationsCassandraMessageHandler.TypeingestQueryINSERTquerystatementExpressionstatementProcessorSTATEMENT

以下代码片段演示了此通道适配器或网关的各种配置：

Java DSL
Kotlin DSL
Java
XML

@Bean
IntegrationFlow cassandraSelectFlow(ReactiveCassandraOperations cassandraOperations) {
    return flow -> flow
            .handle(Cassandra.outboundGateway(cassandraOperations)
                    .query("SELECT * FROM book WHERE author = :author limit :size")
                    .parameter("author", "payload")
                    .parameter("size", m -> m.getHeaders().get("limit")))
            .channel(c -> c.flux("resultChannel"));
}

@Bean
fun outboundReactive(cassandraOperations: ReactiveCassandraOperations) =
    integrationFlow {
        handle(
            Cassandra.outboundChannelAdapter(cassandraOperations)
                              .statementExpression("T(QueryBuilder).truncate('book').build()")
        ) { async(false) }
    }

@ServiceActivator(inputChannel = "cassandraSelectChannel")
@Bean
public MessageHandler cassandraMessageHandler() {
    CassandraMessageHandler cassandraMessageHandler = new CassandraMessageHandler(this.template);
    cassandraMessageHandler.setQuery("SELECT * FROM book WHERE author = :author limit :size");

    Map<String, Expression> params = new HashMap<>();
    params.put("author", PARSER.parseExpression("payload"));
    params.put("size", PARSER.parseExpression("headers.limit"));

    cassandraMessageHandler.setParameterExpressions(params);

    cassandraMessageHandler.setOutputChannel(resultChannel());
    cassandraMessageHandler.setProducesReply(true);
    return cassandraMessageHandler;
}

<int-cassandra:outbound-channel-adapter id="outboundAdapter"
                                        cassandra-template="cassandraTemplate"
                                        write-options="writeOptions"
                                        auto-startup="false"
                                        async="false"/>

<int-cassandra:outbound-gateway id="outgateway"
                                request-channel="input"
                                cassandra-template="cassandraTemplate"
                                mode="STATEMENT"
                                write-options="writeOptions"
                                query="SELECT * FROM book limit :size"
                                reply-channel="resultChannel"
                                auto-startup="true">
    <int-cassandra:parameter-expression name="author" expression="payload"/>
    <int-cassandra:parameter-expression name="size" expression="headers.limit"/>
</int-cassandra:outbound-gateway>

如果在默认异步模式下将 a 用作网关，则会生成 a，该网关根据提供的实现进行处理。对于真正的反应式处理，建议将 a 用于输出通道配置。调用同步模式以获取应答值。CassandraMessageHandlerMono<WriteResult>MessageChannelFluxMessageChannelMono.block()

如果执行或操作，则请求消息有效负载中应有一个实体（已标记）。如果有效负载是实体列表，则执行相应的批处理操作。INSERTUPDATEDELETEorg.springframework.data.cassandra.core.mapping.Table

该模式要求有效负载以要插入 - 的值矩阵形式存在。例如，如果实体是这样的：ingestQueryList<List<?>>

@Table("book")
public record Book(@PrimaryKey String isbn,
                   String title,
                   @Indexed String author,
                   int pages,
                   LocalDate saleDate,
                   boolean isInStock) {

}

通道适配器具有以下配置：

@Bean
public MessageHandler cassandraMessageHandler3() {
    CassandraMessageHandler cassandraMessageHandler = new CassandraMessageHandler(this.template);
    String cqlIngest = "insert into book (isbn, title, author, pages, saleDate, isInStock) values (?, ?, ?, ?, ?, ?)";
    cassandraMessageHandler.setIngestQuery(cqlIngest);
    cassandraMessageHandler.setAsync(false);
    return cassandraMessageHandler;
}

请求消息有效负载必须按如下方式转换：

List<List<Object>> ingestBooks =
    payload.stream()
            .map(book ->
                    List.<Object>of(
                            book.isbn(),
                            book.title(),
                            book.author(),
                            book.pages(),
                            book.saleDate(),
                            book.isInStock()))
            .toList();

对于更复杂的用例，有效负载可以作为的实例。建议使用 API 构建各种语句以针对 Apache Cassandra 执行。例如，要从表中删除所有数据，可以将具有类似此类有效负载的消息发送到：。或者，对于基于请求消息的逻辑，可以提供 or 用于基于该消息构建 a。为方便起见，将 a 注册为 SpEL 评估上下文，因此目标表达式可以如此简单：com.datastax.oss.driver.api.core.cql.Statementcom.datastax.oss.driver.api.querybuilder.QueryBuilderBookCassandraMessageHandlerQueryBuilder.truncate("book").build()statementExpressionstatementProcessorCassandraMessageHandlerStatementcom.datastax.oss.driver.api.querybuilderimport

statement-expression="T(QueryBuilder).selectFrom("book").all()"

表示可绑定的命名查询参数，并且仅与选项一起使用。请参阅上面提到的 Java 和 XML 示例。setParameterExpressions(Map<String, Expression> parameterExpressions)setQuery(String query)