出站网关
JPA 入站通道适配器允许您轮询数据库以检索一个或多个 JPA 实体。检索到的数据随后用于启动一个 Spring Integration 流,该流将检索到的数据用作消息载荷。
此外,您可以在流的末尾使用 JPA 出站通道适配器来持久化数据,这实际上会在持久化操作结束时停止流。
但是,如何在流的中间执行 JPA 持久化操作呢?例如,您可能在 Spring Integration 消息流中处理业务数据,并且希望将其持久化,但您仍然需要进一步下游使用其他组件。或者,除了使用轮询器轮询数据库外,您还需要执行 JPQL 查询并主动检索数据,这些数据随后在流中的后续组件中进行处理。
这时,JPA 出站网关就派上用场了。它们使您能够持久化数据以及检索数据。为了方便这些用途,Spring Integration 提供了两种类型的 JPA 出站网关
-
更新出站网关
-
检索出站网关
每当出站网关用于执行保存、更新或仅删除数据库中某些记录的操作时,您都需要使用更新出站网关。例如,如果您使用一个 entity
来持久化它,结果会返回一个已合并并持久化的实体。在其他情况下,则返回受影响的记录数(更新或删除)。
从数据库中检索(选择)数据时,我们使用检索出站网关。通过检索出站网关,我们可以使用 JPQL、命名查询(native 或基于 JPQL)或 native 查询(SQL)来选择数据并检索结果。
更新出站网关在功能上类似于出站通道适配器,不同之处在于更新出站网关在执行 JPA 操作后会将结果发送到网关的回复通道。
检索出站网关类似于入站通道适配器。
这种相似性是使用核心 JpaExecutor
类尽可能统一常见功能的主要因素。
所有 JPA 出站网关的共同点,类似于 outbound-channel-adapter
,我们可以用它们来执行各种 JPA 操作
-
实体类
-
JPA 查询语言 (JPQL)
-
原生查询
-
命名查询
有关配置示例,请参阅JPA 出站网关示例。
通用配置参数
JPA 出站网关始终可以访问 Spring Integration Message
作为输入。因此,以下参数可用
parameter-source-factory
-
o.s.i.jpa.support.parametersource.ParameterSourceFactory
的一个实例,用于获取o.s.i.jpa.support.parametersource.ParameterSource
的实例。ParameterSource
用于解析查询中提供的参数值。如果您使用 JPA 实体执行操作,则会忽略parameter-source-factory
属性。parameter
子元素与parameter-source-factory
互斥,必须在提供的ParameterSourceFactory
上配置。可选。 use-payload-as-parameter-source
-
如果设置为
true
,则使用Message
的载荷作为参数的来源。如果设置为false
,则整个Message
可作为参数的来源。如果没有传入 JPA 参数,此属性默认为true
。这意味着,如果您使用默认的BeanPropertyParameterSourceFactory
,载荷的 bean 属性将用作 JPA 查询参数值的来源。但是,如果传入了 JPA 参数,此属性默认评估为false
。原因是 JPA 参数允许您提供 SpEL 表达式。因此,能够访问整个Message
(包括头部)是非常有益的。可选。
更新出站网关
以下列表显示了您可以在更新出站网关上设置的所有属性,并描述了关键属性
<int-jpa:updating-outbound-gateway request-channel="" (1)
auto-startup="true"
entity-class=""
entity-manager=""
entity-manager-factory=""
id=""
jpa-operations=""
jpa-query=""
named-query=""
native-query=""
order=""
parameter-source-factory=""
persist-mode="MERGE"
reply-channel="" (2)
reply-timeout="" (3)
use-payload-as-parameter-source="true">
<int:poller/>
<int-jpa:transactional/>
<int-jpa:parameter name="" type="" value=""/>
<int-jpa:parameter name="" expression=""/>
</int-jpa:updating-outbound-gateway>
1 | 出站网关接收消息以执行所需操作的通道。此属性类似于 outbound-channel-adapter 的 channel 属性。可选。 |
2 | 网关在执行所需的 JPA 操作后发送响应的通道。如果未定义此属性,则请求消息必须具有 replyChannel 头部。可选。 |
3 | 指定网关等待发送结果到回复通道的时间。仅当回复通道本身可能阻塞发送操作时适用(例如,当前已满的有界 QueueChannel )。该值以毫秒为单位指定。可选。 |
使用 Java 配置进行配置
以下 Spring Boot 应用程序展示了如何使用 Java 配置出站适配器的示例
@SpringBootApplication
@EntityScan(basePackageClasses = StudentDomain.class)
@IntegrationComponentScan
public class JpaJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(JpaJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private EntityManagerFactory entityManagerFactory;
@MessagingGateway
interface JpaGateway {
@Gateway(requestChannel = "jpaUpdateChannel")
@Transactional
void updateStudent(StudentDomain payload);
}
@Bean
@ServiceActivator(channel = "jpaUpdateChannel")
public MessageHandler jpaOutbound() {
JpaOutboundGateway adapter =
new JpaOutboundGateway(new JpaExecutor(this.entityManagerFactory));
adapter.setOutputChannelName("updateResults");
return adapter;
}
}
使用 Java DSL 进行配置
以下 Spring Boot 应用程序展示了如何使用 Java DSL 配置出站适配器的示例
@SpringBootApplication
@EntityScan(basePackageClasses = StudentDomain.class)
public class JpaJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(JpaJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private EntityManagerFactory entityManagerFactory;
@Bean
public IntegrationFlow updatingGatewayFlow() {
return f -> f
.handle(Jpa.updatingGateway(this.entityManagerFactory),
e -> e.transactional(true))
.channel(c -> c.queue("updateResults"));
}
}
检索出站网关
以下示例演示了如何配置检索出站网关
-
Java DSL
-
Kotlin DSL
-
Java
-
XML
@SpringBootApplication
@EntityScan(basePackageClasses = StudentDomain.class)
public class JpaJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(JpaJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private EntityManagerFactory entityManagerFactory;
@Bean
public IntegrationFlow retrievingGatewayFlow() {
return f -> f
.handle(Jpa.retrievingGateway(this.entityManagerFactory)
.jpaQuery("from Student s where s.id = :id")
.expectSingleResult(true)
.parameterExpression("id", "payload"))
.channel(c -> c.queue("retrieveResults"));
}
}
@Bean
fun retrievingGatewayFlow() =
integrationFlow {
handle(Jpa.retrievingGateway(this.entityManagerFactory)
.jpaQuery("from Student s where s.id = :id")
.expectSingleResult(true)
.parameterExpression("id", "payload"))
channel { queue("retrieveResults") }
}
@SpringBootApplication
@EntityScan(basePackageClasses = StudentDomain.class)
public class JpaJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(JpaJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private EntityManagerFactory entityManagerFactory;
@Bean
public JpaExecutor jpaExecutor() {
JpaExecutor executor = new JpaExecutor(this.entityManagerFactory);
jpaExecutor.setJpaQuery("from Student s where s.id = :id");
executor.setJpaParameters(Collections.singletonList(new JpaParameter("id", null, "payload")));
jpaExecutor.setExpectSingleResult(true);
return executor;
}
@Bean
@ServiceActivator(channel = "jpaRetrievingChannel")
public MessageHandler jpaOutbound() {
JpaOutboundGateway adapter = new JpaOutboundGateway(jpaExecutor());
adapter.setOutputChannelName("retrieveResults");
adapter.setGatewayType(OutboundGatewayType.RETRIEVING);
return adapter;
}
}
<int-jpa:retrieving-outbound-gateway request-channel=""
auto-startup="true"
delete-after-poll="false"
delete-in-batch="false"
entity-class=""
id-expression="" (1)
entity-manager=""
entity-manager-factory=""
expect-single-result="false" (2)
id=""
jpa-operations=""
jpa-query=""
max-results="" (3)
max-results-expression="" (4)
first-result="" (5)
first-result-expression="" (6)
named-query=""
native-query=""
order=""
parameter-source-factory=""
reply-channel=""
reply-timeout=""
use-payload-as-parameter-source="true">
<int:poller></int:poller>
<int-jpa:transactional/>
<int-jpa:parameter name="" type="" value=""/>
<int-jpa:parameter name="" expression=""/>
</int-jpa:retrieving-outbound-gateway>
1 | (自 Spring Integration 4.0 起) 一个 SpEL 表达式,用于确定 EntityManager.find(Class entityClass, Object primaryKey) 方法的 primaryKey 值,将 requestMessage 作为评估上下文的根对象。如果存在 entity-class 属性,则 entityClass 参数由此确定。否则,由 payload 类确定。如果使用 id-expression ,则不允许使用所有其他属性。可选。 |
2 | 一个布尔标志,指示 select 操作期望返回单个结果还是结果的 List 。如果此标志设置为 true ,则单个实体作为消息的载荷发送。如果返回多个实体,则抛出异常。如果设置为 false ,则实体 List 作为消息的载荷发送。默认为 false 。可选。 |
3 | 此非零、非负整数值告诉适配器在执行 select 操作时,选择的行数不超过指定数量。默认情况下,如果未设置此属性,则给定查询会选择所有可能的记录。此属性与 max-results-expression 互斥。可选。 |
4 | 一个表达式,可用于查找结果集中的最大结果数。它与 max-results 互斥。可选。 |
5 | 此非零、非负整数值告诉适配器从哪个第一条记录开始检索结果。此属性与 first-result-expression 互斥。版本 3.0 引入了此属性。可选。 |
6 | 此表达式针对消息进行评估,以查找结果集中第一条记录的位置。此属性与 first-result 互斥。版本 3.0 引入了此属性。可选。 |
当您选择在检索后删除实体,并且您检索到了一组实体时,默认情况下,实体是逐个删除的。这可能会导致性能问题。 或者,您可以将属性 JSR 317: Java™ Persistence 2.0 在第 4.10 章“批量更新和删除操作”中指出 “删除操作仅适用于指定类及其子类的实体。它不级联到相关实体。” 更多信息,请参阅JSR 317: Java™ Persistence 2.0 |
从 6.0 版本开始,当查询没有返回任何实体时,Jpa.retrievingGateway() 会返回一个空列表结果。以前会返回 null ,从而结束流或抛出异常,具体取决于 requiresReply 。或者,要恢复以前的行为,可以在网关后添加一个 filter 来过滤掉空列表。在空列表处理是下游逻辑一部分的应用中,这需要额外的配置。有关可能的空列表处理选项,请参阅拆分器丢弃通道。 |
JPA 出站网关示例
本节包含使用更新出站网关和检索出站网关的各种示例
使用实体类进行更新
在以下示例中,通过使用 org.springframework.integration.jpa.test.entity.Student
实体类作为 JPA 定义参数来持久化更新出站网关
<int-jpa:updating-outbound-gateway request-channel="entityRequestChannel" (1)
reply-channel="entityResponseChannel" (2)
entity-class="org.springframework.integration.jpa.test.entity.Student"
entity-manager="em"/>
1 | 这是出站网关的请求通道。它类似于 outbound-channel-adapter 的 channel 属性。 |
2 | 这是网关与出站适配器的不同之处。这是接收 JPA 操作回复的通道。但是,如果您对收到的回复不感兴趣,只想执行操作,则使用 JPA outbound-channel-adapter 是合适的选择。在此示例中,我们使用实体类,回复是作为 JPA 操作结果创建或合并的实体对象。 |
使用 JPQL 进行更新
以下示例通过使用 Java Persistence Query Language (JPQL) 更新实体,这需要使用更新出站网关
<int-jpa:updating-outbound-gateway request-channel="jpaqlRequestChannel"
reply-channel="jpaqlResponseChannel"
jpa-query="update Student s set s.lastName = :lastName where s.rollNumber = :rollNumber" (1)
entity-manager="em">
<int-jpa:parameter name="lastName" expression="payload"/>
<int-jpa:parameter name="rollNumber" expression="headers['rollNumber']"/>
</int-jpa:updating-outbound-gateway>
1 | 网关执行的 JPQL 查询。由于我们使用了更新出站网关,只有 update 和 delete JPQL 查询是合理的选择。 |
当您发送一条消息,其载荷为 String
且包含一个名为 rollNumber
的头部,其值为 long
时,具有指定学号的学生的姓氏将更新为消息载荷中的值。使用更新网关时,返回值始终是一个整数值,表示执行 JPA QL 影响的记录数。
使用 JPQL 检索实体
以下示例使用检索出站网关和 JPQL 从数据库中检索(选择)一个或多个实体
<int-jpa:retrieving-outbound-gateway request-channel="retrievingGatewayReqChannel"
reply-channel="retrievingGatewayReplyChannel"
jpa-query="select s from Student s where s.firstName = :firstName and s.lastName = :lastName"
entity-manager="em">
<int-jpa:parameter name="firstName" expression="payload"/>
<int-jpa:parameter name="lastName" expression="headers['lastName']"/>
</int-jpa:outbound-gateway>
使用 id-expression
检索实体
以下示例使用带 id-expression
的检索出站网关从数据库中检索(查找)一个且仅一个实体:primaryKey
是 id-expression
评估的结果。entityClass
是消息 payload
的类。
<int-jpa:retrieving-outbound-gateway
request-channel="retrievingGatewayReqChannel"
reply-channel="retrievingGatewayReplyChannel"
id-expression="payload.id"
entity-manager="em"/>
使用命名查询进行更新
使用命名查询与直接使用 JPQL 查询基本相同。不同之处在于使用 named-query
属性代替,如下例所示
<int-jpa:updating-outbound-gateway request-channel="namedQueryRequestChannel"
reply-channel="namedQueryResponseChannel"
named-query="updateStudentByRollNumber"
entity-manager="em">
<int-jpa:parameter name="lastName" expression="payload"/>
<int-jpa:parameter name="rollNumber" expression="headers['rollNumber']"/>
</int-jpa:outbound-gateway>
您可以在此处找到使用 Spring Integration 的 JPA 适配器的完整示例应用程序。 |