文档章节

Spring Integration

引鸩怼孑
 引鸩怼孑
发布于 2015/05/29 09:14
字数 2583
阅读 2987
收藏 18
The Cafe Sample(小卖部订餐例子)

    小卖部有一个订饮料服务,客户可以通过订单来订购所需要饮料。小卖部提供两种咖啡饮料
        LATTE(拿铁咖啡)和MOCHA(摩卡咖啡)。每种又都分冷饮和热饮
    整个流程如下:
        1.有一个下订单模块,用户可以按要求下一个或多个订单。
        2.有一个订单处理模块,处理订单中那些是关于订购饮料的。
        3.有一个饮料订购处理模块,处理拆分订购的具体是那些种类的饮料,把具体需要生产的饮料要求发给生产模块
        4.有一个生产模块,进行生产。
        5.等生成完成后,有一个订单确认模块(Waiter),把订单的生成的饮料输出。

        
    这个例子利用Spring Integration实现了灵活的,可配置化的模式集成了上述这些服务模块。

Spring Integration提供两种模式的工作方式(Annotation和XML)
先来看一下XML方式,进行示例的开发:
配置文件如下:
<? xml version="1.0" encoding="UTF-8" ?>
< beans:beans  xmlns ="http://www.springframework.org/schema/integration"
    xmlns:xsi
="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:beans
="http://www.springframework.org/schema/beans"
    xmlns:stream
="http://www.springframework.org/schema/integration/stream"
    xsi:schemaLocation
="http://www.springframework.org/schema/beans
            http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
            http://www.springframework.org/schema/integration
            http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
            http://www.springframework.org/schema/integration/stream
            http://www.springframework.org/schema/integration/stream/spring-integration-stream-1.0.xsd"
>

        
<!--  首先来配置一个GateWay组件,提供消息的发送和接收。接口Cafe,提供一个void placeOrder(Order order);方法
        该方法标记了@Gateway(requestChannel="orders"), 实现向orders队列实现数据的发送
         
-->
    
< gateway  id ="cafe"  service-interface ="org.springframework.integration.samples.cafe.Cafe" />

        
<!--  订单Channel  -->
    
< channel  id ="orders" />
    
<!--  实现Splitter模式, 接收 orders队列的消息,调用orderSplitter Bean的split方法,进行消息的分解
    并把分解后的消息,发送到drinks队列.
     
-->
    
< splitter  input-channel ="orders"  ref ="orderSplitter"  method ="split"  output-channel ="drinks" />

        
<!--  饮料订单Channel,处理饮料的类别  -->
    
< channel  id ="drinks" />
    
<!--  实现Router模式,接收 drinks队列的消息, 并触发 drinkRouter Bean的 resolveOrderItemChannel方法
        由在 resolveOrderItemChannel该方法的返回值(String--队列名称)表示把消息路由到那个队列上
    
-->
    
< router  input-channel ="drinks"  ref ="drinkRouter"  method ="resolveOrderItemChannel" />
        
         
<!--  冷饮生产Channel 最大待处理的数据量为 10 -->
    
< channel  id ="coldDrinks" >
        
< queue  capacity ="10" />
    
</ channel >
    
<!--  定义一个服务处理器,其作用是定义一个消息接收队列 codeDrinks,一但收到消息,则
    触发 barista Bean的 prepareColdDrink方法, 再把 prepareColdDrink方法的值,封成Message的
    payLoad属性,把消息再发送到preparedDrinks队列, 
-->
    
< service-activator  input-channel ="coldDrinks"  ref ="barista"
                       method
="prepareColdDrink"  output-channel ="preparedDrinks" />

         
<!--  热饮生产Channel 最大待处理的数据量为 10 -->
    
< channel  id ="hotDrinks" >
        
< queue  capacity ="10" />
    
</ channel >
    
<!--  定义一个服务处理器,其作用是定义一个消息接收队列 hotDrinks,一但收到消息,则
    触发 barista Bean的 prepareHotDrink 再把 prepareColdDrink方法的值,封成Message的
    payLoad属性,把消息再发送到preparedDrinks队列, 
-->
    
< service-activator  input-channel ="hotDrinks"  ref ="barista"
                       method
="prepareHotDrink"  output-channel ="preparedDrinks" />
        
<!--  定义最终进行生产的消息队列  -->
    
< channel  id ="preparedDrinks" />
    
