Spring Cloud构建微服务架构:消息驱动的微服务(入门)【Dalston版】

2019独角兽企业重金招聘Python工程师标准>>> hot3.png

之前在写Spring Boot基础教程的时候写过一篇《Spring Boot中使用RabbitMQ》。在该文中,我们通过简单的配置和注解就能实现向RabbitMQ中生产和消费消息。实际上我们使用的对RabbitMQ的starter就是通过Spring Cloud Stream中对RabbitMQ的支持来实现的。下面我们就通过本文来了解一下Spring Cloud Stream。

Spring Cloud Stream是一个用来为微服务应用构建消息驱动能力的框架。它可以基于Spring Boot来创建独立的、可用于生产的Spring应用程序。它通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动的微服务应用。Spring Cloud Stream为一些供应商的消息中间件产品提供了个性化的自动化配置实现,并且引入了发布-订阅、消费组以及消息分区这三个核心概念。简单的说,Spring Cloud Stream本质上就是整合了Spring Boot和Spring Integration,实现了一套轻量级的消息驱动的微服务框架。通过使用Spring Cloud Stream,可以有效地简化开发人员对消息中间件的使用复杂度,让系统开发人员可以有更多的精力关注于核心业务逻辑的处理。由于Spring Cloud Stream基于Spring Boot实现,所以它秉承了Spring Boot的优点,实现了自动化配置的功能帮忙我们可以快速的上手使用,但是目前为止Spring Cloud Stream只支持下面两个著名的消息中间件的自动化配置:

  • RabbitMQ
  • Kafka

快速入门

下面我们通过构建一个简单的示例来对Spring Cloud Stream有一个初步认识。该示例主要目标是构建一个基于Spring Boot的微服务应用,这个微服务应用将通过使用消息中间件RabbitMQ来接收消息并将消息打印到日志中。所以,在进行下面步骤之前请先确认已经在本地安装了RabbitMQ,具体安装步骤请参考此文。

构建一个Spring Cloud Stream消费者

  • 创建一个基础的Spring Boot工程,命名为:stream-hello

  • 编辑pom.xml中的依赖关系,引入Spring Cloud Stream对RabbitMQ的支持,具体如下:

<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>1.5.9.RELEASE</version><relativePath/> <!-- lookup parent from repository -->
</parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId>     </dependency>
</dependencies><dependencyManagement><dependencies><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-dependencies</artifactId><version>Dalston.SR4</version><type>pom</type><scope>import</scope></dependency></dependencies>
</dependencyManagement>
  • 创建用于接收来自RabbitMQ消息的消费者SinkReceiver,具体如下:
@EnableBinding(Sink.class)
public class SinkReceiver {private static Logger logger = LoggerFactory.getLogger(SinkReceiver.class);@StreamListener(Sink.INPUT)public void receive(Object payload) {logger.info("Received: " + payload);}}
  • 创建应用主类,这里同其他Spring Boot一样,没有什么特别之处,具体如下:
@SpringBootApplication
public class SinkApplication {public static void main(String[] args) {SpringApplication.run(SinkApplication.class, args);}}

到这里,我们快速入门示例的编码任务就已经完成了。下面我们分别启动RabbitMQ以及该Spring Boot应用,然后做下面的试验,看看它们是如何运作的。

手工测试验证

  • 我们先来看一下Spring Boot应用的启动日志。
