知识大全 Spring Integration学习笔记(一)

Posted 2022-07-18 消息

篇首语：知之者不如好之者，好之者不如乐之者。本文由小常识网(cha138.com)小编为大家整理，主要介绍了知识大全 Spring Integration学习笔记(一)相关的知识，希望对你有一定的参考价值。

Spring Integration学习笔记(一) 以下文字资料是由(全榜网网www.cha138.com)小编为大家搜集整理后发布的内容，让我们赶快一起来看一下吧！

　　The Cafe Sample（小卖部订餐例子）

　　小卖部有一个订饮料服务客户可以通过订单来订购所需要饮料小卖部提供两种咖啡饮料LATTE（拿铁咖啡）和MOCHA（摩卡咖啡）每种又都分冷饮和热饮整个流程如下

　　有一个下订单模块用户可以按要求下一个或多个订单

　　有一个订单处理模块处理订单中那些是关于订购饮料的

　　有一个饮料订购处理模块处理拆分订购的具体是那些种类的饮料把具体需要生产的饮料要求发给生产模块有一个生产模块

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

　　先来看一下配置文件

　　 <beans:beans xmlns= xmlns:xsi= instance xmlns:beans= xmlns:context= xsi:schemaLocation= beans xsd integration xsd context xsd > <! 启动Message bus 消息服务总线支持四个属性 auto startup[boolean是否自动启动 default=true]如果设置false 则需要手动调用applicationContext start()方法 auto create channels[boolean是否自动注册MessageChannel default=false] 如果使用的MessagChannle不存在 error channel 设置错误时信息发送的MessageChannle 如果不设置则使用DefaultErrorChannel dispatcher pool size 使用的启动线程数默认为 > <message bus/> <! 启动支持元数据标记 > <annotation driven/> <! 设置 @Component标识的元数据扫描包(package) > <context:ponent scan base package= springframework integration samples cafe /> <! 下面启动了四个 MessageChannel服务处理接收发送端发过来的消息和把消息流转到消息的消费端 > <! 属性说明 capacity 消息最大容量默认为 publish subscribe是否是发布订阅模式默认为否 id bean的id名称 datatype ？ > <channel id= orders /> <! 订单Channel > <channel id= drinks /> <! 饮料订单Channel 处理饮料的类别 > <channel id= coldDrinks /> <! 热饮生产Channel > <channel id= hotDrinks /> <! 冷饮生产Channel > <! 消息处理终端接收 channel coldDrinks的消息后执行barista prepareColdDrink方法生产冷饮 > <! 属性说明 input channel 接收消息的Channel必须 default output channel设置默认回复消息Channel handler ref 引用bean的id名称 handler method Handler处理方法名(参数类型必须与发送消息的payLoad使用的一致) error handler设置错误时信息发送的MessageChannle reply handler 消息回复的Channel > <endpoint input channel= coldDrinks handler ref= barista handler method= prepareColdDrink /> <! 消息处理终端接收 channel hotDrinks的消息后执行barista prepareHotDrink方法生产热饮 > <endpoint input channel= hotDrinks handler ref= barista handler method= prepareHotDrink /> <! 定义一个启动下定单操作的bean 它通过 channel orders下定单 > <beans:bean id= cafe class= springframework integration samples cafe Cafe > <beans:property name= orderChannel ref= orders /> </beans:bean></beans:beans>

　　下面我们来看一下源代码目录

　　我们来看一下整体服务是怎么启动的

　　首先我们来看一下CafeDemo这个类它触发下定单操作

　　 public class CafeDemo public static void main(String[] args) //加载Spring 配置文件 AbstractApplicationContext context = null; if(args length > ) context = new FileSystemXmlApplicationContext(args); else context = new ClassPathXmlApplicationContext( cafeDemo xml CafeDemo class); //启动 Spring容器(启动所有实现 ntext Lifecycle接口的实现类的start方法) context start(); //从Spring容器取得cafe实例 Cafe cafe = (Cafe) context getBean( cafe ); DrinkOrder order = new DrinkOrder(); //一杯热饮参数说明饮料类型数量是否是冷饮(true表示冷饮) Drink hotDoubleLatte = new Drink(DrinkType LATTE false); Drink icedTripleMocha = new Drink(DrinkType MOCHA true); order addDrink(hotDoubleLatte); order addDrink(icedTripleMocha); //下个订单 for (int i = ; i < ; i++) //调用cafe的placeOrder下订单 cafe placeOrder(order);

　　下面是Cafe的源代码

　　 public class Cafe private MessageChannel orderChannel; public void setOrderChannel(MessageChannel orderChannel) this orderChannel = orderChannel; //其实下订单操作调用的是orderChannel(orders channel)的send方法把消息发出去 public void placeOrder(DrinkOrder order) this orderChannel send(new GenericMessage<DrinkOrder>(order)); //GenericMessage有三个构建方法参考如下 //new GenericMessage<T>(Object id T payload); //new GenericMessage<T>(T payload); //new GenericMessage<T>(T payload MessageHeader headerToCopy)

　　下面我们来看一下哪个类标记有@MessageEndpoint（input= orders ）表示它会消费orders Channel的消息我们发现OrderSplitter类标记这个元数据下面是源代码我们来分析