<!--  实现 aggregator 模式, 接收 preparedDrinks 消息, 并触发 waiter Bean的prepareDelivery方法
     再把处理好的数据,发送到 deliveries队列 
-->
    
< aggregator  input-channel ="preparedDrinks"  ref ="waiter"
                method
="prepareDelivery"  output-channel ="deliveries" />

        
<!--  定义一个 stream 适配器,接收 deliveries队列的消息后,直接输出到屏幕 -->
    
< stream:stdout-channel-adapter  id ="deliveries" />

    
< beans:bean  id ="orderSplitter"
                class
="org.springframework.integration.samples.cafe.xml.OrderSplitter" />

    
< beans:bean  id ="drinkRouter"
                class
="org.springframework.integration.samples.cafe.xml.DrinkRouter" />

    
< beans:bean  id ="barista"  class ="org.springframework.integration.samples.cafe.xml.Barista" />

    
< beans:bean  id ="waiter"  class ="org.springframework.integration.samples.cafe.xml.Waiter" />

</ beans:beans >

我们来看一下整体服务是怎么启动的
    
    首先我们来看一下CafeDemo这个类,它触发下定单操作
org.springframework.integration.samples.cafe.xml.CafeDemo
 1  public   class  CafeDemo {
 2 
 3       public   static   void  main(String[] args) {
 4           //// 加载Spring 配置文件 "cafeDemo.xml"
 5          AbstractApplicationContext context  =   null ;
 6           if (args.length  >   0 ) {
 7              context  =   new  FileSystemXmlApplicationContext(args);
 8          }
 9           else  {
10              context  =   new  ClassPathXmlApplicationContext( " cafeDemo.xml " , CafeDemo. class );
11          }
12           // 取得 Cafe实列
13          Cafe cafe  =  (Cafe) context.getBean( " cafe " );
14           // 准备 发送100条消息(订单)
15           for  ( int  i  =   1 ; i  <=   100 ; i ++ ) {
16              Order order  =   new  Order(i);
17               //  一杯热饮  参数说明1.饮料类型 2.数量 3.是否是冷饮(true表示冷饮)
18              order.addItem(DrinkType.LATTE,  2 false );
19               //  一杯冷饮  参数说明1.饮料类型 2.数量 3.是否是冷饮(true表示冷饮)
20              order.addItem(DrinkType.MOCHA,  3 true );
21               // 下发订单,把消息发给 orders 队列
22              cafe.placeOrder(order);
23          }
24      }
25 
26  }

下面是Cafe接口的源代码
public   interface  Cafe {

    
// 定义GateWay, 把消息发送到 orders 队列, Message的payLoad属性,保存 order参数值
    @Gateway(requestChannel = " orders " )
    
void  placeOrder(Order order);

}

OrderSplitter 源代码
 1  public   class  OrderSplitter {
 2 
 3       // 接收 从 orders队列接收的 order 消息后,调用 order.getItems方法
 4       // 进行订单的分解, 返回的List<OrderItem>可会,被拆分为多个消息后(Message.payLoad),发到指定队列
 5       public  List < OrderItem >  split(Order order) {
 6           return  order.getItems();
 7      }
 8 
 9  }
10 

OrderSplitter.split把消息拆分后,变成多个消息,发送到 drinks队列.由drinkRouter进行消息的接收。
1  public   class  DrinkRouter {
2 
3       // 从 drinks队列的消息后,根据orderItem的属性,选择路由到不同的队列 coldDrinks或hotDrinks
4       public  String resolveOrderItemChannel(OrderItem orderItem) {
5           return  (orderItem.isIced())  ?   " coldDrinks "  :  " hotDrinks " ;
6      }
7 
8  }

