Spring Integration 系统集成

梦里梦外; 2022-07-14 02:19 221阅读 0赞

Spring Ingegration 提供了基于Spring的EIP(Enterprise Integration Patterns,企业集成模式)的实现。Spring Integration 主要解决的问题是不同系统之间交互的问题，通过异步消息驱动来达到系统交互时系统之间的松耦合。

Spring Integration 主要有**Message、Channel、Message EndPoint**组成。

**Message**

Message是用来在不同部分之间传递的数据。Message有两部分组成:消息体(payload)与消息头(header)。消息体可以是任何数据类型；消息头表示的元数据就是解释消息体的内容。

/** * A generic message representation with headers and body. * * @author Mark Fisher * @author Arjen Poutsma * @since 4.0 * @see org.springframework.messaging.support.MessageBuilder */
    public interface Message<T> {
    
        /** * Return the message payload. */
        T getPayload();
    
        /** * Return message headers for the message (never { @code null} but may be empty). */
        MessageHeaders getHeaders();
    
    }

**Channel**

在消息系统中，消息发送者发送消息到通道(Channel)，消息接受者从通道(Channel)接收消息。

1、顶级接口

(1) **MessageChannel**

MessageChannel 是Spring Integration消息通道的顶级接口：

public interface MessageChannel {
    
        /**
         * Constant for sending a message without a prescribed timeout.
         */
        long INDEFINITE_TIMEOUT = -1;
    
    
        /**
         * Send a {@link Message} to this channel. If the message is sent successfully,
         * the method returns {@code true}. If the message cannot be sent due to a
         * non-fatal reason, the method returns {@code false}. The method may also
         * throw a RuntimeException in case of non-recoverable errors.
         * <p>This method may block indefinitely, depending on the implementation.
         * To provide a maximum wait time, use {@link #send(Message, long)}.
         * @param message the message to send
         * @return whether or not the message was sent
         */
        boolean send(Message<?> message);
    
        /**
         * Send a message, blocking until either the message is accepted or the
         * specified timeout period elapses.
         * @param message the message to send
         * @param timeout the timeout in milliseconds or {@link #INDEFINITE_TIMEOUT}
         * @return {@code true} if the message is sent, {@code false} if not
         * including a timeout of an interrupt of the send
         */
        boolean send(Message<?> message, long timeout);
    
    }

MessageChannel 有两大子接口，分别是PollableChannel (可轮询)和SubscribableChannel(可订阅)。我们所有的消息通道类都是现实这两个接口。

(2) **PollableChannel**

PollableChannel 具备轮询获得消息的能力。

public interface PollableChannel extends MessageChannel { 
    
        /** * Receive a message from this channel, blocking indefinitely if necessary. * @return the next available {@link Message} or {@code null} if interrupted */
        Message<?> receive();
    
        /** * Receive a message from this channel, blocking until either a message is available * or the specified timeout period elapses. * @param timeout the timeout in milliseconds or {@link MessageChannel#INDEFINITE_TIMEOUT}. * @return the next available {@link Message} or {@code null} if the specified timeout * period elapses or the message reception is interrupted */
        Message<?> receive(long timeout);
    
    }

(3) **SubscribableChannel**

SubscribableChannel 发送消息给订阅了MessageHanlder的订阅者

public interface SubscribableChannel extends MessageChannel { 
    
        /** * Register a message handler. * @return {@code true} if the handler was subscribed or {@code false} if it * was already subscribed. */
        boolean subscribe(MessageHandler handler);
    
        /** * Un-register a message handler. * @return {@code true} if the handler was un-registered, or {@code false} * if was not registered. */
        boolean unsubscribe(MessageHandler handler);
    
    }

**2、常用消息通道**

(1)、PublishSubscribeChannel

PublishSubscribeChannel允许广播消息给所有订阅者，配置方式如下：

/** * 允许广播消息给所有订阅者，当前消息通道的id为publishSubscribeChannel * @return */
        @Bean
        public PublishSubscribeChannel publishSubscribeChannel(){
            PublishSubscribeChannel channel = new PublishSubscribeChannel();
            return channel;
        }

其中，当前消息通道的id为publishSubscribeChannel。

(2)、QueueChannel

QueueChannel允许消息接收者轮询获得消息，用一个队列(queue)接收消息，队列的容量大小可配置，配置方式如下：

@Bean
        public QueueChannel queueChannel(){
            QueueChannel channel = new QueueChannel(10);
            return channel;
        }

其中，QueueChannel构造参数10即为队列的容量。

(3)、PriorityChannel

PriorityChannel可按照优先级将数据存储到队列，它依据于消息的消息头priority属性，配置方式如下：

@Bean
        public PriorityChannel priorityChannel(){
            PriorityChannel channel = new PriorityChannel(10);
            return channel;
        }

(4)、RendezvousChannel

RendezvousChannel确保每一个接收者都接收到消息后再发送消息，配置方式如下：

@Bean
        public RendezvousChannel rendezvousChannel(){
            RendezvousChannel channel = new RendezvousChannel();
            return channel;
        }

(5) DirectChannel

DirectChannel是Spring Integration默认的消息通道，它允许将消息发送给为一个订阅者，然后阻碍发送直到消息被接收，配置方式如下：

@Bean
        public DirectChannel directChannel(){
            DirectChannel channel = new DirectChannel();
            return channel;
        }

(6)、ExecutorChannel

ExecutorChannel可绑定一个多线程的task executor，配置方式如下：

@Bean
        public ExecutorChannel executorChannel(){
            ExecutorChannel channel = new ExecutorChannel(executor());
            return channel;
        }
    
        @Bean
        public Executor executor(){
            ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
            taskExecutor.setCorePoolSize(5);
            taskExecutor.setMaxPoolSize(10);
            taskExecutor.setQueueCapacity(25);
            taskExecutor.initialize();
            return taskExecutor;
        }

**3、通道拦截器**

Spring Integration给消息通道提供了通道拦截器(ChannelInterceptor)，用来拦截发送和接收消息的操作.

ChannelInterceptor接口定义如下，我们只需要实现这个接口即可：

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(someInterceptor);

**Message EndPoint**

消息端点(Message EndPoint)是真正处理消息的(Message)组件，它还可以控制通道的路由。我们可用的消息端点包含如下：

(1) Channel Adapter

通道适配器(Channel Adapter)是一种连接外部系统或传输协议的端点(EndPoint)，可以分为入站(inbound)和出站(outbound)。  
通道适配器是单向的，入站通道适配器只支持接收消息，出站通道适配器只支持输出消息。

Spring Integration内置了如下的适配器：

RabbitMQ、Feed、File、FTP/SFTP、Gemfire、HTTP、TCP/UDP、JDBC、JPA、JMS、Mail、MongoDB、Redis、RMI
    Twitter、XMPP、WebServices(SOAP、REST)、WebSocket

(2) Gateway

消息网关(Gateway)类似于Adapter，但是提供了双向的请求/返回集成方式，也分为入站(inbound)和出站(outbound)。  
Spring Integration 对响应的Adapter都提供了Gateway。

(3) Service Activator

Service Activator 可调用Spring的Bean来处理消息，并将处理后的结果输出到指定的消息通道。