EIP——Spring Integration4.3使用篇

曾经终败给现在 2022-07-14 01:50 197阅读 0赞

### 文章目录 ###

*   *  2-3、 Message Endpoint 消息终端

\#1、前言  
\#\#1-1、兼容Java版本  
使用Spring integration 4.3.x，最小的兼容java 版本是Java SE 6. 更旧的java版是不支持的  
Spring integration4.3.x完美兼容 java SE 7 和java SE 8  
具体的功能(如 `Option<?>` payloads 和 `CompletableFuture`gateway 方法返回类型 )需要java 8或以上

\#\#1-2、兼容 Spring 框架的版本  
Spring integration 4.3.x 需要 Spring framework 4.3及以上

\#\#1-3、代码约定  
Spring integration 命名空间使用前缀 **int**，每个Spring integration适配器（模块）都将提供它自己的命名空间，它的配置如下所示  
**int-随后的是不同模块的名称，如int-twitter**, **int-stream**, ……

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xmlns:int="http://www.springframework.org/schema/integration"
      xmlns:int-twitter="http://www.springframework.org/schema/integration/twitter"
      xmlns:int-stream="http://www.springframework.org/schema/integration/stream"
      xsi:schemaLocation="
       http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/integration
       http://www.springframework.org/schema/integration/spring-integration.xsd
       http://www.springframework.org/schema/integration/twitter
       http://www.springframework.org/schema/integration/twitter/spring-integration-twitter.xsd
       http://www.springframework.org/schema/integration/stream
       http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd">
    …
    </beans>

\#\#1-4、目标和原则  
Spring Integration主要基于以下目标：

*  提供一个简单的模型来实现复杂的企业集成解决方案
 *  为基于Spring的应用添加异步的、消息驱动的行为
 *  让Spring用户可以直观的、增量的采用

Spring Integration基于以下原则：

*  组件应该松耦合，方便模型化和易测试
 *  框架应该强制业务逻辑和集成逻辑的关注点分离
 *  扩展点本质上应该是抽象的，而且限定在一个清晰的边界内，来促进可重用性和可移植性。

\#\#2、主要组件  
纵向地看，一个分层的架构利于关注点的分离，各层间基于接口的契约实现了松耦合。典型地，基于Spring的应用就是这样设计的，Spring框架和工具集提供了一个遵从最佳实践、全栈式开发企业级应用的强大基础。横向地看，消息驱动架构添加了横向视图，虽然这些目标还是相关的。分层架构是一个非常通用和抽象的范式，消息系统非常符合同样抽象的“管道和过滤器”模型。“过滤器”代表任何能够生产和（或）消费消息的组件，“管道”在过滤器间传输消息，所以组件间保持松耦合。值得指出的是，这两个高级范式不是互斥的。其中的消息基础设施支持包装“管道”到契约被定义为接口的层。同样地，“过滤器”一般被管理在逻辑上位于应用系统的服务层之上的层中，其与那些服务在相同的方式下通过接口交互。

\#\#2-1、Message 消息  
在Spring Integration中，消息是任何Java对象连同框架处理对象时使用的元数据的一个通用包装。它由payload和headers组成。 payload可以是任何类型，headers通常一些必须的数据，如id，timestamp，correlation id, 和返回address。 Headers也被用作存储传输协议的变量。举个例子，当要接收一个文件，要创建一条消息，可以把文件名会存在header中，供下游组件进行存取。同样地，当使用对外接出的邮件适配器(outbound mail adatper)发出一段消息内容，各种属性（如cc, from , subject,等等），会被下游组件配置到Message header中。  
开发人员可以 存储任意的keay-value在header中  
**如下图所示 Message的结构**

![这里写图片描述][message.jpg]

\#\#2-2、Message Channel 消息通道  
一个消息通道代表“管道和过滤器”架构中的一个“管道”。消息生产者发送消息到通道，消息消费者从通道接收消息。消息通道因此解耦了消息组件，同时也提供了消息拦截和监控的切入点。

**如下图所示 Message Channel 的结构**  
![这里写图片描述][channel.jpg]

一个消息通道可能符合点对点模式或者发布-订阅模式。如果是点对点模式的通道，发布到通道中的每个消息，最多只有一个消费者可以接收。如果是发布-订阅模式的通道，则会尝试广播每个消息给其所有的订阅者。Spring Integration支持这两种模式。

