Spring Boot集成Java DSL的实现代码

2023-09-01 18:29:05 64

SpringIntegrationJavaDSL已经融合到SpringIntegrationCore5.0，这是一个聪明而明显的举动，因为：

基于JavaConfig启动新Spring项目的每个人都使用它
SIJavaDSL使您可以使用Lambdas等新的强大Java8功能
您可以使用基于IntegrationFlowBuilder的Builder模式构建流

让我们看看基于ActiveMQJMS的示例如何使用它。

Maven依赖：



org.springframework.boot
spring-boot-starter-activemq



org.springframework.integration
spring-integration-core



org.springframework.integration
spring-integration-jms



org.springframework.boot
spring-boot-starter-test
test



org.apache.activemq
activemq-kahadb-store




org.springframework.integration
spring-integration-java-dsl
1.2.3.RELEASE

示例1：Jms入站网关

我们有以下ServiceActivator：

@Service
publicclassActiveMQEndpoint{
@ServiceActivator(inputChannel="inboundChannel")
publicvoidprocessMessage(finalStringinboundPayload){
System.out.println("Inboundmessage:"+inboundPayload);
}
}

如果您想使用SIJavaDSL将inboundPayload从Jms队列发送到Gateway风格的激活器，那么请使用DSLJms工厂：

@Bean
publicDynamicDestinationResolverdynamicDestinationResolver(){
returnnewDynamicDestinationResolver();
}

@Bean
publicActiveMQConnectionFactoryconnectionFactory(){
returnnewActiveMQConnectionFactory();
}

@Bean
publicDefaultMessageListenerContainerlistenerContainer(){
finalDefaultMessageListenerContainerdefaultMessageListenerContainer=newDefaultMessageListenerContainer();
defaultMessageListenerContainer.setDestinationResolver(dynamicDestinationResolver());
defaultMessageListenerContainer.setConnectionFactory(connectionFactory());
defaultMessageListenerContainer.setDestinationName("jms.activeMQ.Test");
returndefaultMessageListenerContainer;
}

@Bean
publicMessageChannelinboundChannel(){
returnMessageChannels.direct("inboundChannel").get();
}

@Bean
publicJmsInboundGatewaydataEndpoint(){
returnJms.inboundGateway(listenerContainer())
.requestChannel(inboundChannel()).get();
}

通过dataEndpointbean返回JmsInboundGatewaySpec，您还可以向SI通道或Jms目标发送回复。查看文档。

示例2：Jms消息驱动的通道适配器

如果您正在寻找替换消息驱动通道适配器的XMLJMS配置，那么JmsMessageDrivenChannelAdapter是一种适合您的方式：

@Bean
publicDynamicDestinationResolverdynamicDestinationResolver(){
returnnewDynamicDestinationResolver();
}

@Bean
publicActiveMQConnectionFactoryconnectionFactory(){
returnnewActiveMQConnectionFactory();
}

@Bean
publicDefaultMessageListenerContainerlistenerContainer(){
finalDefaultMessageListenerContainerdefaultMessageListenerContainer=newDefaultMessageListenerContainer();
defaultMessageListenerContainer.setDestinationResolver(dynamicDestinationResolver());
defaultMessageListenerContainer.setConnectionFactory(connectionFactory());
defaultMessageListenerContainer.setDestinationName("jms.activeMQ.Test");
returndefaultMessageListenerContainer;
}

@Bean
publicMessageChannelinboundChannel(){
returnMessageChannels.direct("inboundChannel").get();
}

@Bean
publicJmsMessageDrivenChannelAdapterdataEndpoint(){
finalChannelPublishingJmsMessageListenerchannelPublishingJmsMessageListener=
newChannelPublishingJmsMessageListener();
channelPublishingJmsMessageListener.setExpectReply(false);
finalJmsMessageDrivenChannelAdaptermessageDrivenChannelAdapter=new
JmsMessageDrivenChannelAdapter(listenerContainer(),channelPublishingJmsMessageListener
);

messageDrivenChannelAdapter.setOutputChannel(inboundChannel());
returnmessageDrivenChannelAdapter;
}

与前面的示例一样，入站有效负载如样本1中一样发送给激活器。

示例3：使用JAXB的Jms消息驱动的通道适配器

在典型的场景中，您希望通过Jms接受XML作为文本消息，将其转换为JAXB存根并在服务激活器中处理它。我将向您展示如何使用SIJavaDSL执行此操作，但首先让我们为xml处理添加两个依赖项：


org.springframework.integration
spring-integration-xml



org.springframework
spring-oxm

