Spring Cloud Stream 体系及原理介绍

简介: Spring Cloud Stream在 Spring Cloud 体系内用于构建高度可扩展的基于事件驱动的微服务,其目的是为了简化消息在 Spring Cloud 应用程序中的开发。

头图.png

作者 | 洛夜
来源 | 阿里巴巴云原生公众号

Spring Cloud Stream在 Spring Cloud 体系内用于构建高度可扩展的基于事件驱动的微服务,其目的是为了简化消息在 Spring Cloud 应用程序中的开发。

Spring Cloud Stream (后面以 SCS 代替 Spring Cloud Stream) 本身内容很多,而且它还有很多外部的依赖,想要熟悉 SCS,必须要先了解 Spring Messaging 和 Spring Integration 这两个项目,接下来,文章将围绕以下三点进行展开:

  • 什么是 Spring Messaging
  • 什么是 Spring Integration
  • 什么是 SCS 体系及其原理

1.png

本文配套可交互教程已登录阿里云知行动手实验室,PC 端登录 start.aliyun.com_ _在浏览器中立即体验。

Spring Messaging

Spring Messaging 是 Spring Framework 中的一个模块,其作用就是统一消息的编程模型。

  • 比如消息 Messaging 对应的模型就包括一个消息体 Payload 和消息头 Header:

2.png

package org.springframework.messaging;
public interface Message<T> {T getPayload();MessageHeaders getHeaders();
}
  • 消息通道 MessageChannel 用于接收消息,调用send方法可以将消息发送至该消息通道中:

3.png

@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 等内容。

这里为大家介绍几种消息的处理方式:

  • 消息的分割:

4.png

  • 消息的聚合:

5.png

  • 消息的过滤:

6.png

  • 消息的分发:

7.png

接下来,我们以一个最简单的例子来尝试一下 Spring Integration。

这段代码解释为:

SubscribableChannel messageChannel =new DirectChannel(); // 1messageChannel.subscribe(msg-> { // 2System.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 与各模块之间的关系是:

  • SCS 在 Spring Integration 的基础上进行了封装,提出了Binder, Binding, @EnableBinding, @StreamListener等概念。
  • SCS 与 Spring Boot Actuator 整合,提供了/bindings, /channelsendpoint。
  • SCS 与 Spring Boot Externalized Configuration 整合,提供了BindingProperties, BinderProperties等外部化配置类。
  • SCS 增强了消息发送失败的和消费失败情况下的处理逻辑等功能。
  • SCS 是 Spring Integration 的加强,同时与 Spring Boot 体系进行了融合,也是 Spring Cloud Bus 的基础。它屏蔽了底层消息中间件的实现细节,希望以统一的一套 API 来进行消息的发送/消费,底层消息中间件的实现细节由各消息中间件的 Binder 完成。

Binder是提供与外部消息中间件集成的组件,为构造Binding提供了 2 个方法,分别是bindConsumer和bindProducer,它们分别用于构造生产者和消费者。目前官方的实现有 Rabbit Binder 和 Kafka Binder, Spring Cloud Alibaba 内部已经实现了 RocketMQ Binder。

8.png

从图中可以看出,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 // 2public CustomRunner customRunner() {return new CustomRunner();}public static class CustomRunner implements CommandLineRunner {@Autowiredprivate Source source;@Overridepublic 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) // 4public 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-groupspring.cloud.stream.bindings.input.destination=test-topic
spring.cloud.stream.bindings.input.content-type=text/plain
spring.cloud.stream.bindings.input.group=test-group1
  1. 构造CommandLineRunner,程序启动的时候会执行CustomRunner的run方法。
  2. 调用Source接口里的 output 方法获取DirectChannel,并发送消息到这个消息通道中。这里跟之前 Spring Integration 章节里的代码一致。
  • Source 里的 output 发送消息到DirectChannel消息通道之后会被AbstractMessageChannelBinder#SendingHandler这个MessageHandler处理,然后它会委托给AbstractMessageChannelBinder#createProducerMessageHandler创建的 MessageHandler 处理(该方法由不同的消息中间件实现)。
  • 不同的消息中间件对应的AbstractMessageChannelBinder#createProducerMessageHandler方法返回的 MessageHandler 内部会把 Spring Message 转换成对应中间件的 Message 模型并发送到对应中间件的 broker。
  1. 使用@StreamListener进行消息的订阅。请注意,注解里的Sink.input对应的值是 "input",会根据配置文件里 binding 对应的 name 为 input 的值进行配置:
  • 不同的消息中间件对应的AbstractMessageChannelBinder#createConsumerEndpoint方法会使用 Consumer 订阅消息,订阅到消息后内部会把中间件对应的 Message 模型转换成 Spring Message。
  • 消息转换之后会把 Spring Message 发送至 name 为 input 的消息通道中。
  • @StreamListener对应的StreamListenerMessageHandler订阅了 name 为 input 的消息通道,进行了消息的消费。

