资讯

精准传达 • 有效沟通

从品牌网站建设到网络营销策划，从策略到执行的一站式服务

网站建设 >

查看其它板块

SpringBoot集成JavaDSL的实现代码

Spring Integration Java DSL已经融合到Spring Integration Core 5.0，这是一个聪明而明显的举动，因为：

创新互联公司长期为近1000家客户提供的网站建设服务，团队从业经验10年，关注不同地域、不同群体，并针对不同对象提供差异化的产品和服务；打造开放共赢平台，与合作伙伴共同营造健康的互联网生态环境。为井陉矿企业提供专业的网站设计、做网站，井陉矿网站改版等技术服务。拥有10多年丰富建站经验和众多成功案例,为您定制开发。

  • 基于Java Config启动新Spring项目的每个人都使用它
  • SI Java DSL使您可以使用Lambdas等新的强大Java 8功能
  • 您可以使用 基于IntegrationFlowBuilder的Builder模式构建流

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

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
public class ActiveMQEndpoint {
  @ServiceActivator(inputChannel = "inboundChannel")
  public void processMessage(final String inboundPayload) {
    System.out.println("Inbound message: "+inboundPayload);
  }
}

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

@Bean
public DynamicDestinationResolver dynamicDestinationResolver() {
  return new DynamicDestinationResolver();
}

@Bean
public ActiveMQConnectionFactory connectionFactory() {
  return new ActiveMQConnectionFactory();
}

@Bean
public DefaultMessageListenerContainer listenerContainer() {
  final DefaultMessageListenerContainer defaultMessageListenerContainer = new DefaultMessageListenerContainer();
  defaultMessageListenerContainer.setDestinationResolver(dynamicDestinationResolver());
  defaultMessageListenerContainer.setConnectionFactory(connectionFactory());
  defaultMessageListenerContainer.setDestinationName("jms.activeMQ.Test");
  return defaultMessageListenerContainer;
}

@Bean
public MessageChannel inboundChannel() {
  return MessageChannels.direct("inboundChannel").get();
}

@Bean
public JmsInboundGateway dataEndpoint() {
  return Jms.inboundGateway(listenerContainer())
      .requestChannel(inboundChannel()).get();
}

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

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

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

@Bean
public DynamicDestinationResolver dynamicDestinationResolver() {
  return new DynamicDestinationResolver();
}

@Bean
public ActiveMQConnectionFactory connectionFactory() {
  return new ActiveMQConnectionFactory();
}

@Bean
public DefaultMessageListenerContainer listenerContainer() {
  final DefaultMessageListenerContainer defaultMessageListenerContainer = new DefaultMessageListenerContainer();
  defaultMessageListenerContainer.setDestinationResolver(dynamicDestinationResolver());
  defaultMessageListenerContainer.setConnectionFactory(connectionFactory());
  defaultMessageListenerContainer.setDestinationName("jms.activeMQ.Test");
  return defaultMessageListenerContainer;
}

@Bean
public MessageChannel inboundChannel() {
  return MessageChannels.direct("inboundChannel").get();
}