鉴于“点对点模式”和“发布订阅模式”定义了每个消息最终有多少消息消费者接收的两种方式，这里有一个重要的考虑：通道是否应该缓冲消息？Spring Integration中，轮询通道（Pollable Channels）具有在一个队列中缓冲消息的能力。缓冲的优势在于它能够调节接入消息流量，从而防止系统负荷过载。然而，正如其名称所示，这也增加了一些复杂性，只有配置了轮询器后，一个消息消费者才能从这个通道中接收消息。另外来说，订阅通道（Subscribable Channel）要求连接它的消费者依从简单的消息驱动模式。

## 2-3、 Message Endpoint 消息终端 ##

Spring Integration的主要目标之一是通过控制反转简化企业集成解决方案的开发。这意味着你应该不需要直接实现消息消费者和生产者，更不需要在消息通道中构建和调用发送接收操作的细节。相反地，你只需要关注于你基于普通对象实现的特定领域模型。然后，通过声明式的配置，你可以“连接”你的领域特定代码到Spring Integration提供的消息基础设施。负责这些连接的代码是消息终端。这并不是说必须直接连接到现有应用的代码。任何现有的企业集成解决方案，都需要一些用于集成相关的代码，例如路由选择和协议转换。其中的一个要点就是实现集成逻辑和业务逻辑关注点的分离。换句话说，作为web应用中的MVC模式，其目标应该是提供一个简单而专用的层，转换接入的请求到服务层调用，然后转换服务层返回值到接出的响应。

\#3、消息端点的分类

\#\#3-1、 Transformer 转换器  
消息转换器负责转换一个消息的内容或结构，并返回翻修改后的消息。最常见的转换器可能是把消息负载从一种格式转换成另一种格式的转换器（例如从XML文档转换成java.lang.String字符串）。同样地，转换器也可以被用于添加、删除和修改消息头中的值。

\#\#3-2、 Filter 过滤器

消息过滤器最终决定消息是否被发送到输出通道。这简单地依赖于一个布尔型的测试方法，它可以检查消息是否包含特定的负载内容的类型，一个属性值，是否包含消息头等等。如果一个消息被接受，它将被发送到输出通道，否则消息将被丢弃（或者在一个更严格的实现里，应该抛出一个异常信息）。消息过滤器通常和发布订阅通道一起使用，此时多个消费者接收到相同的消息，而在某些条件下使用过滤器可以减少需要处理的消息数量。

\#\#3-3、Router 路由  
消息路由负责决定下个消息将由哪些通道接收。典型地，这种决定是基于消息的内容和（或）消息头部中的可用元数据。一个消息路由常常作为动态配置输出通道的一个动态选择，用在消息激活器或是其它的能发送响应消息的端点上。同样地，相对于被动的消息过滤器方式，消息路由提供一个主动的方式被多个订阅者使用，就像上面描述的那样。  
**如下图所示**  
![这里写图片描述][router.jpg]

\#\#3-4、 Splitter分解器

消息分解器是另外一种类型的消息端点，它负责从一个输入通道中接收一个消息，把消息分解成多个消息，然后把它们发送到相应的输出通道。一个典型的例子就是把一个复合的负载对象分解成一组包含各个子负载的消息。

\#\#3-5、 Aggregator 聚合器  
集合器基本上是分解器的镜像，也是一种消息端点，它接收多个消息，然后把他们合并成一个单独的消息。事实上，聚合器经常和管道中的分解器一起用来减少消费者数量。技术上来说，聚合器比分解器更复杂，因为它需要维持状态（被聚合的多个消息），决定整组消息何时可用，以及必要时处理超时。此外，在超时的情况下，聚合器需要知道是发送部分的消息，还是丢弃它们到一个单独的通道。Spring Integration提供了可配置的超时处理策略，在超时时发送部分结果，还是发送到收集丢弃消息的通道。  
\#\#3-6、Service Activator 服务催化剂  
在消息系统中，服务催化剂是一种通用的消息端点用来调用服务实例。输入消息通道必须是配置好的，并且服务的方法调用后须返回一个值，然后提供一个输出消息的通道。  
服务器催化剂调用服务实例对象中的方法来响应请求消息，提取请求消息的payload，需要的话可能要做消息转换。不管怎样，服务对象方法须返回值，这个返回值同样地会转换成一条回复消息。这条回复消息会被发送到输出通道。如果没有配置输出通道，这条回复内容会被发送给对应名称的通道

**消息路由模式，如下图所示**

