Spring Boot实战（九）9.4 系统集成Spring Integration

9.4.1 Spring Integration快速入门

Spring Integration提供了基于Spring的EIP（Enterprise Integration Patterns，企业集成模式的实现。Spring Integration主要解决的问题是不同系统之间交互的问题，通过异步消息驱动来达到系统交互时系统之间的松耦合。本节将基于无XML配置的原则使用Java配置、注解以及Spring Integration Java DSL来使用Spring Integration。

9.4.2 Message

Message是用来在不同部分之间传递的数据。Message由两部分组成：消息体（payload）与消息头（header）。消息体可以是任何数据类型（如XML、JSON，Java对象）；消息头表示的元数据就是解释消息体的内容。

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

9.4.3 Channel

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

1.*接口

（1）MessageChannel

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

public interface MessageChannel{
	public static final long INDEFINITE_TIMEOUT = -1;
	boolean send(Message<?>message);
	boolean send(Message<?>message,long timeout);
}

当使用send方法发送消息时，返回值为true，则表示消息发送成功。MessageChannel有两大子接口，分别为PollableChannel（可轮询）和SubscribableChannel（可订阅）。我们所有的消息通道类都是实现这两个接口。

（2）PollableChannel

PollableChannel具备轮询获得消息的能力，定义如下：

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

（3）SubscribableChannel

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

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

2.常用消息通道

（1）PublishSubScribeChannel

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

@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<?>messaeg,MessageChannel channel);
	void afterReceiveCompletion(Message<?>message,MessageChannel channel,Exception ex);
}

我们通过下面的代码给所有的channel增加拦截器：

channel.addInterceptor(someInterceptor);

9.4.4 Message EndPoint

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

（1）Channel Adapter

通道适配器（Channel Adapter）是一种连接外部系统或传输协议的端点（EndPoint），可以分入站（inbound）和出站（outbound）。
通道适配器是单向的，入站通道适配器只支持接收消息，出站通道适配器只支持输出消息。
Spring Integration内置如下的适配器：
RabbitMQ、Feed、File、FTPd/SFTP、Gemfire、HTTP、TCP/UDP、JDBC、JPA、JMS、Mail、MongoDB、Redis、RMI、Twitter、XMPP、WebServices（SOAP、RESTaurant）、WebSocket等。

（2）Gateway

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

（3）Service Activator

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

（4）Router

路由（Router）可根据消息体类型（Payload Type Router）、消息头的值（Header Value Router）以及定义好的接收表（Recipient List Router）作为条件，来决定消息传递到的通道。

（5）Filter

过滤器（Filter）类似于路由（Router），不同的是过滤器不决定消息路由到哪里，而是决定消息是否可以传递给消息通道。

（6）Splitter

拆分器（Splitter）将消息拆分为几个部分单独处理，拆分器处理的返回值是一个集合或者数组。

（7）Aggregator

聚合器（Aggregator）与拆分器相反，它接收一个java.util.List作为参数，将多个消息合并为一个消息。

（8）Enricher

当我们从外部获得消息后，需要增加额外的消息到已有的消息中，这时就需要使用消息增强器（Enricher）。消息增强器主要有消息体增强器（Payload Enricher）和消息头增强器（Header Enricher）两种。

（9）Transformer

转换器（Transformer）是地获得的消息进行一定的逻辑转换处理（如数据格式转换）。

（10）Bridge

使用连接桥（Bridge）可以简单地将两个消息通道连接起来。

9.4.5 Spring Integration Java DSL

Spring Integration提供了一个IntegrationFlow来定义系统继承流程，而通过IntegrationFlows和IntegrationFlowBuilder来实现使用Fluent API来定义流程。在Fulent API里，分别提供了下面方法来映射Spring Integration的端点（EndPoint）。

transform() -> Transformer
filter() -> Filter
handle() -> ServiceActivator、Adapter、Gateway
split() -> Splitter
aggregate() -> Aggregator
route() -> Router
bridge() -> Bridge

一个简单的流程定义如下：

@Bean
public IntegrationFlow demoFlow(){
 return IntegrationFlows.from("input")  //从Channel  input获取消息
 	.<String,Integer>transform(Integer::parseint) //将消息转换成整数
 	.get();  //获得集成流程并注册为Bean
}

9.4.6 实战

本章将演示读取https://spring.io/blog.atom的新闻聚合文件，atom是一种xml文件，且格式是固定的，示例如下：

<?xml version="1.0" encoding="UTF-8"?>
<feed xmlns="http://www.w3.org/2005/Atom">
	<title>Spring</title>
	<link rel="alternate" href="https://spring.io/blog" />
	<link rel="self" href="https://spring.io/blog.atom" />
	<id>http://spring.io/blog.atom</id>
	<icon>https://spring.io/favicon.ico</icon>
	<upadated>2015-07-29T14:46:00z</upadated>
	<entry>
		<title>Spring Cloud Connectors 1.2.0 released</title>
		<link rel="alternate" href="http://..." />
		<category term="releases" label="Releases" />
		<author>
			<name>some author</name>
		</author>
		<id>tag:spring.io,2015-07-27:2196</id>
		<updated>2015-07-29T14:46:00z</updated>
		<content type="html">...</content>
	</entry>
</feed>

我们将读取到消息通过分类（Category），将消息转到不同的消息通道，将分类为releases和engineering的消息写入磁盘文件，将分类为news的消息通过邮件发送。

1.新建Spring Boot项目

新建Spring Boot项目，依赖为Integration（spring-boot-starter-integration）和mail（spring-boot-starter-mail）。
项目信息

groupId:com.wisely
arctifactId:ch9_4
package:com.wisely.ch9_4