@Bean
public JmsMessageDrivenChannelAdapter dataEndpoint() {
  final ChannelPublishingJmsMessageListener channelPublishingJmsMessageListener =
      new ChannelPublishingJmsMessageListener();
  channelPublishingJmsMessageListener.setExpectReply(false);
  final JmsMessageDrivenChannelAdapter messageDrivenChannelAdapter = new
      JmsMessageDrivenChannelAdapter(listenerContainer(), channelPublishingJmsMessageListener
      );

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

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

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

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


    org.springframework.integration
    spring-integration-xml
  

  
    org.springframework
    spring-oxm

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

<?xml version="1.0" encoding="UTF-8" ?>

新增JAXB maven plugin 生成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

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

/**
 * Sample 3: Jms message driven adapter with JAXB
 */
@Bean
public JmsMessageDrivenChannelAdapter dataEndpoint() {
  final ChannelPublishingJmsMessageListener channelPublishingJmsMessageListener =
      new ChannelPublishingJmsMessageListener();
  channelPublishingJmsMessageListener.setExpectReply(false);
  channelPublishingJmsMessageListener.setMessageConverter(new MarshallingMessageConverter(shipOrdersMarshaller()));
  final JmsMessageDrivenChannelAdapter messageDrivenChannelAdapter = new
      JmsMessageDrivenChannelAdapter(listenerContainer(), channelPublishingJmsMessageListener
  );

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

@Bean
public Jaxb2Marshaller shipOrdersMarshaller() {
  Jaxb2Marshaller marshaller = new Jaxb2Marshaller();
  marshaller.setContextPath("com.example.stubs");
  return marshaller;
}

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

/**
 * Sample 3
 * @param shiporder
 */
@ServiceActivator(inputChannel = "inboundChannel")
public void processMessage(final Shiporder shiporder) {
  System.out.println(shiporder.getOrderid());
  System.out.println(shiporder.getOrderperson());
}

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

 <?xml version="1.0" encoding="UTF-8"?>    
  
   John Smith
    
     Ola Nordmann
     
Langgt 23

     4000 Stavanger
     Norway
    
    
     Empire Burlesque
     Special Edition
     1
     10.90
     
    
     Hide your heart
     1
     9.90

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

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

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

然后添加到jaxb maven插件配置：

 
      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

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

@Bean
public Jaxb2Marshaller ordersMarshaller() {
  Jaxb2Marshaller marshaller = new Jaxb2Marshaller();
  marshaller.setContextPath("com.example.stubs");
  return marshaller;
}

/**
 * Sample 4: Jms message driven adapter with Jaxb and Payload routing.
 * @return
 */
@Bean
public JmsMessageDrivenChannelAdapter dataEndpoint() {
  final ChannelPublishingJmsMessageListener channelPublishingJmsMessageListener =
      new ChannelPublishingJmsMessageListener();
  channelPublishingJmsMessageListener.setMessageConverter(new MarshallingMessageConverter(ordersMarshaller()));
  final JmsMessageDrivenChannelAdapter messageDrivenChannelAdapter = new
      JmsMessageDrivenChannelAdapter(listenerContainer(), channelPublishingJmsMessageListener
  );

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

@Bean
public IntegrationFlow payloadRootMapping() {
  return IntegrationFlows.from(inboundChannel()).>route(Object::getClass, m->m
      .subFlowMapping(Shiporder.class, sf->sf.handle((MessageHandler) message -> {
        final Shiporder shiporder = (Shiporder) message.getPayload();
        System.out.println(shiporder.getOrderperson());
        System.out.println(shiporder.getOrderid());
      }))
      .subFlowMapping(PurchaseOrder.class, sf->sf.handle((MessageHandler) message -> {
        final PurchaseOrder purchaseOrderType = (PurchaseOrder) message.getPayload();
        System.out.println(purchaseOrderType.getBillTo().getName());
      }))
  ).get();
}

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

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

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

<?xml version="1.0" encoding="utf-8"?> 
 
  
  name1 
  street1 
  city1 
  state1 
  1 
  
  
  name2 
  street2 
  city2 
  state2 
  -79228162514264337593543950335 
  
  
  name1 
  street1 
  city1 
  state1 
  1

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

示例5：IntegrationFlowAdapter

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

[url=https://bitbucket.org/Component/]@Component[/url] 
public class MyFlowAdapter extends IntegrationFlowAdapter {

@Autowired
 private ConnectionFactory rabbitConnectionFactory;

 @Override
 protected IntegrationFlowDefinition<?> buildFlow() {
   return from(Amqp.inboundAdapter(this.rabbitConnectionFactory, "myQueue"))
        .transform(String::toLowerCase)
        .channel(c -> c.queue("myFlowAdapterOutput"));
 }

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

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

public class JmsEndpoint extends IntegrationFlowAdapter {

  private String queueName;

  private String channelName;

  private String contextPath;

  /**
   * @param queueName
   * @param channelName
   * @param contextPath
   */
  public JmsEndpoint(String queueName, String channelName, String contextPath) {
    this.queueName = queueName;
    this.channelName = channelName;
    this.contextPath = contextPath;
  }

  @Override
  protected IntegrationFlowDefinition<?> buildFlow() {
    return from(Jms.messageDrivenChannelAdapter(listenerContainer())
      .jmsMessageConverter(new MarshallingMessageConverter(shipOrdersMarshaller()))
    ).channel(channelName);
  }

  @Bean
  public Jaxb2Marshaller shipOrdersMarshaller() {
    Jaxb2Marshaller marshaller = new Jaxb2Marshaller();
    marshaller.setContextPath(contextPath);
    return marshaller;
  }

  @Bean
  public DynamicDestinationResolver dynamicDestinationResolver() {
    return new DynamicDestinationResolver();
  }

  @Bean
  public ActiveMQConnectionFactory connectionFactory() {
    return new ActiveMQConnectionFactory();
  }

  @Bean
  public DefaultMessageListenerContainer listenerContainer() {
    final DefaultMessageListenerContainer defaultMessageListenerContainer = new DefaultMessageListenerContainer();
    defaultMessageListenerContainer.setDestinationResolver(dynamicDestinationResolver());
    defaultMessageListenerContainer.setConnectionFactory(connectionFactory());
    defaultMessageListenerContainer.setDestinationName(queueName);
    return defaultMessageListenerContainer;
  }

  @Bean
  public MessageChannel inboundChannel() {
    return MessageChannels.direct(channelName).get();
  }
}

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

@Bean
public JmsEndpoint jmsEndpoint() {
  return new JmsEndpoint("jms.activeMQ.Test", "inboundChannel", "com.example.stubs");
}

inboundChannel的服务激活器：

/**
 * Sample 3, 5
 * @param shiporder
 */
@ServiceActivator(inputChannel = "inboundChannel")
public void processMessage(final Shiporder shiporder) {
  System.out.println(shiporder.getOrderid());
  System.out.println(shiporder.getOrderperson());
}

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

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

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

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

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


当前标题：SpringBoot集成JavaDSL的实现代码
文章出自：http://cdkjz.cn/article/jhehoe.html
返回首页 了解更多建站资讯

相关资讯

多年建站经验

多一份参考，总有益处

联系快上网，免费获得专属《策划方案》及报价

咨询相关问题或预约面谈，可以通过以下方式与我们联系

业务热线：400-028-6601 / 大客户专线   成都：13518219792   座机：028-86922220

在线咨询 提交需求
友情链接 交换友情链接
网站设计成都网络营销四川雅安服务器托管蜜朵婚庆成都做网站成都画册快印香港虚拟主机空间物流公司网站建设方案四川电信高防服务器托管主机托管

成都网站建设公司地址：成都市青羊区太升南路288号锦天国际A座10层 建设咨询400-028-6601

成都快上网科技有限公司-四川网站建设设计公司 | 蜀ICP备19037934号 Copyright 2020,ALL Rights Reserved cdkjz.cn | 成都网站建设 | © Copyright 2020版权所有.

专家团队为您提供成都网站建设,成都网站设计,成都品牌网站设计,成都营销型网站制作等服务,成都建网站就找快上网！ | 成都网站建设哪家好？ | 网站建设地图