通道适配器 (Channel Adapter)
通道适配器是一种消息端点,用于将单个发送方或接收方连接到消息通道。Spring Integration 提供了许多适配器来支持各种传输方式,例如 JMS、文件、HTTP、Web 服务、邮件等等。本参考指南的后续章节将讨论每个适配器。然而,本章重点介绍简单但灵活的方法调用通道适配器支持。有入站和出站适配器,每个都可以使用核心命名空间中提供的 XML 元素进行配置。只要您有一个可以作为源或目标调用的方法,这些适配器就提供了一种扩展 Spring Integration 的简便方法。
配置入站通道适配器
inbound-channel-adapter
元素(在 Java 配置中是 SourcePollingChannelAdapter
)可以调用 Spring 管理对象上的任何方法,并将非空返回值转换为 Message
后发送到 MessageChannel
。当适配器的订阅被激活时,轮询器会尝试从源接收消息。轮询器会根据提供的配置使用 TaskScheduler
进行调度。要配置单个通道适配器的轮询间隔或 cron 表达式,您可以提供一个带有调度属性(如 'fixed-rate' 或 'cron')的 'poller' 元素。以下示例定义了两个 inbound-channel-adapter
实例
-
Java DSL
-
Java
-
Kotlin DSL
-
XML
@Bean
public IntegrationFlow source1() {
return IntegrationFlow.from(() -> new GenericMessage<>(...),
e -> e.poller(p -> p.fixedRate(5000)))
...
.get();
}
@Bean
public IntegrationFlow source2() {
return IntegrationFlow.from(() -> new GenericMessage<>(...),
e -> e.poller(p -> p.cron("30 * 9-17 * * MON-FRI")))
...
.get();
}
public class SourceService {
@InboundChannelAdapter(channel = "channel1", poller = @Poller(fixedRate = "5000"))
Object method1() {
...
}
@InboundChannelAdapter(channel = "channel2", poller = @Poller(cron = "30 * 9-17 * * MON-FRI"))
Object method2() {
...
}
}
@Bean
fun messageSourceFlow() =
integrationFlow( { GenericMessage<>(...) },
{ poller { it.fixedRate(5000) } }) {
...
}
<int:inbound-channel-adapter ref="source1" method="method1" channel="channel1">
<int:poller fixed-rate="5000"/>
</int:inbound-channel-adapter>
<int:inbound-channel-adapter ref="source2" method="method2" channel="channel2">
<int:poller cron="30 * 9-17 * * MON-FRI"/>
</int:channel-adapter>
另请参阅通道适配器表达式和脚本。
如果未提供轮询器,则必须在上下文中注册一个默认轮询器。有关更多详细信息,请参阅端点命名空间支持。 |
轮询端点的默认触发器是具有 1 秒固定延迟周期的 PeriodicTrigger 实例。 |
重要:轮询器配置
所有
在第一个配置中,轮询任务每次轮询调用一次,并且在每个任务(轮询)期间,根据
请注意,未指定 然而,在 然而,如果您确定您的方法可以返回 null,并且需要在每次轮询时轮询尽可能多的可用源,则应明确将
从 5.5 版本开始, 从 6.2 版本开始, 另请参阅全局默认轮询器以获取更多信息。 |
配置出站通道适配器
outbound-channel-adapter
元素(在 Java 配置中为 @ServiceActivator
)也可以将 MessageChannel
连接到任何 POJO 消费者方法,该方法应使用发送到该通道的消息的载荷进行调用。以下示例显示了如何定义出站通道适配器
-
Java DSL
-
Java
-
Kotlin DSL
-
XML
@Bean
public IntegrationFlow outboundChannelAdapterFlow(MyPojo myPojo) {
return f -> f
.handle(myPojo, "handle");
}
public class MyPojo {
@ServiceActivator(channel = "channel1")
void handle(Object payload) {
...
}
}
@Bean
fun outboundChannelAdapterFlow(myPojo: MyPojo) =
integrationFlow {
handle(myPojo, "handle")
}
<int:outbound-channel-adapter channel="channel1" ref="target" method="handle"/>
<beans:bean id="target" class="org.MyPojo"/>
如果被适配的通道是 PollableChannel
,则必须提供一个 poller 子元素(在 @ServiceActivator
上使用 @Poller
子注解),如下例所示
-
Java
-
XML
public class MyPojo {
@ServiceActivator(channel = "channel1", poller = @Poller(fixedRate = "3000"))
void handle(Object payload) {
...
}
}
<int:outbound-channel-adapter channel="channel2" ref="target" method="handle">
<int:poller fixed-rate="3000" />
</int:outbound-channel-adapter>
<beans:bean id="target" class="org.MyPojo"/>
如果 POJO 消费者实现可以在其他 <outbound-channel-adapter>
定义中重用,则应使用 ref
属性。但是,如果消费者实现仅由 <outbound-channel-adapter>
的单个定义引用,则可以将其定义为内部 bean,如下例所示
<int:outbound-channel-adapter channel="channel" method="handle">
<beans:bean class="org.Foo"/>
</int:outbound-channel-adapter>
在同一 <outbound-channel-adapter> 配置中同时使用 ref 属性和内部处理程序定义是不允许的,因为它会创建模糊条件。这种配置会导致抛出异常。 |
任何通道适配器都可以不带 channel
引用创建,在这种情况下,它会隐式创建一个 DirectChannel
实例。创建的通道的名称与 <inbound-channel-adapter>
或 <outbound-channel-adapter>
元素的 id
属性匹配。因此,如果未提供 channel
,则 id
是必需的。
通道适配器表达式和脚本
与许多其他 Spring Integration 组件一样,<inbound-channel-adapter>
和 <outbound-channel-adapter>
也支持 SpEL 表达式求值。要使用 SpEL,请在 'expression' 属性中提供表达式字符串,而不是提供用于在 bean 上进行方法调用的 'ref' 和 'method' 属性。当表达式求值时,它遵循与方法调用相同的契约:对于 <inbound-channel-adapter>
,表达式求值结果为非空值时会生成一条消息;而对于 <outbound-channel-adapter>
,表达式必须等同于返回 void 的方法调用。
从 Spring Integration 3.0 开始,<int:inbound-channel-adapter/>
也可以使用 SpEL <expression/>
(甚至使用 <script/>
)子元素进行配置,以应对需要比简单 'expression' 属性更复杂的情况。如果通过使用 location
属性将脚本提供为 Resource
,您还可以设置 refresh-check-delay
,这允许定期刷新资源。如果您希望在每次轮询时检查脚本,则需要协调此设置与轮询器的触发器,如下例所示
<int:inbound-channel-adapter ref="source1" method="method1" channel="channel1">
<int:poller max-messages-per-poll="1" fixed-delay="5000"/>
<script:script lang="ruby" location="Foo.rb" refresh-check-delay="5000"/>
</int:inbound-channel-adapter>
另请参阅使用 <expression/>
子元素时 ReloadableResourceBundleExpressionSource
上的 cacheSeconds
属性。有关表达式的更多信息,请参阅Spring Expression Language (SpEL)。有关脚本的信息,请参阅Groovy 支持和脚本支持。
<int:inbound-channel-adapter/> (SourcePollingChannelAdapter ) 是一个端点,它通过定期触发以轮询底层 MessageSource 来启动消息流。由于在轮询时没有消息对象,表达式和脚本无法访问根 Message ,因此不像大多数其他消息处理 SpEL 表达式那样具有可用的 payload 或 headers 属性。脚本可以生成并返回包含 headers 和 payload 的完整 Message 对象,或者只返回一个 payload,该 payload 将由框架添加到包含基本 headers 的消息中。 |