下面看一下,如果是一杯冷饮,则消息发送到 coldDrinks队列
接收根据配置,由barista Bean的prepareColdDrink方法接收消息后,进行处理
如果是一杯热饮,则消息发送到 hotDrinks队列
接收根据配置,由barista Bean的prepareHotDrink方法接收消息后,进行处理
 1  public   class  Barista {
 2 
 3       private   long  hotDrinkDelay  =   5000 ;
 4 
 5       private   long  coldDrinkDelay  =   1000 ;
 6 
 7       private  AtomicInteger hotDrinkCounter  =   new  AtomicInteger();
 8 
 9       private  AtomicInteger coldDrinkCounter  =   new  AtomicInteger();
10 
11 
12       public   void  setHotDrinkDelay( long  hotDrinkDelay) {
13           this .hotDrinkDelay  =  hotDrinkDelay;
14      }
15 
16       public   void  setColdDrinkDelay( long  coldDrinkDelay) {
17           this .coldDrinkDelay  =  coldDrinkDelay;
18      }
19 
20       // 处理热饮订单,并生成Drink冷料
21       public  Drink prepareHotDrink(OrderItem orderItem) {
22           try  {
23              Thread.sleep( this .hotDrinkDelay);
24              System.out.println(Thread.currentThread().getName()
25                       +   "  prepared hot drink # "   +  hotDrinkCounter.incrementAndGet()  +   "  for order # "
26                       +  orderItem.getOrder().getNumber()  +   " "   +  orderItem);
27               return   new  Drink(orderItem.getOrder().getNumber(), orderItem.getDrinkType(), orderItem.isIced(),
28                      orderItem.getShots());
29          }  catch  (InterruptedException e) {
30              Thread.currentThread().interrupt();
31               return   null ;
32          }
33      }
34 
35       // 处理冷饮订单,并生成Drink冷料
36       public  Drink prepareColdDrink(OrderItem orderItem) {
37           try  {
38              Thread.sleep( this .coldDrinkDelay);
39              System.out.println(Thread.currentThread().getName()
40                       +   "  prepared cold drink # "   +  coldDrinkCounter.incrementAndGet()  +   "  for order # "
41                       +  orderItem.getOrder().getNumber()  +   " "   +  orderItem);
42               return   new  Drink(orderItem.getOrder().getNumber(), orderItem.getDrinkType(), orderItem.isIced(),
43                      orderItem.getShots());
44          }  catch  (InterruptedException e) {
45              Thread.currentThread().interrupt();
46               return   null ;
47          }
48      }
49 
50  }

接下来,已经把订单需要生产的饮料已经完成,现在可以交给服务员(waier)交给客人了。
这里使用的aggregate模式,让服务器等待这个订单的所有饮料生产完后的,交给客户.

下面来介绍该应用
<!--  一旦定义了 aggregator,其会自动监测队列的消息,把消息合并后再发生指定的队列
    一般aggregator的参照 splitter一起使用。Spring Integration会根据接收到的消息中的消息头CORRELATION_ID 来判断,如果有相同的CORRELATION_ID发现,则认为它们需要合成一组,并返回(如果没有自定义合组接口)。
    当然Spring Integration也提供一个用户自定的接口来判定消息合组是否满足要求
   
public   interface  CompletionStrategy {

  
boolean  isComplete(List < Message <?>>  messages);

}


  isComplete的方法,收到的messages消息,都是拥用相同消息头CORRELATION_ID的消息。

-->
< aggregator  input-channel ="preparedDrinks"  ref ="waiter"
                method
="prepareDelivery"  output-channel ="deliveries" />

最后,完成订单的消息会发到 waiter队列
 1  public   class  Waiter {
 2 
 3       public  Delivery prepareDelivery(List < Drink >  drinks) {
 4           return   new  Delivery(drinks);
 5      }
 6 
 7 
 8  }
 9 
10  public   class  Delivery {
11 
12       private   static   final  String SEPARATOR  =   " ----------------------- " ;
13 
14 
15       private  List < Drink >  deliveredDrinks;
16 
17       private   int  orderNumber;
18 
19 
20       public  Delivery(List < Drink >  deliveredDrinks) {
21           assert (deliveredDrinks.size()  >   0 );
22           this .deliveredDrinks  =  deliveredDrinks;
23           this .orderNumber  =  deliveredDrinks.get( 0 ).getOrderNumber();
24      }
25 
26 
27       public   int  getOrderNumber() {
28           return  orderNumber;
29      }
30 
31       public  List < Drink >  getDeliveredDrinks() {
32           return  deliveredDrinks;
33      }
34 
35      @Override
36       public  String toString() {
37          StringBuffer buffer  =   new  StringBuffer(SEPARATOR  +   " \n " );
38          buffer.append( " Order # "   +  getOrderNumber()  +   " \n " );
39           for  (Drink drink : getDeliveredDrinks()) {
40              buffer.append(drink);
41              buffer.append( " \n " );
42          }
43          buffer.append(SEPARATOR  +   " \n " );
44           return  buffer.toString();
45      }
46 
47  }