<table> 
 <thead> 
  <tr> 
   <th>模式</th> 
   <th align="center">消息消费数</th> 
   <th align="right">消息发布数</th> 
   <th align="right">是否有状态</th> 
   <th align="right">评论</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Content-Based Router</td> 
   <td align="center">1</td> 
   <td align="right">1</td> 
   <td align="right">No (mostly)</td> 
   <td align="right"></td> 
  </tr> 
  <tr> 
   <td>Message Filter</td> 
   <td align="center">1</td> 
   <td align="right">0 or 1</td> 
   <td align="right">No (mostly)</td> 
   <td align="right"></td> 
  </tr> 
  <tr> 
   <td>Recipient List</td> 
   <td align="center">1</td> 
   <td align="right">multiple (incl. 0)</td> 
   <td align="right">No</td> 
   <td align="right"></td> 
  </tr> 
  <tr> 
   <td>Splitter</td> 
   <td align="center">1</td> 
   <td align="right">multiple</td> 
   <td align="right">No</td> 
   <td align="right"></td> 
  </tr> 
  <tr> 
   <td>Aggregator</td> 
   <td align="center">multiple</td> 
   <td align="right">1</td> 
   <td align="right">Yes</td> 
   <td align="right"></td> 
  </tr> 
  <tr> 
   <td>Resequencer</td> 
   <td align="center">multiple</td> 
   <td align="right">multiple</td> 
   <td align="right">Yes</td> 
   <td align="right">Publishes same number it consumes</td> 
  </tr> 
 </tbody> 
</table>

\#\#3-7、Channel Adapter 通道适配器

通道适配器通过连接消息通道到其它的子系统或传输接口的端点。通道适配器可以接入或接出的。典型地，通道适配器可以在消息和其他任何发送到/接收于其他系统（文件、HTTP请求，JMS消息等等）的对象或资源之间做一些映射。依赖于传输接口，通道适配器也可以构造或扩展消息header的值。  
**下图所示一个接入通道适配器端点连接一个源系统到一个消息通道**  
![这里写图片描述][source-endpoint.jpg]  
**下图所示一个接出通道适配器端点连接一个消息通道到一个目标系统**  
![这里写图片描述][target-endpoint.jpg]

\#4、消息核心  
\#\#4-1、消息通道  
Message作为封装数据至关重要的角色，那么MessageChannel的作用是解耦消费者与生产者

\#\#\#4-1-1、消息通道接口定义  
代码如下

public interface MessageChannel {
    
        boolean send(Message message);
    
        boolean send(Message message, long timeout);
    }

当进行发送消息是，如果返回值为 true，则说明该消息成功发送， 如果发送超时或者被中断，方法将返回 false，表示消息发送失败。

**PollableChannel**  
开始提到 消息通道支持两种模式，是否对支持消息缓存。如下的两个子接口分别针对这两种不同需求进行了定义： pollable\[支持缓存\] 与 subscribable\[不支持缓存\] 通道，如下为PollableChannel类代码

public interface PollableChannel extends MessageChannel {
    
        Message<?> receive();
    
        Message<?> receive(long timeout);
    
    }

当进行发送消息是，如果返回值为 true，则说明该消息成功发送， 如果发送超时或者被中断，方法将返回 false，表示消息发送失败  
**SubscribableChannel**  
SubscribableChannel 继承了 MessageChannel 接口，并再次基础上提供了发送消息给订阅者的接口。 因此，他没有提供用户轮询接收消息的方法。代码如下

public interface SubscribableChannel extends MessageChannel {
    
        boolean subscribe(MessageHandler handler);
    
        boolean unsubscribe(MessageHandler handler);
    
    }

\#\#\#4-1-2、通道实现类  
Spring Integration提供了几种不同消息通道的实现类，每个介绍如下描述  
**PublishSubscribeChannel**  
PublishSubscribeChannel(广播订阅通道) 实现广播发送给他的任何消息给全部的订阅者。 这是最常见的用于发送事件消息，其主要作用是通知，而不是文件的消息一般都是为了要处理一个单一的处理。 注意 PublishSubscribeChannel 的send 方法是可以被直接调用的。因为广播的动作是在 send 触发后，自动调用的。 所以 作为消息订阅者，必须拥有一个 MessageHandler 并且注册到 PublishSubscribeChannel 上。 这个不同于 PollableChannel 接口的 receive() 方法，该方法是被客户端自动触发的

