-
什么是 Spring Integration
- 什么是 SCS 体系及其原理
本文配套可交互教程已登录阿里云知行动手实验室，PC 端登录 [start.aliyun.com](
)_ _在浏览器中立即体验。
[](
)Spring Messaging
=====================================================================================
Spring Messaging 是 Spring Framework 中的一个模块，其作用就是统一消息的编程模型。
- 比如消息 Messaging 对应的模型就包括一个消息体 Payload 和消息头 Header：
package org.springframework.messaging;
public interface Message<T> {
T getPayload();
MessageHeaders getHeaders();
}
- 消息通道 MessageChannel 用于接收消息，调用send方法可以将消息发送至该消息通道中：
@FunctionalInterface
public interface MessageChannel {
long INDEFINITE_TIMEOUT = -1;
default boolean send(Message<?> message) {
return send(message, INDEFINITE_TIMEOUT);
}
boolean send(Message<?> message, long timeout);
}
[](
)消息通道里的消息如何被消费呢？
- 由消息通道的子接口可订阅的消息通道SubscribableChannel实现，被MessageHandler消息处理器所订阅：
public interface SubscribableChannel extends MessageChannel {
boolean subscribe(MessageHandler handler);
boolean unsubscribe(MessageHandler handler);
}
- 由MessageHandler真正地消费/处理消息：
@FunctionalInterface
public interface MessageHandler {
void handleMessage(Message<?> message) throws MessagingException;
}
[](
)Spring Messaging 内部在消息模型的基础上衍生出了其它的一些功能，如：
-
消息接收参数及返回值处理：消息接收参数处理器HandlerMethodArgumentResolver配合@Header, @Payload等注解使用；消息接收后的返回值处理器HandlerMethodReturnValueHandler配合@SendTo注解使用；
-
消息体内容转换器MessageConverter；
-
统一抽象的消息发送模板AbstractMessageSendingTemplate；
- 消息通道拦截器ChannelInterceptor；
[](
)Spring Integration
=======================================================================================
Spring Integration 提供了 Spring 编程模型的扩展用来支持企业集成模式(Enterprise Integration Patterns)，是对 Spring Messaging 的扩展。
它提出了不少新的概念，包括消息路由MessageRoute、消息分发MessageDispatcher、消息过滤Filter、消息转换Transformer、消息聚合Aggregator、消息分割Splitter等等。同时还提供了MessageChannel和MessageHandler的实现，分别包括 DirectChannel、ExecutorChannel、PublishSubscribeChannel和MessageFilter、ServiceActivatingHandler、MethodInvokingSplitter 等内容。
[](
)这里为大家介绍几种消息的处理方式：
- 消息的分割：
- 消息的聚合：
- 消息的过滤：
- 消息的分发：
[](
)接下来，我们以一个最简单的例子来尝试一下 Spring Integration。
这段代码解释为：
SubscribableChannel messageChannel =new DirectChannel(); // 1
messageChannel.subscribe(msg-> { // 2
System.out.println("receive: " +msg.getPayload());
});
messageChannel.send(MessageBuilder.withPayload("msgfrom alibaba").build()); // 3
-
构造一个可订阅的消息通道messageChannel。
-
使用MessageHandler去消费这个消息通道里的消息。
-
发送一条消息到这个消息通道，消息最终被消息通道里的MessageHandler所消费。
- 最后控制台打印出:receive: msg from alibaba。
DirectChannel内部有个UnicastingDispatcher类型的消息分发器，会分发到对应的消息通道MessageChannel中，从名字也可以看出来，UnicastingDispatcher是个单播的分发器，只能选择一个消息通道。那么如何选择呢? 内部提供了LoadBalancingStrategy负载均衡策略，默认只有轮询的实现，可以进行扩展。
我们对上段代码做一点修改，使用多个 MessageHandler 去处理消息：
SubscribableChannel messageChannel = new DirectChannel();
messageChannel.subscribe(msg -> {
System.out.println("receive1: " + msg.getPayload());
});
messageChannel.subscribe(msg -> {
System.out.println("receive2: " + msg.getPayload());
});
messageChannel.send(MessageBuilder.withPayload("msg from alibaba").build());
messageChannel.send(MessageBuilder.withPayload("msg from alibaba").build());
由于DirectChannel内部的消息分发器是UnicastingDispatcher单播的方式，并且采用轮询的负载均衡策略，所以这里两次的消费分别对应这两个MessageHandler。控制台打印出：
receive1: msg from alibaba
receive2: msg from alibaba
既然存在单播的消息分发器UnicastingDispatcher，必然也会存在广播的消息分发器，那就是BroadcastingDispatcher，它被 PublishSubscribeChannel 这个消息通道所使用。广播消息分发器会把消息分发给所有的 MessageHandler：
SubscribableChannel messageChannel = new PublishSubscribeChannel();
messageChannel.subscribe(msg -> {
System.out.println("receive1: " + msg.getPayload());
});
messageChannel.subscribe(msg -> {
System.out.println("receive2: " + msg.getPayload());
});
messageChannel.send(MessageBuilder.withPayload("msg from alibaba").build());
messageChannel.send(MessageBuilder.withPayload("msg from alibaba").build());
[](
)Spring Cloud Stream
========================================================================================
[](
)SCS 与各模块之间的关系是：
最后
CodeChina开源项目：【一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频】
由于篇幅原因，就不多做展示了
原创文章，作者：3628473679，如若转载，请注明出处：https://blog.ytso.com/149969.html