　　 //标记 MessageEndpoint 元数据 input表示设置后所有 orders Channel消息都会被OrderSplitter收到 @MessageEndpoint(input= orders ) public class OrderSplitter //@Splitter表示接收消息后调用这个类的该方法其的参数类型必须与message的 payload属性一致 //即在new GenericMessage<T>的泛型中指定 //元数据设置的 channel属性表示方法执行完成后会把方法返回的结果保存到message的payload属性后发送到指定的channel中去 //这里指定发送到 drinks channel @Splitter(channel= drinks ) public List<Drink> split(DrinkOrder order) return order getDrinks(); //方法中是把订单中的饮料订单取出来

　　接下来与找OrderSplitter方法相同我们要找哪个类标记有@MessageEndpoint（input= drinks ）表示它会消费drinks Channel的消息找到DrinkRouter这个类

　　 @MessageEndpoint(input= drinks ) public class DrinkRouter //@Router表示接收消息后调用这个类的该方法其的参数类型必须与message的 payload属性一致 //方法执行完毕后其返回值为在容器中定义的channel名称 channel名称必须存在 @Router public String resolveDrinkChannel(Drink drink) return (drink isIced()) ? coldDrinks : hotDrinks ; //方法中是根据处理饮料是否是冷饮送不同的channel处理

备注 @Router可以把消息路由到多个channel 实现方式如下 @Router public MessageChannel route(Message message) @Router public List<MessageChannel> route(Message message) @Router public String route(Foo payload) @Router public List<String> route(Foo payload)

　　接下来我们就要找 MessageEndpoint 标记为处理 coldDrinks 和 hotDrinks 的类我们发现这个两个类并不是通过元数据@MessageEndpoint来实现的而是通过容器配置（下面会演示如何用元数据配置但元数据配置有局限性这两种配置方式看大家喜好系统中都是可以使用）

　　下面是容器配置信息

　　 <! 消息处理终端接收 channel coldDrinks的消息后执行barista prepareColdDrink方法生产冷饮 > <endpoint input channel= coldDrinks handler ref= barista handler method= prepareColdDrink /> <! 消息处理终端接收 channel hotDrinks的消息后执行barista prepareHotDrink方法生产热饮 > <endpoint input channel= hotDrinks handler ref= barista handler method= prepareHotDrink />

　　我们来看一下源代码

　　 @Component //这个必须要有表示是一个消息处理组件 public class Barista private long hotDrinkDelay = ; private long coldDrinkDelay = ; private AtomicInteger hotDrinkCounter = new AtomicInteger(); private AtomicInteger coldDrinkCounter = new AtomicInteger(); public void setHotDrinkDelay(long hotDrinkDelay) this hotDrinkDelay = hotDrinkDelay; public void setColdDrinkDelay(long coldDrinkDelay) ldDrinkDelay = coldDrinkDelay; public void prepareHotDrink(Drink drink) try Thread sleep(this hotDrinkDelay); catch (InterruptedException e) Thread currentThread() interrupt(); System out println( prepared hot drink # + hotDrinkCounter incrementAndGet() + : + drink); public void prepareColdDrink(Drink drink) try Thread sleep(ldDrinkDelay); catch (InterruptedException e) Thread currentThread() interrupt(); System out println( prepared cold drink # + coldDrinkCounter incrementAndGet() + : + drink);

　　如果要用元数据标识实现上述方法要用元数据配置它不像容器配置可以在一个类中支持多个不同的Handler方法以处理prepareColdDrink方法为例

　　 @MessageEndpoint(input= coldDrinks ) //加了该元数据它会自动扫描并作为@Componet标记处理 public class Barista private long hotDrinkDelay = ; private long coldDrinkDelay = ; private AtomicInteger hotDrinkCounter = new AtomicInteger(); private AtomicInteger coldDrinkCounter = new AtomicInteger(); public void setHotDrinkDelay(long hotDrinkDelay) this hotDrinkDelay = hotDrinkDelay; public void setColdDrinkDelay(long coldDrinkDelay) ldDrinkDelay = coldDrinkDelay; public void prepareHotDrink(Drink drink) try Thread sleep(this hotDrinkDelay); catch (InterruptedException e) Thread currentThread() interrupt(); System out println( prepared hot drink # + hotDrinkCounter incrementAndGet() + : + drink); @Handler//回调处理的方法 public void prepareColdDrink(Drink drink) try Thread sleep(ldDrinkDelay); catch (InterruptedException e) Thread currentThread() interrupt(); System out println( prepared cold drink # + coldDrinkCounter incrementAndGet() + : + drink);

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

　　初充下面是针对 Spring Integration adapter扩展的学习笔记

　　JMS Adapters

　　jms adapters 目前有两种实现

　　JmsPollingSourceAdapter 和 JmsMessageDrivenSourceAdapter 前者是使用Srping的JmsTemplate模板类通过轮循的方式接收消息

　　后者是使用则通过代理Spring的DefaultMessageListenerContainer实例实现消息驱动的方式

　　xml配置如下

　　JmsMessageDrivenSourceAdapter

　　另外还有一个比较有用的类JmsTargetAdapter 它实现了MessageHandler接口它提把Spring Integration Message对象转换成JMS消息并发送到指定的消息队列与JMS服务连接的实现可以通过设定 jmsTemplate属性引用或是connectionFactory和destination或destinationName属性

cha138/Article/program/Java/ky/201311/28092