我们将通过JMS接受shiporders，所以首先XSD命名为shiporder.xsd：

新增JAXBmavenplugin生成JAXB存根：


org.codehaus.mojo
jaxb2-maven-plugin
2.3.1


xjc-schema1

xjc




src/main/resources/xsds/shiporder.xsd



com.example.stubs
src/main/java
false

我们已经准备好了存根类和一切，现在使用Jaxbmagic的JavaDSLJMS消息驱动适配器：

/**
*Sample3:JmsmessagedrivenadapterwithJAXB
*/
@Bean
publicJmsMessageDrivenChannelAdapterdataEndpoint(){
finalChannelPublishingJmsMessageListenerchannelPublishingJmsMessageListener=
newChannelPublishingJmsMessageListener();
channelPublishingJmsMessageListener.setExpectReply(false);
channelPublishingJmsMessageListener.setMessageConverter(newMarshallingMessageConverter(shipOrdersMarshaller()));
finalJmsMessageDrivenChannelAdaptermessageDrivenChannelAdapter=new
JmsMessageDrivenChannelAdapter(listenerContainer(),channelPublishingJmsMessageListener
);

messageDrivenChannelAdapter.setOutputChannel(inboundChannel());
returnmessageDrivenChannelAdapter;
}

@Bean
publicJaxb2MarshallershipOrdersMarshaller(){
Jaxb2Marshallermarshaller=newJaxb2Marshaller();
marshaller.setContextPath("com.example.stubs");
returnmarshaller;
}

XML配置在Java中使用它可以为您提供如此强大的功能和灵活性。要完成此示例，inboundChannel的服务激活器将如下所示：

/**
*Sample3
*@paramshiporder
*/
@ServiceActivator(inputChannel="inboundChannel")
publicvoidprocessMessage(finalShipordershiporder){
System.out.println(shiporder.getOrderid());
System.out.println(shiporder.getOrderperson());
}

要测试流，您可以使用以下XML通过JConsole发送到JMS队列：



JohnSmith

OlaNordmann
Langgt23
4000Stavanger
Norway


EmpireBurlesque
SpecialEdition
1
10.90


Hideyourheart
1
9.90

示例4：具有JAXB和有效负载根路由的Jms消息驱动的通道适配器

另一种典型情况是接受XML作为JMS文本消息，将其转换为JAXB存根并根据有效负载根类型将有效负载路由到某个服务激活器。当然SIJavaDSL支持所有类型的路由，我将向您展示如何根据有效载荷类型进行路由。

首先，将以下XSD添加到shiporder.xsd所在的文件夹中，并将其命名为purchaseorder.xsd：

然后添加到jaxbmaven插件配置：


org.codehaus.mojo
jaxb2-maven-plugin
2.3.1


xjc-schema1

xjc




src/main/resources/xsds/shiporder.xsd
src/main/resources/xsds/purchaseorder.xsd



com.example.stubs
src/main/java
false

运行mvncleaninstall以生成新XSD的JAXB存根。现在承诺有效负载根映射：

@Bean
publicJaxb2MarshallerordersMarshaller(){
Jaxb2Marshallermarshaller=newJaxb2Marshaller();
marshaller.setContextPath("com.example.stubs");
returnmarshaller;
}

/**
*Sample4:JmsmessagedrivenadapterwithJaxbandPayloadrouting.
*@return
*/
@Bean
publicJmsMessageDrivenChannelAdapterdataEndpoint(){
finalChannelPublishingJmsMessageListenerchannelPublishingJmsMessageListener=
newChannelPublishingJmsMessageListener();
channelPublishingJmsMessageListener.setMessageConverter(newMarshallingMessageConverter(ordersMarshaller()));
finalJmsMessageDrivenChannelAdaptermessageDrivenChannelAdapter=new
JmsMessageDrivenChannelAdapter(listenerContainer(),channelPublishingJmsMessageListener
);

messageDrivenChannelAdapter.setOutputChannel(inboundChannel());
returnmessageDrivenChannelAdapter;
}

@Bean
publicIntegrationFlowpayloadRootMapping(){
returnIntegrationFlows.from(inboundChannel()).>route(Object::getClass,m->m
.subFlowMapping(Shiporder.class,sf->sf.handle((MessageHandler)message->{
finalShipordershiporder=(Shiporder)message.getPayload();
System.out.println(shiporder.getOrderperson());
System.out.println(shiporder.getOrderid());
}))
.subFlowMapping(PurchaseOrder.class,sf->sf.handle((MessageHandler)message->{
finalPurchaseOrderpurchaseOrderType=(PurchaseOrder)message.getPayload();
System.out.println(purchaseOrderType.getBillTo().getName());
}))
).get();
}