这个过程文字描述有点啰嗦,用一张图总结一下(黄色部分涉及到各消息中间件的 Binder 实现以及 MQ 基本的订阅发布功能):

9.png

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。

它们的类名一模一样,甚至内部的方法名也一样。

总结

10.png

上图是 SCS 体系相关类说明的总结,关于 SCS 以及 RocketMQ Binder 更多相关的示例,可以参考 RocketMQ Binder Demos,包含了消息的聚合、分割、过滤;消息异常处理;消息标签、SQL 过滤;同步、异步消费等等。

原文链接

本文为阿里云原创内容,未经允许不得转载。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/513655.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

python半圆_如何使用Python中的Zelle图形生成一个半圆?

Zelle图形模块不提供直接绘制半圆(圆弧)的代码。但是&#xff0c;由于模块是用Python编写的&#xff0c;构建在tkinter之上&#xff0c;并且tkinter提供了一个圆弧绘制例程&#xff0c;因此我们可以添加自己的arc子类&#xff0c;该类继承了Zelle Oval类并实现了圆弧&#xff1…

pythonpyquery去掉br_如何在将pyquery对象转换为字符串时取消特殊字符的转义

我试图用python请求模块获取一个远程页面&#xff0c;重建一个DOM树&#xff0c;做一些处理并将结果保存到文件中。当我获取一个页面&#xff0c;然后将其写入文件时&#xff0c;一切都正常(我可以稍后在浏览器中打开一个html文件&#xff0c;它会正确呈现)。在但是&#xff0c…

工商银行分布式服务C10K场景的解决方案

简介&#xff1a; 未来&#xff0c;中国工商银行将持续致力于 Dubbo 的金融级规模化应用。 作者&#xff1a;颜高飞&#xff0c;微服务领域架构师&#xff0c;主要从事服务发现、高性能网络通信等研发工作&#xff0c;擅长 ZooKeeper、Dubbo、RPC 协议等技术方向。 Dubbo是一款…

蚂蚁链发布新一代网络平台「FAIR」 区块链进入隐私计算原生时代

10月22日&#xff0c;2021杭州云栖大会上&#xff0c;蚂蚁集团旗下蚂蚁链宣布升级区块链架构&#xff0c;在新架构上&#xff0c;隐私计算成为一种原生能力。蚂蚁链还基于这一新架构推出全新区块链网络平台「FAIR」&#xff0c;在「FAIR」上&#xff0c;隐私计算将在出厂设置中…

使用html() undefined_SweetAlert2使用教程

SweetAlert2是一款功能强大的纯Js模态消息对话框插件。SweetAlert2用于替代浏览器默认的弹出对话框&#xff0c;它提供各种参数和方法&#xff0c;支持嵌入图片&#xff0c;背景&#xff0c;HTML标签等&#xff0c;并提供5种内置的情景类&#xff0c;功能非常强大。SweetAlert2…

埃森哲携手阿里云,采用K8s容器云服务为客户提供无限弹性

简介&#xff1a; 埃森哲作为全球领先的专业服务公司&#xff0c;在数字化、云计算等领域拥有全球领先的能力&#xff0c;我们在多年的实际客户项目中&#xff0c;找到并沉淀出了适合企业数字化转型的方法论&#xff0c;积累了丰富的落地经验。 作者&#xff1a;姚迪、周警伟 …

4阶范德蒙德行列式例题_线性代数入门——“爪型行列式”的计算及其应用

系列简介&#xff1a;这个系列文章讲解线性代数的基础内容&#xff0c;注重学习方法的培养。线性代数课程的一个重要特点(也是难点)是概念众多&#xff0c;而且各概念间有着千丝万缕的联系&#xff0c;对于初学者不易理解的问题我们会不惜笔墨加以解释。在内容上&#xff0c;以…

如何使用Arthas提高日常开发效率?

简介&#xff1a; 1. Arthas有什么功能&#xff0c;怎么用&#xff0c;请看&#xff1a;Arthas使用手册 2. Arthas命令比较复杂&#xff0c;一个帮助生成命令的IDEA插件&#xff1a;arthas idea plugin 使用文档 3. 基于Arthas实现的简单好用的热部署插件&#xff1a;ArthasHot…

stringutils 用哪个包 apache spring_spring整合mq、jsonp跨越、httpclient工具的使用

训练大纲(第087天)大家如果想快速有效的学习&#xff0c;思想核心是“以建立知识体系为核心”&#xff0c;具体方法是“守破离”。确保老师课堂上做的操作&#xff0c;反复练习直到熟练。第173次(ActiveMQ)学习主题&#xff1a;ActiveMQ学习目标&#xff1a;1 掌握什么是spring…