**PriorityChannel**  
由于 QueueChannel 强制实现 先进先出 的消息消费模式，而 PriorityChannel 实现允许消息在队列内改变 自己顺序的功能（基于优先级）。默认的策略属性 定义在消息头中 的 priority映射的属性。但是需要自定义优先级判断逻辑 ，比较器实现了`Comparator<Message<?>>`接口，通过调用该接口实现 优先级策略。  
**QueueChannel**  
QueueChannel 类内部封装了一个 队列用户存储消息。QueueChannel 支持 点对点 的传输策略。 通俗理解：即便QueueChannel 同时拥有多个消费者时，一个消息仅能给其中之一的消费者持有\[广播模式对消息进行了复制操作 \])，同时 该类提供了无参数的构造函数\[队列的容量默认是Integer.MAX\_VALUE\]，同时也提供了一个指定容量大小 的构造函数

当QueueChannel内部的队列容量没有达到使用上限，那么，调用send() 方法发送消息给channel时则会立刻返回， 消息被缓存于消息队列中等待被接收者接受。当内部队列达到上限时，这是在调用 send() 方法，将被阻塞，直到 队列出现空位或者超时被终端才返回。这种情况同样适用于接受。如果队列中存在自己需要的消息，则直接返回。当队列为空时， 接收方法同样阻塞，直到有消息到达或者超时被中断。当然，如果强制想让 receive() 方法不管是否接受到消息都 立即返回，则设置 超时时间参数为0即可。如果没有参数，阻塞将被无限期延长。  
**RendezvousChannel**  
RendezvousChannel 实现 “direct-handoff” 功能，当一个消息生产者发送消息到 RendezvousChannel后将被阻塞， 直到消费者调用 该 channel的 receive() 方法\[反之亦然\]。其内部实现是这样子的， 他实现很类似 QueueChannel 除了他使用的队列是 SynchronousQueue\[JDK5 有标准提供，读者自己翻阅资料：队列大小 为 0 的 BlockingQueue的实现。\]，RendezvousChannel 是为了实现当 消费 与 生产 消息位于不同的线程时进行同步 使用。这样发送者可以明确知道消息是否被消费。而QueueChannel 则无法提供该功能的反馈。  
**DirectChannel**  
DirectChannel 从字面理解是 基于 点对点 传输模式。事实上，他更像是一个 PublishSubscribeChannel 的实现，除了内部实现消息队列机制以外。 该类实现了 SubscribableChannel 接口 而不是 PollableChannel 接口。所以他会直接分发消息给订阅者。 同时作为 点对点 的传输队列，他有不同于传统意义的 PublishSubscribeChannel。他每次仅仅发消息发送给一个订阅者

由于 DirectChannel 的简单性，使其无需增加任何额外的开销即可提供调度以及管理线程轮询，这是 Spring Integration 内部默认实现的channel。如果顶一个应用程序的channel，首先需要考虑的是 该channel是否需要提供队列缓存功能，之后是是否需要提供基于优先级策略的PollableChannels， 如果你仅仅是为了实现消息广播，则采用PublishSubscribeChannel是正确的选择而不是DirectChannel

DirectChannel 可以充当消息分发的角色定位，并且该分发是基于负载均衡策略。负载均衡实现了对订阅该通道 的消息订阅者按照顺序循环调用。其内部实现的默认策略是 循环 机制。这个是负载均衡的最基本的一种实现。

<int:channel id="lbRefChannel">
      <int:dispatcher load-balancer-ref="lb"/>
    </int:channel>
    
    <bean id="lb" class="foo.bar.SampleLoadBalancingStrategy"/>

负载均衡 常与 故障转移 配合使用，对于DirectChannel的 failover 值，默认是 开启的。当前一个订阅者调用发生异常，分发器会回滚当前状态到下一个订阅者重新处理，其调用的顺序是由handlers自身的 order 属性决定。如果不存在这样的值 则由订阅者默认的顺序决定。  
**ExecutorChannel**  
ExecutorChannel 是 点对点 通道的实现，同时他支持 DirectChannel 的分发器特性，（负载均衡与故障转移）。 两者之间一个显著的不同之处是 ExecutorChannel 内部是通用一个 TaskExecutor 实例实现的。这意味着调用该 通道的 send 方法不会发生阻塞，同时这将意味着 对handler的调用将不会与sender 公用统一线程， 该模式不支持事务，发送与接受分别受限于不同的线程  
注意在某些场合下，调用ExecutorChannel 的send 也有可能被阻塞。例如 ，当使用的 TaskExecutor 节流以一个数量限制的客户端（线程池），当池中没有可用的线程是，或者执行队列 当前没有工作的队列，则会被阻塞。因为这种情况的发生是不可预测的，所以它不能依赖于事务  
**Scoped Channel**  
在Spring Integration 1.0 版本中，提供了基于线程范围的 ThreadLocalChannel实例，但是在2.0中被移除了。 现在提供一个更加通用的方法来满足该需求，那就是通过 channel 的scope的属性。该属性的值可以是在Spring 容器中可被识别的任何值。例如在 web环境中，一些特定的Scope 是可用的。 并且这些都已经被注册到Spring 容器中。如下的代码 展示了 基于ThreadLocal 的配置scope在一个通道中，包括这个线程Scope 到Spring容器，示例代码如下所示