最后我们使用一个 stream channel adaptor把订单生产完成的饮料输出。
         <!--  定义一个 stream 适配器,接收 deliveries队列的消息后,直接输出到屏幕 -->
    
< stream:stdout-channel-adapter  id ="deliveries" />


这样整个流程就执行完了,最终我们的饮料产品就按照订单生产出来了。累了吧,喝咖啡提神着呢!!!

spring-integration官网: http://www.springsource.org/spring-integration

关于 Annotation的介绍,将在 下篇 介绍。

附:xml配置介绍
Service Activator 配置
1  <!-- 配置 Service Activator,接收exampleChannel队列消息。注:exampleHandler至少有一个方法@ServiceActivator -->
2  < service-activator  input-channel ="exampleChannel"  ref ="exampleHandler" />
3  <!--  会检查 someMethod方法,是否有 @ServiceActivato 标注 output-channel -->
4  < service-activator  input-channel ="exampleChannel"  ref ="somePojo"  method ="someMethod" />
5  < service-activator  input-channel ="exampleChannel"  output-channel ="replyChannel"
6                     ref ="somePojo"  method ="someMethod" />

<inbound-channel-adapter>
 触发指定的方法,接收消息队列配置(触发轮循访问的方式)
 1  < inbound-channel-adapter  ref ="source1"  method ="method1"  channel ="channel1" >
 2       < poller >
 3           < interval-trigger  interval ="5000" />
 4       </ poller >
 5  </ inbound-channel-adapter >
 6 
 7  < inbound-channel-adapter  ref ="source2"  method ="method2"  channel ="channel2" >
 8       < poller >
 9           < cron-trigger  expression ="30 * * * * MON-FRI" />
10       </ poller >
11  </ channel-adapter >

 <outbound-channel-adapter/>
 触发指定的方法,发送消息
1    < outbound-channel-adapter  channel ="channel1"  ref ="target1"  method ="method1" />
2   
3    < outbound-channel-adapter  channel ="channel2"  ref ="target2"  method ="method2" >
4       < poller >
5           < interval-trigger  interval ="3000" />
6       </ poller >
7  </ outbound-channel-adapter >

Router
消息路由方式
1  < bean  id ="payloadTypeRouter"  class ="org.springframework.integration.router.PayloadTypeRouter" >
2       < property  name ="payloadTypeChannelMap" >
3           < map >
4               < entry  key ="java.lang.String"  value-ref ="stringChannel" />
5               < entry  key ="java.lang.Integer"  value-ref ="integerChannel" />
6           </ map >
7       </ property >
8  </ bean >

Aggregator 消息合并
 1  < channel  id ="inputChannel" />
 2 
 3  < aggregator  id ="completelyDefinedAggregator"  1
 4      input-channel ="inputChannel"  2
 5      output-channel ="outputChannel"   3
 6      discard-channel ="discardChannel"   4
 7      ref ="aggregatorBean"  5
 8      method ="add"  6
 9      completion-strategy ="completionStrategyBean"   7
10      completion-strategy-method ="checkCompleteness"  8
11      timeout ="42"  9
12      send-partial-result-on-timeout ="true"  10
13      reaper-interval ="135"  11
14      tracked-correlation-id-capacity ="99"  12
15      send-timeout ="86420000"  13  />  
16 
17  < channel  id ="outputChannel" />
18 
19  < bean  id ="aggregatorBean"  class ="sample.PojoAggregator" />
20 
21  < bean  id ="completionStrategyBean"  class ="sample.PojoCompletionStrategy" />

1

The id of the aggregator is optional.

2

The input channel of the aggregator. Required.

3

The channel where the aggregator will send the aggregation results. Optional (because incoming messages can specify a reply channel themselves).

4

The channel where the aggregator will send the messages that timed out (if send-partial-results-on-timeout is false)Optional.

5

A reference to a bean defined in the application context. The bean must implement the aggregation logic as described above. Required.

6

A method defined on the bean referenced by ref, that implements the message aggregation algorithm. Optional, with restrictions (see above).

7

