- Spring Integration主要解决的问题是不同系统之间交互的问题，通过异步消息驱动来达到系统交互时系统之间的松耦合。
- Spring Integration主要由Message，Channel和Message EndPoint组成。
- Message：用来在不同部分之间传递的数据。由消息体（payload）和消息头（header）。消息体可以是任何数据类型（xml，json，Java对象）；消息头的元数据就是解释消息体的内容。
- Channel：消息发送者发送消息到通道，消息接收者从通道接收消息。
- Message EndPoint：是真正处理xiao消息的组件，还可以控制通道的路由。可用的消息端点有：
（1）Channel Adapter：是一种连接外部或传输协议的端点，分为入站（inbound）和出站（outbound），通道适配器是单向的，入站通道适配器只支持接收消息，出站通道适配器只支持输出消息。
6. Spring Integration Java DSL ：Spring Integration 提供IntegrationFlow来定义系统继承流程，
通过IntegrationFlows和IntegrationFlowBuilder来实现使用Fluent API来定义流程。
一个简单的流程定义如下：
@Bean public IntegrationFlow demoFlow() { return IntegrationFlows.from("input")//从Channel input获取消息 .<SyndEntry, String> transform(Integer::parseInt)//将消息转换成整数 .get();//获取集成流程并注册为Bean }
完整示例：
package com.integration;
import com.rometools.rome.feed.synd.SyndEntry; 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.channel.MessageChannels; import org.springframework.integration.dsl.core.Pollers; import org.springframework.integration.dsl.file.Files; import org.springframework.integration.dsl.mail.Mail; import org.springframework.integration.feed.inbound.FeedEntryMessageSource; import org.springframework.integration.file.support.FileExistsMode; import org.springframework.integration.scheduling.PollerMetadata;
import java.io.File; import java.io.IOException;
import static java.lang.System.getProperty;
@SpringBootApplication public class IntegrationApplication {
//自动获得资源 @Value("https://spring.io/blog.atom") Resource resource;
public static void main(String[] args) { SpringApplication.run(IntegrationApplication.class, args); }
//配置默认的轮询方式 @Bean(name = PollerMetadata.DEFAULT_POLLER) public PollerMetadata poller() { return Pollers.fixedRate(500).get(); }
@Bean public FeedEntryMessageSource feedMessageSource() throws IOException { FeedEntryMessageSource messageSource = new FeedEntryMessageSource(resource.getURL(), "news"); return messageSource; }
@Bean public IntegrationFlow myFlow() throws IOException { return IntegrationFlows.from(feedMessageSource()) //通过route来选择路由，消息体（payload）的类型为SyndEntry,判断条件类型为String //判断的值通过payload获得的分类（Categroy） //不同的分类值转向不同的消息通道 .<SyndEntry, String>route(payload -> payload.getCategories().get(0).getName(), mapping -> mapping.channelMapping("releases", "releasesChannel") .channelMapping("engineering", "engineeringChannel") .channelMapping("news", "newsChannel")) .get();//获得IntegrationFlow实体，配置为Spring的Bean }
//releases流程 @Bean public IntegrationFlow releasesFlow() { return IntegrationFlows.from(MessageChannels.queue("releasesChannel", 10)) //从releasesChannel获取数据 .<SyndEntry, String>transform( //transform方法做数据转换，payload类型为SynEntry，将其转换为字符串类型， // 并自定义数据的格式 payload -> "《" + payload.getTitle() + "》" + payload.getLink() + getProperty("line.separator")) //handle方法处理file的出站适配器，Files类是由Spring Integration Java DSL提供的 // Fluent API 用来构造文件输出的适配器 .handle(Files.outboundAdapter(new File("g:/springblog")) .fileExistsMode(FileExistsMode.APPEND) .charset("UTF-8") .fileNameGenerator(message -> "releases.txt") .get()) .get(); }
//engineering流程，与releases流程相同 @Bean public IntegrationFlow engineeringFlow() { return IntegrationFlows.from(MessageChannels.queue("engineeringChannel", 10)) .<SyndEntry, String> transform( e -> "《" + e.getTitle() + "》" + e.getLink() + getProperty("line.separator")) .handle(Files.outboundAdapter(new File("g:/springblog")) .fileExistsMode(FileExistsMode.APPEND) .charset("UTF-8") .fileNameGenerator(message -> "engineering.txt") .get()) .get(); }
//news流程 @Bean public IntegrationFlow newsFlow() { return IntegrationFlows.from(MessageChannels.queue("newsChannel", 10)) .<SyndEntry, String> transform( payload -> "《" + payload.getTitle() + "》" + payload.getLink() + getProperty("line.separator")) //提供enrichHeaders方法增加消息头的信息 .enrichHeaders( Mail.headers() .subject("来自Spring的新闻") .to("guang2_tan@126.com") .from("guang2_tan@126.com")) //邮件发送的相关信息通过Spring Integration Java DSL提供的Mail的headers方法来构造 //使用handle方法自定义邮件发送的出站适配器，使用Spring Integration Java DSL提供的Mail.outboundAdapter来构造 //使用guang2_tan@126.com向自己发送邮件 .handle(Mail.outboundAdapter("smtp.126.com") .port(25) .protocol("smtp") .credentials("guang2_tan@126.com", "abc1234") .javaMailProperties(p -> p.put("mail.debug", "false")), e -> e.id("smtpOut")) .get(); } }
可能会出现：javax.mail.AuthenticationFailedException: 550
解决办法：可能邮箱没有开通pop/stmp协议