<int:channel id="threadScopedChannel" scope="thread">
         <int:queue />
    </int:channel>
    
    <bean class="org.springframework.beans.factory.config.CustomScopeConfigurer">
        <property name="scopes">
            <map>
                <entry key="thread" value="org.springframework.context.support.SimpleThreadScope" />
            </map>
        </property>
    </bean>

上述的channel 同时作用于其内部的队列。由于channel 绑定了当前的工作线程，所以其内部队列同样被作用。 这样线程发送给通道可能晚于接收同样的消息。但其他线程是有权限访问他们的。事实上，基于thread-scoped 的channle 很少被用到。一个比较有用的场景是 当通过 DirectChannels 在强制在单线程下执行一个消息发送后，任何响应的消息 可以发送到这个终端的 channel 中。如果这个channel 是基于 thread-scoped，那么原先的发送线程可以通过他获取任何响应的消息  
\#\#\#4-1-3、Channel Interceptors 通道拦截器  
作为消息传输架构的另外一个高级特性就是通过非侵入的方式从传输的消息中提供有效的数据。 当消息通过 消息通道 从开始发送到被接受，消息拦截器会在发送和接受时的进行拦截。

通道拦截器接口定义了如下方法：

public interface ChannelInterceptor {
    
        Message<?> preSend(Message<?> message, MessageChannel channel);
    
        void postSend(Message<?> message, MessageChannel channel, boolean sent);
    
        void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex);
    
        boolean preReceive(MessageChannel channel);
    
        Message<?> postReceive(Message<?> message, MessageChannel channel);
    
        void afterReceiveCompletion(Message<?> message, MessageChannel channel, Exception ex);
    }

实现了这个接口后，通过如下的方式注册到 指定的 channel 中

channel.addInterceptor(someChannelInterceptor);

preReceive()允许返回 false 值，从而避免接收者继续进行操作

Spring Integration 同样也提供了 窃听 模式的拦截机制，它是一种简单的拦截实现，当发送的消息被他拦截时， 他会发送一个消息到指定的 channel中，而不影响原先的 channle处理流程。 这个对于程序调试以及监控是很有用的，可以异步执行

由于在实际情况下，很少需要实现全部的interceptor接口定义的方法。所以系统也提供了 ChannelInterceptorAdapter 适配器，他实现了全部接口的空方法 … 在实际开发的过程中，开发者按需重载指定的方法，从而实现拦截的功能。

public abstract class ChannelInterceptorAdapter implements ChannelInterceptor {
    
    	@Override
    	public Message<?> preSend(Message<?> message, MessageChannel channel) {
    		return message;
    	}
    
    	@Override
    	public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
    	}
    
    	@Override
    	public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
    	}
    
    	public boolean preReceive(MessageChannel channel) {
    		return true;
    	}
    
    	@Override
    	public Message<?> postReceive(Message<?> message, MessageChannel channel) {
    		return message;
    	}
    
    	@Override
    	public void afterReceiveCompletion(Message<?> message, MessageChannel channel, Exception ex) {
    	}
    
    }

public class CountingChannelInterceptor extends ChannelInterceptorAdapter {
    
        private final AtomicInteger sendCount = new AtomicInteger();
    
        @Override
        public Message<?> preSend(Message<?> message, MessageChannel channel) {
            sendCount.incrementAndGet();
            return message;
        }
    }

\#\#\#4-1-4、 MessagingTemplate  
Spring integration 提供一个基本的消息组件，可以无侵入式的调用消息系统中的代码。然而，有时也可以 通过反射的方式调用消息系统的代码。为了方便地使用这些实现的案例，Spring integration提供了MessagingTemplate类，它支持通过Messages Channel来调用各种操作，包括请求/返回这些场景。  
例如，如下代码所示，发送一个请求同时等待回复