另外，我们还要添加Spring Integration对atom及mail的支持。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>1.5.20</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>
	<groupId>com.wisely</groupId>
	<artifactId>ch9_4</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<name>ch9_3_5-1</name>
	<description>Demo project for Spring Boot</description>

	<properties>
		<java.version>1.8</java.version>
	</properties>

	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-integration</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-mail</artifactId>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		
		<dependency>
			<groupId>org.springframework.integration</groupId>
			<artifactId>spring-integration-feed</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.integration</groupId>
			<artifactId>spring-integration-mail</artifactId>
		</dependency>
		
		<dependency>
		    <groupId>org.springframework.integration</groupId>
		    <artifactId>spring-integration-java-dsl</artifactId>
		    <version>1.2.3.RELEASE</version>
		</dependency>
		
		<!-- https://mvnrepository.com/artifact/org.springframework.integration/spring-integration-file -->
		<dependency>
		    <groupId>org.springframework.integration</groupId>
		    <artifactId>spring-integration-file</artifactId>
		</dependency>
				
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>

</project>

本例所有的代码都在入口类中完成。

2.读取流程

package com.wisely.ch9_4;

import java.io.IOException;


import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.Resource;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.feed.inbound.FeedEntryMessageSource;
import org.springframework.integration.scheduling.PollerMetadata;

import com.rometools.rome.feed.synd.SyndEntry;

@SpringBootApplication
public class Ch9351Application {
	
	@Value("https://spring.io/blog.atom")  //通过@Value注解自动获得https://spring.id/blog.atom的资源
	Resource resource;
	
	@Bean(name = PollerMetadata.DEFAULT_POLLER)
	public PollerMetadata poller() {  //使用Fluent API和Pollers配置默认的轮询方式
		return Pollers.fixedRate(500).get();
	}
	
	@Bean
	public FeedEntryMessageSource feedEntryMessageSource()throws IOException{ //FeedEntryMessageSource实际为feed:inbound-channel-adapter,此处即构造feed的入站通道适配器作为数据输入
		FeedEntryMessageSource messageSource = new FeedEntryMessageSource(resource.getURL(), "news");
		return messageSource;
	}
	
	@Bean
	public IntegrationFlow myFlow() throws IOException {
		return IntegrationFlows.from(feedEntryMessageSource())  //流程从from方法开始
				.<SyndEntry,String>route(payload ->
				payload.getCategories().get(0).getName(),  //通过路由方法route来选择路由，消息体（payload）的类型为SyndEntry，作为判断条件的类型为String，判断的值是通过payload获得的分类（Categroy）;
				mapping -> mapping.channelMapping("releases",
						"releasesChannel")  //通过不同分类的值转向不同的消息通道，若分类为releases，则转向releasesChannel;
				.channelMapping("engineering",
						"engineeringChannel")
				.channelMapping("news","newsChannel"))
				.get();   //通过get方法获得IntegrationFlow实体，配置为Spring的Bean。
	}

	public static void main(String[] args) {
		SpringApplication.run(Ch9351Application.class, args);
	}

}

3.releases流程

	@Bean
	public IntegrationFlow releasesFlow() {
		return IntegrationFlows.from(MessageChannels.queue("releasesChannel", 10)) //从消息通道releasesChannel开始获取数据。
				.<SyndEntry, String> transform(
						payload -> "《" + payload.getTitle() + "》 " + payload.getLink() + getProperty("line.separator")) //使用transform方法进行数据转换。payload类型为SyndEntry，将其转换为字符串类型，并自定义数据的格式。
				.handle(Files.outboundAdapter(new File("e:/springblog")) //用handle方法处理file的出站适配器。Files类是由Spring Integration Java DSL提供的 Fluent API用来构造文件输出的适配器。
						.fileExistsMode(FileExistsMode.APPEND) 
						.charset("UTF-8") 
						.fileNameGenerator(message -> "releases.txt") 
						.get())
				.get();
	}

4.engineering流程

	@Bean
	public IntegrationFlow engineeringFlow() {
		return IntegrationFlows.from(MessageChannels.queue("engineeringChannel", 10))
				.<SyndEntry, String> transform(
						payload -> "《" + payload.getTitle() + "》" + payload.getLink() + getProperty("line.separator"))
				.handle(Files.outboundAdapter(new File("e:/springblog"))
						.fileExistsMode(FileExistsMode.APPEND)
						.charset("UTF-8")
						.fileNameGenerator(message -> "engineering.txt")
						.get())
				.get();
	}
	}

5.news流程

	@Bean
	public IntegrationFlow newsFlow() {
		return
				IntegrationFlows.from(MessageChannels.queue("newsChannel",10))
				.<SyndEntry,String>transform(
						payload -> "《" + payload.getTitle() + "》" +
				payload.getLink() + getProperty("line.separator"))
				.enrichHeaders(                   //通过enricherHeader来增加消息头的信息
						Mail.headers()
						.subject("来自Spring的新闻")
						.to("[email protected]")
						.from("[email protected]"))
				.handle(Mail.outboundAdapter("smtp.qq.com")  //邮件发送的相关信息通过Spring Integration JavaDSL提供的Mail的headers方法来构造。
						.port(25)
						.protocol("smtp")
						.credentials("[email protected]", "******") //使用handle方法来定义邮件发送的出站适配器，使用Spring Integration Java DSL提供的Mail。来构造，这里使用[email protected]给自己发送邮件
						.javaMailProperties(p -> p.put("mail.debug", false)),
						e -> e.id("smtpOut"))
				.get();
				
	}

6.运行

（1）写文件结果

查看E:\springblog目录，发现多了两个文件，如图

engineering.txt文件内容如图

releases.txt文件内容如图

（2）邮箱接收结果
登录邮箱可以看取刚才发送的邮件