简介: Spring Cloud Stream在 Spring Cloud 体系内用于构建高度可扩展的基于事件驱动的微服务,其目的是为了简化消息在 Spring Cloud 应用程序中的开发。
Spring Cloud Stream在 Spring Cloud 体系内用于构建高度可扩展的基于事件驱动的微服务,其目的是为了简化消息在 Spring Cloud 应用程序中的开发。
Spring Cloud Stream (后面以 SCS 代替 Spring Cloud Stream) 本身内容很多,而且它还有很多外部的依赖,想要熟悉 SCS,必须要先了解 Spring Messaging 和 Spring Integration 这两个项目,接下来,文章将围绕以下三点进行展开:
本文配套可交互教程已登录阿里云知行动手实验室,PC 端登录 start.aliyun.com_ _在浏览器中立即体验。
Spring Messaging 是 Spring Framework 中的一个模块,其作用就是统一消息的编程模型。
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);
}
消息通道里的消息如何被消费呢?
public interface SubscribableChannel extends MessageChannel {
boolean subscribe(MessageHandler handler);
boolean unsubscribe(MessageHandler handler);
}
@FunctionalInterface
public interface MessageHandler {
void handleMessage(Message<?> message) throws MessagingException;
}
Spring Messaging 内部在消息模型的基础上衍生出了其它的一些功能,如:
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
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());
SCS 与各模块之间的关系是:
从图中可以看出,Binding是连接应用程序跟消息中间件的桥梁,用于消息的消费和生产。我们来看一个最简单的使用 RocketMQ Binder 的例子,然后分析一下它的底层处理原理:
启动类及消息的发送:
@SpringBootApplication
@EnableBinding({ Source.class, Sink.class }) // 1
public class SendAndReceiveApplication {
public static void main(String[] args) {
SpringApplication.run(SendAndReceiveApplication.class, args);
}
@Bean // 2
public CustomRunner customRunner() {
return new CustomRunner();
}
public static class CustomRunner implements CommandLineRunner {
@Autowired
private Source source;
@Override
public void run(String... args) throws Exception {
int count = 5;
for (int index = 1; index <= count; index++) {
source.output().send(MessageBuilder.withPayload("msg-" + index).build()); // 3
}
}
}
}
@Service
public class StreamListenerReceiveService {
@StreamListener(Sink.INPUT) // 4
public void receiveByStreamListener1(String receiveMsg) {
System.out.println("receiveByStreamListener: " + receiveMsg);
}
}
这段代码很简单,没有涉及到 RocketMQ 相关的代码,消息的发送和接收都是基于 SCS 体系完成的。如果想切换成 RabbitMQ 或 Kafka,只需修改配置文件即可,代码无需修改。
我们来分析下这段代码的原理:
1.@EnableBinding对应的两个接口属性Source和Sink是 SCS 内部提供的。SCS 内部会基于Source和Sink构造BindableProxyFactory,且对应的 output 和 input 方法返回的 MessageChannel 是DirectChannel。output 和 input 方法修饰的注解对应的 value 是配置文件中 binding 的 name。
public interface Source {
String OUTPUT = "output";
@Output(Source.OUTPUT)
MessageChannel output();
}
public interface Sink {
String INPUT = "input";
@Input(Sink.INPUT)
SubscribableChannel input();
}
配置文件里 bindings 的 name 为 output 和 input,对应Source和Sink接口的方法上的注解里的 value:
spring.cloud.stream.bindings.output.destination=test-topic
spring.cloud.stream.bindings.output.content-type=text/plain
spring.cloud.stream.rocketmq.bindings.output.producer.group=demo-group
spring.cloud.stream.bindings.input.destination=test-topic
spring.cloud.stream.bindings.input.content-type=text/plain
spring.cloud.stream.bindings.input.group=test-group1
这个过程文字描述有点啰嗦,用一张图总结一下(黄色部分涉及到各消息中间件的 Binder 实现以及 MQ 基本的订阅发布功能):
SCS 章节的最后,我们来看一段 SCS 关于消息的处理方式的一段代码:
@StreamListener(value = Sink.INPUT, condition = "headers['index']=='1'")
public void receiveByHeader(Message msg) {
System.out.println("receive by headers['index']=='1': " + msg);
}
@StreamListener(value = Sink.INPUT, condition = "headers['index']=='9999'")
public void receivePerson(@Payload Person person) {
System.out.println("receive Person: " + person);
}
@StreamListener(value = Sink.INPUT)
public void receiveAllMsg(String msg) {
System.out.println("receive allMsg by StreamListener. content: " + msg);
}
@StreamListener(value = Sink.INPUT)
public void receiveHeaderAndMsg(@Header("index") String index, Message msg) {
System.out.println("receive by HeaderAndMsg by StreamListener. content: " + msg);
}
有没有发现这段代码跟 Spring MVC Controller 中接收请求的代码很像? 实际上他们的架构都是类似的,Spring MVC 对于 Controller 中参数和返回值的处理类分别是org.springframework.web.method.support.HandlerMethodArgumentResolver、org.springframework.web.method.support.HandlerMethodReturnValueHandler。
Spring Messaging 中对于参数和返回值的处理类之前也提到过,分别是org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver、org.springframework.messaging.handler.invocation.HandlerMethodReturnValueHandler。
它们的类名一模一样,甚至内部的方法名也一样。
上图是 SCS 体系相关类说明的总结,关于 SCS 以及 RocketMQ Binder 更多相关的示例,可以参考 RocketMQ Binder Demos,包含了消息的聚合、分割、过滤;消息异常处理;消息标签、SQL 过滤;同步、异步消费等等。
原文链接
本文为阿里云原创内容,未经允许不得转载。
用css3将input框写出类似Google登录页面的动画效果 效果一 代码如下 CSS body{ b...
"Code tailor",为前端开发者提供技术相关资讯以及系列基础文章,微信关注“小和...
定义 H5为我们提供了以 data- 为前缀定义需要的属性即可设置自定义属性。 div id...
今天在制作网页特效的过程中遇到了这样的问题,浏览器中预览首选参数中指定的浏...
实现功能: 将桌面图片拖入指定地方,生成相框和相关信息。 相框需要自己配置,...
通用类 overflow:hidden;自动隐藏超出的内容,防止撑开层和表格的范围 !importan...
1.HTML概述 1.HTML:超文本标记语言。是一种标识性语言,非编程语言,不能使用逻...
网站维修,想要做一个提示并自动跳转的网页,该怎么做呢?下面我们就来看看dw 8...
我们在做网页的时候,为了更好,会添加一下自己喜欢的音乐,今天给大家带来这样...
代码示例 在head标签中添加一行代码: XML/HTML Code 复制内容到剪贴板 html hea...