MessagingTemplate template = new MessagingTemplate();
    
    Message reply = template.sendAndReceive(someChannel, new GenericMessage("test"));

在例子中，template可以 创建一个以类部类的形式创建通道。在Template 中设置*sendTimeout*和*receiveTimeout*属性值，同时其它的一些传输的类型也可以被支持的

public boolean send(final MessageChannel channel, final Message<?> message) { ...
    }
    
    public Message<?> sendAndReceive(final MessageChannel channel, final Message<?> request) { ..
    }
    
    public Message<?> receive(final PollableChannel<?> channel) { ...
    }

\#\#\#4-1-5、 配置消息通道

为了创建一个消息通道的实例，你可以使用*channel*元素

<int:channel id="exampleChannel"/>

默认的通道模式是点对点的。为了 创建一个订阅发布通道，可以使用*publish-subscribe-channel*元素

<int:publish-subscribe-channel id="exampleChannel"/>

当直接使用`<channel/>` 元素时，它是不包括子元素的，它将会创建`DirectChannel`实例

\#\#\#\#4-1-5-1、DirectChannel 配置  
如上面提到的，`DirectChannel`是默认的模式

<int:channel id="directChannel"/>

\#\#\#\#4-1-5-2、 Datatype Channel配置  
创建datatype channel指定消息为一个确定的payload 类型，`datatype`属性必须是类的全路径名称

<int:channel id="numberChannel" datatype="java.lang.Number"/>

下面给一个错误的范例

MessageChannel inChannel = context.getBean("numberChannel", MessageChannel.class);
    inChannel.send(new GenericMessage<String>("5"));

执行程序后，由于我们定义的是java.number.Number类型，所以会输出以下异常

Exception in thread "main" org.springframework.integration.MessageDeliveryException:
    Channel 'numberChannel'
    expected one of the following datataypes [class java.lang.Number],
    but received [class java.lang.String]
    …

\#\#\#\#4-1-5-3、QueueChannel 配置  
创建`QueueChannel`可以 使用`<queue/>`子元素，你可以像下面配置

<int:channel id="queueChannel">
        <queue capacity="25"/>
    </int:channel>

默认地，'QueueChannel '会将消息存放在内存中，因此可能会存在丢失消息的情况。然而，Spring Integration提供多种持久化的存储方式，可以使用如`JdbcChannelMessageStore`

<int:channel id="dbBackedChannel">
        <int:queue message-store="channelStore"/>
    </int:channel>
    
    <bean id="channelStore" class="o.s.i.jdbc.store.JdbcChannelMessageStore">
        <property name="dataSource" ref="dataSource"/>
        <property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
    </bean>

\#\#\#\#4-1-5-4 PublishSubscribeChannel 配置  
创建`PublishSubscribeChannel`，可以 使用`<publish-subscribe-channel/>`元素，当使用该元素，你可以 配置`task-executor`属性来标识发送消息

<int:publish-subscribe-channel id="pubsubChannel" task-executor="someExecutor"/>

\#\#\#\#4-1-5-5、ExecutorChannel配置  
创建`ExecutorChannel`，增加子元素`<dispatcher>`，同时再里面配置属性`task-executor`

<int:channel id="executorChannel">
        <int:dispatcher task-executor="someExecutor"/>
    </int:channel>

\#\#\#\#4-1-5-6、PriorityChannel 配置  
创建PriorityChannel ，增加子元素`<priority-queue/>`

<int:channel id="priorityChannel">
        <int:priority-queue capacity="20"/>
    </int:channel>

--------------------

\#\#\#\#4-1-5-7、通道拦截器配置  
元素`<interceptors/>`作为子元素添加到 `<channel/>`中，提供`ref`属性指向Spring管理的对象，该对象须实现`ChannelInterceptor`接口

<int:channel id="exampleChannel">
        <int:interceptors>
            <ref bean="trafficMonitoringInterceptor"/>
        </int:interceptors>
    </int:channel>

\#\#\#\#4-1-5-8、全局通道拦截器配置  
通道拦截器提供了一种简洁的方式横切每一 个通道。如果以同样的方式来配置每一个通道，配置一堆的拦截器，这种方式不是高效的方式。为了避免重复的配置，Spring Interceptors提供了一种全局拦截的配置方式，如下所示

<int:channel-interceptor ref="myInterceptor" pattern="input*, bar*, foo" order="3"/>
    
    <bean id="myInterceptor" class="foo.barSampleInterceptor"/>