A reference to a bean that implements the decision algorithm as to whether a given message group is complete. The bean can be an implementation of the CompletionStrategy interface or a POJO. In the latter case the completion-strategy-method attribute must be defined as well. Optional (by default, the aggregator .

8

A method defined on the bean referenced by completion-strategy, that implements the completion decision algorithm. Optional, with restrictions (requires completion-strategy to be present).

9

The timeout for aggregating messages (counted from the arrival of the first message). Optional.

10

Whether upon the expiration of the timeout, the aggregator shall try to aggregate the already arrived messages. Optional (false by default).

11

The interval (in milliseconds) at which a reaper task is executed, checking if there are any timed out groups. Optional.

12

The capacity of the correlation id tracker. Remembers the already processed correlation ids, preventing the formation of new groups for messages that arrive after their group has been already processed (aggregated or discarded). Optional.

13

The timeout for sending out messages. Optional.


配置消息合并策略
 1  public   class  PojoCompletionStrategy {
 2 
 3     public   boolean  checkCompleteness(List < Long >  numbers) {
 4       int  sum  =   0 ;
 5       for  ( long  number: numbers) {
 6        sum  +=  number;
 7      }
 8       return  sum  >=  maxValue;
 9    }
10  }


Good Luck!
Yours Matthew!

本文转载自:http://www.blogjava.net/xmatthew/archive/2008/12/04/243443.html

引鸩怼孑
粉丝 44
博文 207
码字总数 16947
作品 0
南京
项目经理
私信 提问
Spring Integration 4.1.0.RELEASE 发布

Spring Integration 4.1.0.RELEASE 发布,同时还发布了 Spring AMQP 1.4.0.RELEASE 版本。 这两个版本都是维护版本,包括重要的 bug 修复。更多内容请看 Spring Integration 4.1.0.RELEASE 发...

oschina
2014/11/12
1K
1
Spring AMQP 1.4.4/Spring Integration 3.0.7/4.0.7/4.1.3 发布

Spring Integration/AMQP 团队同时发布了一些维护版本: Spring AMQP 1.4.4 Spring Integration 3.0.7 Spring Integration 4.0.7 Spring Integration 4.1.3 以上所有版本都提供在 Release Re......

oschina
2015/04/08
1K
2
Spring Integration 2.2.4 和 2.1.6 发布

Spring Integration 2.2.4 和 2.1.6 发布了,这两个都是维护版本。主要是升级一些第三方包到最新版本,修复 bug 等。 相关链接: [1]: http://www.springsource.org/spring-social/news/spri...

红薯
2013/06/13
736
0
Spring Integration 4.2.7 和 4.2.8 发布

Spring Integration 4.2.7 和 4.2.8 发布了,Spring Integration能在基于Spring的应用中进行简单的消息通信,并通过简单的适配器与外部系统集成。这些适配器提供了一个更高级别的抽象,超越 ...

淡漠悠然
2016/06/11
1K
0
Spring BlazeDS Integration 1.0.0.M2 发布

Dear Spring Community, I'm pleased to announce that the second public development milestone of Spring BlazeDS Integration, the newest member of the open source Spring portfolio,......

红薯
2009/03/06
688
0

没有更多内容

加载失败,请刷新页面

加载更多

CSS盒子模型

一、什么叫框模型 页面元素皆为框(盒子) 定义了元素框处理元素内容,内边距,外边距以及边框的计算方式 二、外边距 围绕在元素边框外的空白距离(元素与元素之间的距离) 语法:margin,定...

wytao1995
15分钟前
1
0
Replugin借助“UI进程”来快速释放Dex

public static boolean preload(PluginInfo pi) { if (pi == null) { return false; } // 借助“UI进程”来快速释放Dex(见PluginFastInstallProviderProxy的说明) return PluginFastInsta......

Gemini-Lin
今天
4
0
Hibernate 5 的模块/包(modules/artifacts)

Hibernate 的功能被拆分成一系列的模块/包(modules/artifacts),其目的是为了对依赖进行独立(模块化)。 模块名称 说明 hibernate-core 这个是 Hibernate 的主要(main (core))模块。定义...

honeymoose
今天
4
0
精华帖

第一章 jQuery简介 jQuery是一个JavaScript库 jQuery具备简洁的语法和跨平台的兼容性 简化了JavaScript的操作。 在页面中引入jQuery jQuery是一个JavaScript脚本库,不需要特别的安装,只需要...

流川偑
今天
7
0
语音对话英语翻译在线翻译成中文哪个方法好用

想要进行将中文翻译成英文,或者将英文翻译成中文的操作,其实有一个非常简单的工具就能够帮助完成将语音进行翻译转换的软件。 在应用市场或者百度手机助手等各大应用渠道里面就能够找到一款...

401恶户
今天
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部