阿里推出会议AI助理“听悟”,面向未来会议

10月21日&#xff0c;杭州云栖大会上&#xff0c;阿里巴巴达摩院发布软硬件结合语音产品“听悟”。相比只能逐字转写的上一代语音产品&#xff0c;“听悟”是面向未来会议的AI助理&#xff0c;集成了达摩院最新语音及自然语言处理技术&#xff0c;不仅“听”得准&#xff0c;且…

几种Java常用序列化框架的选型与对比

简介&#xff1a; 序列化与反序列化是我们日常数据持久化和网络传输中经常使用的技术&#xff0c;但是目前各种序列化框架让人眼花缭乱&#xff0c;不清楚什么场景到底采用哪种序列化框架。本文会将业界开源的序列化框架进行对比测试&#xff0c;分别从通用性、易用性、可扩展性…

12v小型电机型号大全_电动机型号参数大全,再也不怕看不懂电机型号了

电动机型号是便于使用、设计、制造等部门进行业务联系和简化技术文件中产品名称、规格、型式等叙述而引用的一种代号。下面为大家介绍电动机型号含义等信息。1电动机型号组成及含义由电机类型代号、电机特点代号、设计序号和励磁方式代号等四个小节顺序组成。1、类型代号是表征…

基于DataWorks搭建新零售数据中台

文章作者&#xff1a;许日&#xff08;欢伯&#xff09;&#xff0c;在2016年盒马早期的时候&#xff0c;转到盒马事业部作为在线数据平台的研发负责人&#xff0c;现任阿里云计算平台DataWorks建模引擎团队负责人。 文章简介&#xff1a;本篇文章向大家分享新零售企业如何基于…

身份云平台 Authing 完成 2300 万美元 A 轮融资

10 月 24 日&#xff0c;身份云平台 Authing 宣布完成 2300 万美元 A 轮融资。本轮融资由老虎环球基金领投&#xff0c;鼎晖VGC&#xff08;创新与成长基金&#xff09;、声网 Agora、老股东 GGV纪源资本和奇绩创坛跟投&#xff0c;跃为资本担任独家财务顾问。Authing 表示&…

python实现接口自动化_python 实现接口自动化1

#coding"utf-8"importxlrdfrom xlutils.copy importcopyimportrequestsfrom util.log importloggerfrom util.db_util importDBUtil#xlrd&#xff1a;读取Excel文件数据#xlwt&#xff1a;写入Excel 数据&#xff0c;缺点是无法复用&#xff0c;写入会全部覆盖&#x…

大数据计算存储资源池_管家实践:轻松玩转大数据计算服务

以下是直播内容精华整理&#xff0c;主要包括以下四个方面&#xff1a;1.背景速览&#xff1b;2.功能介绍&#xff1b;3.案例讲解&#xff1b;4.新功能预告。一、背景速览MaxCompute(原ODPS)是一项大数据计算服务&#xff0c;它能提供快速、完全托管的PB级数据仓库解决方案&…

客如云数据中台建设

简介&#xff1a; 本次分享介绍客如云如何利用阿里云大数据产品来建设数据中台。 客如云是2012年成立的一家公司&#xff0c;覆盖餐饮、零售、美业&#xff0c;还有其他的业态以及服务的一家综合性的SaaS公司。到2020年为止&#xff0c;客如云已经服务了60万商家&#xff0c;帮…

微博机器学习平台云上最佳实践

简介&#xff1a; 本文讲述了微博机器学习平台和深度学习平台的业务功能和云上实践&#xff0c;剖析了阿里云大数据在微博这两大学习平台的架构建设上所起到的作用。 作者&#xff1a;新浪微博数据计算平台系统架构师 曹富强 本文讲述了微博机器学习平台和深度学习平台的业务功…

搞懂异地多活,看这篇就够了

来源&#xff1a;水滴与银弹作者&#xff1a;Kaito阅读本文大约需要 20 分钟。你好&#xff0c;我是 Kaito。在软件开发领域&#xff0c;「异地多活」是分布式系统架构设计的一座高峰&#xff0c;很多人经常听过它&#xff0c;但很少人理解其中的原理。异地多活到底是什么&…

python项目部署访问特别卡_【python新人求助】flask+pymssql 通过wsgi发布到Apache,访问接口服务器卡死?...

前端发起http请求 /getData ,项目就卡死index.py 代码如下&#xff1a;from flask import Flask,render_template,requestimport jsonimport pymssqlimport urllib.parseimport sysapp Flask(__name__)app.route(/)def index():return render_template(index.html)app.route(/…