通过元素`channel-interceptor`定义全局拦截器，它会根据patterns定义的规定，去拦截适配的通道，在上面例子中，该拦截器会拦截所有前缀为input、bar、foo的通道。order属性用于当多个拦截器去拦截同一个通道时，它们的执行顺序。例如，通道*inputChannel*添加某个拦截器，如下所示

<int:channel id="inputChannel"> 
      <int:interceptors>
        <int:wire-tap channel="logger"/> 
      </int:interceptors>
    </int:channel>

一个问题是：全局拦截器跟其它配置的本地拦截器或者其它全局拦截，它们之间的关系是怎样的？`order`属性值用于保证拦截的执行顺序，order是正数时，它执行晚于其它存在的拦截器或者是值为负数的拦截器。  
这就意味着在上面的例 子中，全局拦截器的order值大于0，所以它会晚于wire-tap执行。注：默认order的值为0，默认pattern的值是\*，拦截所有的通道  
拦截顺序总结：order值越小越先执行

\#\#\#\#4-1-5-9 Wire Tap (窃听器)  
你可以将*Wire Tap*配置到元素`<interceptors>`中，这种方式在调试时特别有效，它也可以与Spring Integration日志通道适配器一起使用，如下所示

<int:channel id="in">
        <int:interceptors>
            <int:wire-tap channel="logger"/>
        </int:interceptors>
    </int:channel>
    
    <int:logging-channel-adapter id="logger" level="DEBUG"/>

\#\#\#4-1-6 特殊通道  
在Spring 上下文中，定义了两个默认的通道:`errorChannel`和`nullChannel`。`nullChannel`类似于`/dev/null`，如debug级别及以下的日志消息都会舍弃

\#\#4-2 poller 轮询者

\#\#4.3 Channel Adapter通道适配器  
通道适配器作为一个消息端点，连接了发送者或接收者和消息通道。Spring integration提供了很多开箱即用的传输协议，如JMS，File，HTTP，Web Service， Maill等等。它包含出站和入站适配器，它们可以通过XML元素进行配置。只要你提供源或目标对象，Spring integration通过反射的方式，调用方法来实现

\#\#\#4-3-1、 接出通道适配器配置  
元素`inboud-channel-adapter`可以通过反射方式调用被Spring 托管对象的方法，将它转化成Message后，传给`MessageChannel`

<int:inbound-channel-adapter ref="source1" method="method1" channel="channel1">
        <int:poller fixed-rate="5000"/>
    </int:inbound-channel-adapter>
    
    <int:inbound-channel-adapter ref="source2" method="method2" channel="channel2">
        <int:poller cron="30 * 9-17 * * MON-FRI"/>
    </int:channel-adapter>

\#\#\#4-3-2、 接入通道适配器配置  
元素`outbound-channel-adapter`，连接`MessageChannel`和POJO消费者，通过反射调用方法，然后将包含有paloyad的Message传给channel

<int:outbound-channel-adapter channel="channel1" ref="target" method="handle"/>
    
    <beans:bean id="target" class="org.Foo"/>

或

<int:outbound-channel-adapter channel="channel2" ref="target" method="handle">
        <int:poller fixed-rate="3000" />
    </int:outbound-channel-adapter>
    
    <beans:bean id="target" class="org.Foo"/>

\#\#\#4-3-3、通道适配器表达式和脚本使用  
`<int:inbound-channel-adapter/>`元素可以 配置SpEL `<expression/>`或`<script/>`子元素

<int:inbound-channel-adapter ref="source1" method="method1" channel="channel1">
        <int:poller max-messages-per-poll="1" fixed-delay="5000"/>
        <script:script lang="ruby" location="Foo.rb" refresh-check-delay="5000"/>
    </int:inbound-channel-adapter>

\#\#\#4-4、 Messaging Bridge消息桥梁  
消息桥梁用于连接两个消息通道或通道适配器，简单的使用*input-channel*和*output-channel*属性

<int:bridge input-channel="input" output-channel="output"/>
    <int:bridge input-channel="pollable" output-channel="subscribable">
         <int:poller max-messages-per-poll="10" fixed-rate="5000"/>
     </int:bridge>

\#5、消息结构  
\#\#5-1、Message  
Spring integration `Message`是一个简单的数据容器。payload可以是任意的对象类型。每一个`Message`也会包含header，headers也可以 由用户扩展它里面的键值对  
\#\#\#5-1-1、消息接口定义  
如下代码所示

public interface Message<T> {
    
        T getPayload();
    
        MessageHeaders getHeaders();
    
    }