注意payloadRootMappingbean，让我们解释一下重要的部分：

>route-表示来自inboundChannel的输入将是Object，并且将根据Class<？>执行路由
subFlowMapping(Shiporder.class..-ShipOders的处理。
subFlowMapping(PurchaseOrder.class...-处理PurchaseOrders。

要测试ShipOrder有效负载，请使用示例3中的XML，以测试PurchaseOrder有效负载，使用以下XML：




name1
street1
city1
state1
1


name2
street2
city2
state2
-79228162514264337593543950335


name1
street1
city1
state1
1

应根据subflow子流Map路由两个有效载荷。

示例5：IntegrationFlowAdapter

除了企业集成模式的其他实现（checkthemout)），我需要提到IntegrationFlowAdapter。通过扩展此类并实现buildFlow方法，如：

[url=https://bitbucket.org/Component/]@Component[/url]
publicclassMyFlowAdapterextendsIntegrationFlowAdapter{

@Autowired
privateConnectionFactoryrabbitConnectionFactory;

@Override
protectedIntegrationFlowDefinitionbuildFlow(){
returnfrom(Amqp.inboundAdapter(this.rabbitConnectionFactory,"myQueue"))
.transform(String::toLowerCase)
.channel(c->c.queue("myFlowAdapterOutput"));
}

你可以将bean的重复声明包装成一个组件并给它们所需的流量。然后可以配置这样的组件并将其作为一个类实例提供给调用代码！

因此，让我们举例说明这个repo中的示例3更短一些，并为所有JmsEndpoints定义基类，并在其中定义重复bean：

publicclassJmsEndpointextendsIntegrationFlowAdapter{

privateStringqueueName;

privateStringchannelName;

privateStringcontextPath;

/**
*@paramqueueName
*@paramchannelName
*@paramcontextPath
*/
publicJmsEndpoint(StringqueueName,StringchannelName,StringcontextPath){
this.queueName=queueName;
this.channelName=channelName;
this.contextPath=contextPath;
}

@Override
protectedIntegrationFlowDefinitionbuildFlow(){
returnfrom(Jms.messageDrivenChannelAdapter(listenerContainer())
.jmsMessageConverter(newMarshallingMessageConverter(shipOrdersMarshaller()))
).channel(channelName);
}

@Bean
publicJaxb2MarshallershipOrdersMarshaller(){
Jaxb2Marshallermarshaller=newJaxb2Marshaller();
marshaller.setContextPath(contextPath);
returnmarshaller;
}

@Bean
publicDynamicDestinationResolverdynamicDestinationResolver(){
returnnewDynamicDestinationResolver();
}

@Bean
publicActiveMQConnectionFactoryconnectionFactory(){
returnnewActiveMQConnectionFactory();
}

@Bean
publicDefaultMessageListenerContainerlistenerContainer(){
finalDefaultMessageListenerContainerdefaultMessageListenerContainer=newDefaultMessageListenerContainer();
defaultMessageListenerContainer.setDestinationResolver(dynamicDestinationResolver());
defaultMessageListenerContainer.setConnectionFactory(connectionFactory());
defaultMessageListenerContainer.setDestinationName(queueName);
returndefaultMessageListenerContainer;
}

@Bean
publicMessageChannelinboundChannel(){
returnMessageChannels.direct(channelName).get();
}
}

现在声明特定队列的Jms端点很容易：

@Bean
publicJmsEndpointjmsEndpoint(){
returnnewJmsEndpoint("jms.activeMQ.Test","inboundChannel","com.example.stubs");
}

inboundChannel的服务激活器：

/**
*Sample3,5
*@paramshiporder
*/
@ServiceActivator(inputChannel="inboundChannel")
publicvoidprocessMessage(finalShipordershiporder){
System.out.println(shiporder.getOrderid());
System.out.println(shiporder.getOrderperson());
}

您不应该错过在项目中使用IntegrationFlowAdapter。我喜欢它的概念。

我最近在Embedit的新的基于SpringBoot的项目中开始使用SpringIntegrationJavaDSL。即使有一些配置，我发现它非常有用。

它很容易调试。不添加像wiretap这样的配置。
阅读起来要容易得多。是的，即使是lambdas！
它很强大。在Java配置中，您现在有很多选择。

源码地址：https://bitbucket.org/tomask79/spring-integration-java-dsl

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。

声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。

Spring Boot集成Java DSL的实现代码

热门推荐

随机推荐