如何定期发布消息到activemq spring 集成DSL
How to periodically publish a message to activemq spring integration DSL
我安装了 Active MQ 并定期说每 10 秒要发送一条消息到“my.queue”
我正在努力理解 Spring 集成 DSL 语言。
我需要类似的东西
IntegrationFlows.from(every 5 seconds)
.send(message to "my.queue")
是的,您可以使用 Spring 集成 Java DSL 及其 IntegrationFlow
抽象来做到这一点。要进行周期性任务,您需要在 IntegrationFlows
中使用此工厂来启动流程:
/**
* Provides {@link Supplier} as source of messages to the integration flow.
* which will be triggered by a <b>provided</b>
* {@link org.springframework.integration.endpoint.SourcePollingChannelAdapter}.
* @param messageSource the {@link Supplier} to populate.
* @param endpointConfigurer the {@link Consumer} to provide more options for the
* {@link org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean}.
* @param <T> the supplier type.
* @return new {@link IntegrationFlowBuilder}.
* @see Supplier
*/
public static <T> IntegrationFlowBuilder fromSupplier(Supplier<T> messageSource,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
Supplier
可能 return 一个您想要作为有效负载发送到下游的对象。第二个消费者参数可以配置为:
.poller(p -> p.fixedDelay(1000))
这样,每秒都会根据提供的有效负载创建一条消息并发送到下游。
要向 Active MQ 发送消息,您需要使用 org.springframework.integration.jms.dsl.Jms
及其针对相应通道适配器的方法:
/**
* The factory to produce a {@link JmsOutboundChannelAdapterSpec}.
* @param connectionFactory the JMS ConnectionFactory to build on
* @return the {@link JmsOutboundChannelAdapterSpec} instance
*/
public static JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware outboundAdapter(
ConnectionFactory connectionFactory) {
必须在 DSL 回调中使用此工厂的结果,例如:
/**
* Populate a {@link ServiceActivatingHandler} for the provided
* {@link MessageHandler} implementation.
* Can be used as Java 8 Lambda expression:
* <pre class="code">
* {@code
* .handle(m -> logger.info(m.getPayload())
* }
* </pre>
* @param messageHandler the {@link MessageHandler} to use.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B handle(MessageHandler messageHandler) {
所有信息都在文档中:https://docs.spring.io/spring-integration/reference/html/dsl.html#java-dsl
像这样:
@Bean
public IntegrationFlow jmsPeriodicFlow() {
return IntegrationFlows.fromSupplier(() -> "hello",
e -> e.poller(p -> p.fixedDelay(5000)))
.handle(Jms.outboundAdapter(jmsConnectionFactory())
.destination("my.queue"))
.get();
}
我安装了 Active MQ 并定期说每 10 秒要发送一条消息到“my.queue”
我正在努力理解 Spring 集成 DSL 语言。
我需要类似的东西
IntegrationFlows.from(every 5 seconds)
.send(message to "my.queue")
是的,您可以使用 Spring 集成 Java DSL 及其 IntegrationFlow
抽象来做到这一点。要进行周期性任务,您需要在 IntegrationFlows
中使用此工厂来启动流程:
/**
* Provides {@link Supplier} as source of messages to the integration flow.
* which will be triggered by a <b>provided</b>
* {@link org.springframework.integration.endpoint.SourcePollingChannelAdapter}.
* @param messageSource the {@link Supplier} to populate.
* @param endpointConfigurer the {@link Consumer} to provide more options for the
* {@link org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean}.
* @param <T> the supplier type.
* @return new {@link IntegrationFlowBuilder}.
* @see Supplier
*/
public static <T> IntegrationFlowBuilder fromSupplier(Supplier<T> messageSource,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
Supplier
可能 return 一个您想要作为有效负载发送到下游的对象。第二个消费者参数可以配置为:
.poller(p -> p.fixedDelay(1000))
这样,每秒都会根据提供的有效负载创建一条消息并发送到下游。
要向 Active MQ 发送消息,您需要使用 org.springframework.integration.jms.dsl.Jms
及其针对相应通道适配器的方法:
/**
* The factory to produce a {@link JmsOutboundChannelAdapterSpec}.
* @param connectionFactory the JMS ConnectionFactory to build on
* @return the {@link JmsOutboundChannelAdapterSpec} instance
*/
public static JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware outboundAdapter(
ConnectionFactory connectionFactory) {
必须在 DSL 回调中使用此工厂的结果,例如:
/**
* Populate a {@link ServiceActivatingHandler} for the provided
* {@link MessageHandler} implementation.
* Can be used as Java 8 Lambda expression:
* <pre class="code">
* {@code
* .handle(m -> logger.info(m.getPayload())
* }
* </pre>
* @param messageHandler the {@link MessageHandler} to use.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B handle(MessageHandler messageHandler) {
所有信息都在文档中:https://docs.spring.io/spring-integration/reference/html/dsl.html#java-dsl
像这样:
@Bean
public IntegrationFlow jmsPeriodicFlow() {
return IntegrationFlows.fromSupplier(() -> "hello",
e -> e.poller(p -> p.fixedDelay(5000)))
.handle(Jms.outboundAdapter(jmsConnectionFactory())
.destination("my.queue"))
.get();
}