\#\#\#5-1-2、消息Headers类  
如下代码所示

public final class MessageHeaders implements Map<String, Object>, Serializable {
      ...
    }

由于实现了Map，headers可以 通过`get(..)`方法获取header的参数值。另外，你可以提供额外的参数Class类型。甚至更方便的，我人产可以通过预设定的属性，方便地通过其对应的get方法取值。下面是三种列表的例子

Object someValue = message.getHeaders().get("someKey");
    
     CustomerId customerId = message.getHeaders().get("customerId", CustomerId.class);
    
     Long timestamp = message.getHeaders().getTimestamp();

以下是headers预设的属性  
![这里写图片描述][20161123112256318]

\#\#\#5-1-3、Message的实现类  
实现自`Message`的类`GenericMessage<T>`，它提供以下两种构造方法

new GenericMessage<T>(T payload);
    
    new GenericMessage<T>(T payload, Map<String, Object> headers)

当消息被创建后，会生成随机唯一的ID，同时构造函数会把根据用户传进来的headers属性值，加入到header中的map

\#\#\#5-1-4、MessageBuilder Helper类  
大家可能也注意到了，Message接口没有定义set方法，来设置payload和headers的值。原因是，由于当message创建后，是不允许被修改的。因此，当一个Message实例被发送给多个消费者，如果其中一个消费者需要回复一个不同类型payload数据，它需要重新实例化一条消息。换句话说，消息结构就类似于*unmodifiable collection*

为了方便实例GenericMessage，Spring Integeration提供了一种更方便的构建Message:`MessageBuilder`。这个MessageBuilder提供了两种工厂方法来从已有的Message或payload对象创建消息实例。如下代码所示

Message<String> message1 = MessageBuilder.withPayload("test")
            .setHeader("foo", "bar")
            .build();
    
    Message<String> message2 = MessageBuilder.fromMessage(message1).build();
    
    assertEquals("test", message2.getPayload());
    assertEquals("bar", message2.getHeaders().get("foo"));

\#6、Message Routing消息路由  
\#\#6-1、路由  
\#\#\#6-1-1、概览  
路由是消息框架中关键的元素，他们从消息通道中消费消息，同时为传给的通道设置一系列的条件。  
Spring Integration提供以下开箱即用的路由：

*  payload 类型的路由
 *  Header 值路由
 *  Recepient List路由
 *  XPath路由
 *  Error Message Exception 类型路由
 *  通用路由

\#端点汇总  
\#11、AMQP 支持  
\#\#\#11-7、异步输出网关  
前面部分介绍的网关是同步。它是发送线程在等待回复时会一直挂起(可能会超时)。Spring Integration 4.3版本开始添加异步网关。可以使用Spring AMQP中`AsyncRabbitTemplate`类实现。当消息发出后，发送完成的事后线程会立即返回。template 监听容器会将收到消息回复。在轮询线程调用网关时，会非常有用。这个线程会被释放并被框架中的其它任务使用。  
以下是对AMQP接出网关的配置

<int-amqp:outbound-gateway id="inboundGateway" 1
                               request-channel="myRequestChannel" 2
                               async-template="" 3
                               exchange-name="" 4
                               exchange-name-expression="" 5
                               order="1" 6
                               reply-channel="" 7
                               reply-timeout="" 8
                               requires-reply="" 9
                               routing-key="" 10
                               routing-key-expression="" 11
                               default-delivery-mode"" 12
                               confirm-correlation-expression="" 13
                               confirm-ack-channel="" 14
                               confirm-nack-channel="" 15
                               return-channel="" (16)
                               lazy-connect="true" /> (17)

参考

> [http://docs.spring.io/spring-integration/reference/htmlsingle/][http_docs.spring.io_spring-integration_reference_htmlsingle]

[message.jpg]: /images/20220714/e10fb7b896944ef6b299fe1a87cfe812.png
[channel.jpg]: /images/20220714/715f564ebd6945d49ccbca3119b0e536.png
[router.jpg]: /images/20220714/bfcda177c6eb4529bc479d926428a688.png
[source-endpoint.jpg]: /images/20220714/aab4662ee1df48f89ae94b49bd728a04.png
[target-endpoint.jpg]: /images/20220714/b9a9344fdd01496396dabd2c8c278b9c.png
[20161123112256318]: /images/20220714/08a74ada47a1429a9d973e219082ff2a.png
[http_docs.spring.io_spring-integration_reference_htmlsingle]: http://docs.spring.io/spring-integration/reference/htmlsingle/