...
INFO 16272 --- [main] o.s.c.s.b.r.RabbitMessageChannelBinder   : declaring queue for inbound: input.anonymous.Y8VsFILmSC27eS5StsXp6A, bound to: input
INFO 16272 --- [main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: SimpleConnection@3c78e551 [delegate=amqp://guest@127.0.0.1:5672/]
INFO 16272 --- [main] o.s.integration.channel.DirectChannel    : Channel 'input.anonymous.Y8VsFILmSC27eS5StsXp6A.bridge' has 1 subscriber(s).
INFO 16272 --- [main] o.s.i.a.i.AmqpInboundChannelAdapter      : started inbound.input.anonymous.Y8VsFILmSC27eS5StsXp6A
...

从上面的日志内容中,我们可以获得以下信息:

  • 使用guest用户创建了一个指向127.0.0.1:5672位置的RabbitMQ连接,在RabbitMQ的控制台中我们也可以发现它。

  • 声明了一个名为input.anonymous.Y8VsFILmSC27eS5StsXp6A的队列,并通过RabbitMessageChannelBinder将自己绑定为它的消费者。这些信息我们也能在RabbitMQ的控制台中发现它们。

下面我们可以在RabbitMQ的控制台中进入input.anonymous.Y8VsFILmSC27eS5StsXp6A队列的管理页面,通过Publish Message功能来发送一条消息到该队列中。

此时,我们可以在当前启动的Spring Boot应用程序的控制台中看到下面的内容:

INFO 16272 --- [C27eS5StsXp6A-1] com.didispace.HelloApplication           : Received: [B@7cba610e

我们可以发现在应用控制台中输出的内容就是SinkReceiverreceive方法定义的,而输出的具体内容则是来自消息队列中获取的对象。这里由于我们没有对消息进行序列化,所以输出的只是该对象的引用,在后面的小节中我们会详细介绍接收消息后的处理。

在顺利完成上面快速入门的示例后,我们简单解释一下上面的步骤是如何将我们的Spring Boot应用连接上RabbitMQ来消费消息以实现消息驱动业务逻辑的。

首先,我们对Spring Boot应用做的就是引入spring-cloud-starter-stream-rabbit依赖,该依赖包是Spring Cloud Stream对RabbitMQ支持的封装,其中包含了对RabbitMQ的自动化配置等内容。从下面它定义的依赖关系中,我们还可以知道它等价于spring-cloud-stream-binder-rabbit依赖。

<dependencies><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-stream-binder-rabbit</artifactId></dependency>
</dependencies>

接着,我们再来看看这里用到的几个Spring Cloud Stream的核心注解,它们都被定义在SinkReceiver中:

  • @EnableBinding,该注解用来指定一个或多个定义了@Input@Output注解的接口,以此实现对消息通道(Channel)的绑定。在上面的例子中,我们通过@EnableBinding(Sink.class)绑定了Sink接口,该接口是Spring Cloud Stream中默认实现的对输入消息通道绑定的定义,它的源码如下:
public interface Sink {String INPUT = "input";@Input(Sink.INPUT)SubscribableChannel input();}

它通过@Input注解绑定了一个名为input的通道。除了Sink之外,Spring Cloud Stream还默认实现了绑定output通道的Source接口,还有结合了SinkSourceProcessor接口,实际使用时我们也可以自己通过@Input@Output注解来定义绑定消息通道的接口。当我们需要为@EnableBinding指定多个接口来绑定消息通道的时候,可以这样定义:@EnableBinding(value = {Sink.class, Source.class})

  • @StreamListener:该注解主要定义在方法上,作用是将被修饰的方法注册为消息中间件上数据流的事件监听器,注解中的属性值对应了监听的消息通道名。在上面的例子中,我们通过@StreamListener(Sink.INPUT)注解将receive方法注册为对input消息通道的监听处理器,所以当我们在RabbitMQ的控制页面中发布消息的时候,receive方法会做出对应的响应动作。

编写消费消息的单元测试用例

上面我们通过RabbitMQ的控制台完成了发送消息来验证了消息消费程序的功能,虽然这种方法比较low,但是通过上面的步骤,相信大家对RabbitMQ和Spring Cloud Stream的消息消费已经有了一些基础的认识。下面我们通过编写生产消息的单元测试用例来完善我们的入门内容。

  • 在上面创建的工程中创建单元测试类:
@RunWith(SpringRunner.class)
@EnableBinding(value = {SinkApplicationTests.SinkSender.class})
public class SinkApplicationTests {@Autowiredprivate SinkSender sinkSender;@Testpublic void sinkSenderTester() {sinkSender.output().send(MessageBuilder.withPayload("produce a message :http://blog.didispace.com").build());}public interface SinkSender {String OUTPUT = "input";@Output(SinkSender.OUTPUT)MessageChannel output();}}
  • 在应用了上面的消息消费者程序之后,运行这里定义的单元测试程序,我们马上就能在消息消费者的控制台中收到下面的内容:
INFO 50947 --- [L2W-c2AcChb2Q-1] com.didispace.stream.SinkReceiver        : Received: produce a message :http://blog.didispace.com

在上面的单元测试中,我们通过@Output(SinkSender.OUTPUT)定义了一个输出通过,而该输出通道的名称为input,与前文中的Sink中定义的消费通道同名,所以这里的单元测试与前文的消费者程序组成了一对生产者与消费者。到这里,本文的内容就次结束,如果您能够独立的完成上面的例子,那么对于Spring Cloud Stream的基础使用算是入门了。但是,Spring Cloud Stream的使用远不止于此,在近期的博文中,我讲继续更新这部分内容,帮助他们来理解和用好Spring Cloud Stream来构建消息驱动的微服务!

本文完整实例:

  • Github
  • Gitee

如果您对这些感兴趣,欢迎star、follow、收藏、转发给予支持!

本文内容部分节选自我的《Spring Cloud微服务实战》,但对依赖的Spring Boot和Spring Cloud版本做了升级。

本文首发于我的博客:http://blog.didispace.com

系列教程推荐

  • Spring Boot基础教程
  • Spring Cloud基础教程

转载于:https://my.oschina.net/didispace/blog/1592356

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

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

相关文章

CXF 客服端调用报错

服务端已经发布了WSDL&#xff0c;现在在客服端生成web service客服端代码&#xff0c;在eclipse中新建一个project&#xff0c;然后new->web services->web service client生产客户端代码 在调用的时候报如下错误 解决&#xff1a;缺少axis相应的jar包&#xff0c;加入包…

20145225 《信息安全系统设计基础》第10周学习总结

cp1.c 进行复制文件的操作&#xff0c;需要有源文件和目的文件&#xff0c;第一次命令没有加入所以没有正常完成复制文件的操作fileinfo.c 用来实现显示文件信息。先判断命令是否有操作数&#xff0c;有的话才能继续进行下去&#xff0c;如果没有报错就打印出来相关文件信息&am…

做演员是圆梦 做生意学会面对现实

田朴珺是一位拥有多重身份的女性。她是一名演员&#xff0c;也是一位商人&#xff0c;还担任过电影《中国合伙人》的制片人。 作为演员&#xff0c;田朴珺的作品并不是很多&#xff0c;也一直不温不 火。但这并不代表她将放弃演艺生涯。她表示&#xff0c;如果机会合适&…

【深度学习】——模型评估指标MAP计算实例计算

目录 一、知识储备 1、IOU——交集面积与并集面积之比 2、混淆矩阵&#xff08;TP、FP、FN、TN&#xff09; 问题1&#xff1a;上面的TP等具体是如何计算得到的&#xff1f; 3、精度precision&召回率recall 二、ap计算实战 1、计算流程 1&#xff09;准备数据&#xf…

第 52 章 Web Server Optimization

系统配置 Intel(R) Xeon(TM) CPU 3.00GHzMemory 4GEthernet adapter 1000M52.1. ulimit 查看 ulimit ulimit -a core file size (blocks, -c) 0 data seg size (kbytes, -d) unlimited file size (blocks, -f) unlimited pending signals …

hdu5489 Removed Interval dp+线段树优化

现在看这题居然直接秒了。。。去年看的时候还以为神题。。 设以第i项为结尾的lis前缀为f[i]&#xff0c;以第j项为结尾的lis后缀为g[i]&#xff0c;如果求出f[i]和g[j]&#xff0c;然后枚举i&#xff0c;快速找到最大的满足a[j]>a[i]的g[j]就可以了。注意到如果将f[i]从后往…

JS原型链理解

1. 每个对象都有原型属性(__proto__)2. 对象的原型(__proto__)指向其构造函数(Constructor)的prototype属性3. 构造函数(Constructor)的prototype属性本身也是一个对象&#xff0c;其原型(__proto__)亦指向其构造函数的prototype4. 如此形成一个链式结构&#xff0c;而Construc…

【深度学习】——2021年FPN特征金字塔

#!/usr/bin/env python # -*- coding: utf-8 -*- # Time : 2021/4/22 17:06 # Author : linlianqin # Site : # File : fpn.py # Software: PyCharm # description:其搭建的基本流程和resnet是一致的&#xff0c;只是将每一层的卷积结果保存了起来import torch impo…

NoSQL分类及ehcache memcache redis 三大缓存的对比

NoSQL分类 由于NoSQL中没有像传统数据库那样定义数据的组织方式为关系型的&#xff0c;所以只要内部的数据组织采用了非关系型的方式&#xff0c;就可以称之为NoSQL数据库。目前&#xff0c;可以将众多的NoSQL数据库按照内部的数据组织形式进行如下分类&#xff1a; Key/Value的…

52.4. APC Cache (php-apc - APC (Alternative PHP Cache) module for PHP 5)

$ apt-cache search php-apc php-apc - APC (Alternative PHP Cache) module for PHP 5$ sudo apt-get install php-apcapc cache 状态监控 http://pecl.php.net/package/APC 下载解包找到apc.php,放到web服务器上 原文出处&#xff1a;Netkiller 系列 手札 本文作者&#xff1…

乐视云计算基于OpenStack的IaaS实践

本文作者岳龙广&#xff0c;现在就职于乐视云计算有限公司&#xff0c;负责IaaS部门的工作。 从开始工作就混在开源世界里&#xff0c;在虚拟化方面做过CloudStack/Ovirt开发&#xff0c;现在是做以OpenStack为基础的乐视云平台。所以对虚拟化情有独钟&#xff0c;也对虚拟化/云…

【深度学习】——如何提高map值

目录 代码获取 map原理 map提高技巧 技巧总结&#xff1a; 实战&#xff1a; 1、效果不佳map55.55% 1&#xff09;单独调整get_dr_txt.py中的self.iou 0.3 2&#xff09;单独调整get_map,py中的minoverlap: 3)同时调整minoverlap和self.iou 本文是在faster_rcnn模型的…

每日站立会议个人博客(冲刺周)-Wednesday

时间未完成不知道如何获取具体标签里的内容正在做爬虫技术之获取标签里的内容将要做对运用爬虫技术获取的数据进行处理转载于:https://www.cnblogs.com/andibier/p/8075098.html

数据库水平切分的实现原理解析——分库,分表,主从,集群,负载均衡器(转)...

第1章 引言 随着互联网应用的广泛普及&#xff0c;海量数据的存储和访问成为了系统设计的瓶颈问题。对于一个大型的互联网应用&#xff0c;每天几十亿的PV无疑对数据库造成了相当高的负载。对于系统的稳定性和扩展性造成了极大的问题。通过数据切分来提高网站性能&#xff0c;横…

【深度学习】——纠错error: Unable to find vcvarsall.bat:关于安装pycocotools

1、安装包下载 大佬改写支持 Windows 的 COCO 地址&#xff1a;https://github.com/philferriere/cocoapi 下载后如下&#xff1a; 进入pythonAPI 先后运行&#xff1a; python setup.py build_ext --inplacepython setup.py build_ext install 出现以下标志时&#xff0c…

【小贴士】虚拟键盘与fixed带给移动端的痛!

前言今天来公司的主要目的就是研究虚拟键盘与fixed的问题&#xff0c;期间因为同事问起闭包与事件委托&#xff08;阻止冒泡&#xff09;相关问题&#xff0c;便穿插了一篇别的&#xff1a;【小贴士】工作中的”闭包“与事件委托的”阻止冒泡“&#xff0c;有兴趣的朋友可以去看…

[OJ] Wildcard Matching (Hard)

LintCode 192. Wildcard Matching (Hard)LeetCode 44. Wildcard Matching (Hard) 第二次刷还是被这题虐. 其实就是跪在一个地方, 就是关于mustFail的地方. 当*p && !*s的时候, 说明s已经被用完了, p还没有被穷尽, 这种情况下要直接退出所有的递归返回false, 因为s都匹配…

CSS3 -webkit-transition(属性渐变)

-webkit-transition&#xff1a;CSS属性(none|all|属性) 持续时间 时间函数 延迟时间 CSS属性(transition-property)&#xff1a;要变化的属性&#xff0c;比如元素变宽则是width&#xff0c;文字颜色要变色这是color&#xff1b;W3C给出了一个可变换属性的列表&#xff1a;…

vxworks的default boot line说明

boot程序的主要功能是引导vxworks 内核,所以boot程序需要知道vxworks的内核存放在何处&#xff0c;通过什么手段去获取。在vxworks缺省的boot程序里有一条内建的default boot line,它指明了获得vxworks内核的途径&#xff0c;在boot程序启动时&#xff0c;它先寻找NVRAM里面有无…

【机器视觉】——相机和镜头的选择

目录 1、相机选择 2、镜头选择 3、其他计算公式 1)芯片尺寸计算: