Apache Cassandra 支持
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.4.4</version>
</dependency>
compile "org.springframework.integration:spring-integration-cassandra:6.4.4"
Cassandra 出站组件
CassandraMessageHandler
是 AbstractReplyProducingMessageHandler
的一个实现,可以工作在单向(默认)和请求-回复两种模式下(通过 producesReply
选项配置)。它默认是异步的(设置 setAsync(false)
可重置为同步),并对提供的 ReactiveCassandraOperations
执行响应式的 INSERT
、UPDATE
、DELETE
或 STATEMENT
操作。操作类型可以通过 CassandraMessageHandler.Type
选项配置。ingestQuery
将模式设置为 INSERT
;query
、statementExpression
或 statementProcessor
将模式设置为 STATEMENT
。
以下代码片段展示了此通道适配器或网关的各种配置
-
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>
如果 CassandraMessageHandler
在默认异步模式下用作网关,它会产生一个 Mono<WriteResult>
,该结果根据提供的 MessageChannel
实现进行处理。对于真正的响应式处理,建议将输出通道配置为 FluxMessageChannel
。在同步模式下,会调用 Mono.block()
来获取回复值。
如果执行 INSERT
、UPDATE
或 DELETE
操作,则期望请求消息载荷中包含一个实体(标记为 org.springframework.data.cassandra.core.mapping.Table
)。如果载荷是实体列表,则执行相应的批处理操作。
ingestQuery
模式期望载荷是以值矩阵的形式出现,用于插入 - List<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();
对于更复杂的用例,载荷可以是 com.datastax.oss.driver.api.core.cql.Statement
的实例。建议使用 com.datastax.oss.driver.api.querybuilder.QueryBuilder
API 来构建各种语句以针对 Apache Cassandra 执行。例如,要从 Book
表中删除所有数据,可以将一个载荷如下所示的消息发送到 CassandraMessageHandler
:QueryBuilder.truncate("book").build()
。或者,对于基于请求消息的逻辑,可以为 CassandraMessageHandler
提供 statementExpression
或 statementProcessor
,以便基于该消息构建 Statement
。为了方便起见,com.datastax.oss.driver.api.querybuilder
被注册为 SpEL 求值上下文中的一个 import
,因此目标表达式可以像这样简单
statement-expression="T(QueryBuilder).selectFrom("book").all()"
setParameterExpressions(Map<String, Expression> parameterExpressions)
表示可绑定的命名查询参数,并且仅与 setQuery(String query)
选项一起使用。请参阅上面提到的 Java 和 XML 示例。