(十七) 整合spring cloud云架构 -消息驱动 Spring Cloud Stream
在使用spring cloud云架构的时候,我们不得不使用Spring cloud Stream,因为消息中间件的使用在项目中无处不在,我们公司后面做了娱乐方面的APP,在使用spring cloud做架构的时候,其中消息的异步通知,业务的异步处理都需要使用消息中间件机制。spring cloud的官方给出的集成建议(使用rabbit mq和kafka),我看了一下源码和配置,只要把rabbit mq集成,kafka只是换了一个pom配置jar包而已,闲话少说,我们就直接进入配置实施:
- 简介:
Spring cloud Stream 数据流操作开发包,封装了与Redis,Rabbit、Kafka等发送接收消息。 - 使用工具:
rabbit,具体的下载和安装细节我这里不做太多讲解,网上的实例太多了 - 创建commonservice-mq-producer消息的发送者项目,在pom里面配置stream-rabbit的依赖
<span style="font-size: 16px;"><!-- 引入MQ消息驱动的微服务包,引入stream只需要进行配置化即可,是对rabbit、kafka很好的封装 --> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-stream-rabbit</artifactId> </dependency></span>
4.在yml文件里面配置rabbit mq
<span style="font-size: 16px;">server: port: 5666 spring: application: name: commonservice-mq-producer profiles: active: dev cloud: config: discovery: enabled: true service-id: commonservice-config-server <span style="color: rgb(255, 0, 0);"># rabbitmq和kafka都有相关配置的默认值,如果修改,可以再次进行配置 stream: bindings: mqScoreOutput: destination: honghu_exchange contentType: application/json rabbitmq: host: localhost port: 5672 username: honghu password: honghu</span> eureka: client: service-url: defaultZone: http://honghu:123456@localhost:8761/eureka instance: prefer-ip-address: true</span>
5定义接口ProducerService
<span style="font-size: 16px;">package com.honghu.cloud.producer; import org.springframework.cloud.stream.annotation.Output; import org.springframework.messaging.SubscribableChannel; public interface ProducerService { String SCORE_OUPUT = "mqScoreOutput"; @Output(ProducerService.SCORE_OUPUT) SubscribableChannel sendMessage(); }</span>
6 定义绑定
<span style="font-size: 16px;">package com.honghu.cloud.producer; import org.springframework.cloud.stream.annotation.EnableBinding; @EnableBinding(ProducerService.class) public class SendServerConfig { }</span>
7定义发送消息业务ProducerController
<span style="font-size: 16px;">package com.honghu.cloud.controller; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.RestController; import com.honghu.cloud.common.code.ResponseCode; import com.honghu.cloud.common.code.ResponseVO; import com.honghu.cloud.entity.User; import com.honghu.cloud.producer.ProducerService; import net.sf.json.JSONObject; @RestController @RequestMapping(value = "producer") public class ProducerController { @Autowired private ProducerService producerService; /** * 通过get方式发送</span>对象<span style="font-size: 16px;"> * @param name 路径参数 * @return 成功|失败 */ @RequestMapping(value = "/sendObj", method = RequestMethod.GET) public ResponseVO sendObj() { User user = new User(1, "hello User"); <span style="color: rgb(255, 0, 0);">Message<User> msg = MessageBuilder.withPayload(user).build();</span> boolean result = producerService.sendMessage().send(msg); if(result){ return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_SUCCESS, false); } return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_FAILURE, false); } /** * 通过get方式发送字符串消息 * @param name 路径参数 * @return 成功|失败 */ @RequestMapping(value = "/send/{name}", method = RequestMethod.GET) public ResponseVO send(@PathVariable(value = "name", required = true) String name) { Message msg = MessageBuilder.withPayload(name.getBytes()).build(); boolean result = producerService.sendMessage().send(msg); if(result){ return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_SUCCESS, false); } return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_FAILURE, false); } /** * 通过post方式发送</span>json对象<span style="font-size: 16px;"> * @param name 路径参数 * @return 成功|失败 */ @RequestMapping(value = "/sendJsonObj", method = RequestMethod.POST) public ResponseVO sendJsonObj(@RequestBody JSONObject jsonObj) { Message<JSONObject> msg = MessageBuilder.withPayload(jsonObj).build(); boolean result = producerService.sendMessage().send(msg); if(result){ return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_SUCCESS, false); } return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_FAILURE, false); } } </span>
8创建commonservice-mq-consumer1消息的消费者项目,在pom里面配置stream-rabbit的依赖
<!-- 引入MQ消息驱动的微服务包,引入stream只需要进行配置化即可,是对rabbit、kafka很好的封装 --> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-stream-rabbit</artifactId> </dependency>
9在yml文件中配置:
server: port: 5111 spring: application: name: commonservice-mq-consumer1 profiles: active: dev cloud: config: discovery: enabled: true service-id: commonservice-config-server <span style="color: rgb(255, 0, 0);">stream: bindings: mqScoreInput: group: honghu_queue destination: honghu_exchange contentType: application/json rabbitmq: host: localhost port: 5672 username: honghu password: honghu</span> eureka: client: service-url: defaultZone: http://honghu:123456@localhost:8761/eureka instance: prefer-ip-address: true
10定义接口ConsumerService
package com.honghu.cloud.consumer; import org.springframework.cloud.stream.annotation.Input; import org.springframework.messaging.SubscribableChannel; public interface ConsumerService { <span style="color: rgb(255, 0, 0);">String SCORE_INPUT = "mqScoreInput"; @Input(ConsumerService.SCORE_INPUT) SubscribableChannel sendMessage();</span> }
11 定义启动类和消息消费
package com.honghu.cloud; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.cloud.netflix.eureka.EnableEurekaClient; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.StreamListener; import com.honghu.cloud.consumer.ConsumerService; import com.honghu.cloud.entity.User; @EnableEurekaClient @SpringBootApplication @EnableBinding(ConsumerService.class) //可以绑定多个接口 public class ConsumerApplication { public static void main(String[] args) { SpringApplication.run(ConsumerApplication.class, args); } <span style="color: rgb(255, 0, 0);">@StreamListener(ConsumerService.SCORE_INPUT) public void onMessage(Object obj) { System.out.println("消费者1,接收到的消息:" + obj); }</span> }
12 分别启动commonservice-mq-producer、commonservice-mq-consumer1
13 通过postman来验证消息的发送和接收
可以看到接收到了消息,下一章我们介绍mq的集群方案。
到此,整个消息中心方案集成完毕
欢迎大家和我一起学习spring cloud构建微服务云架构,我这边会将近期研发的spring cloud微服务云架构的搭建过程和精髓记录下来,帮助更多有兴趣研发spring cloud框架的朋友,大家来一起探讨spring cloud架构的搭建过程及如何运用于企业项目。
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
新品发布、降价普惠、拥抱开源、出海全球化 | 杭州云栖企业数字化转型峰会上的那些关键词
9月19日,在杭州云栖大会 - 企业数字化转型峰会现场,阿里巴巴中间件产品总监赵林分享了2018 Aliware的最新产品动态。本文将为您梳理Aliware在出海全球化、开源支持、消息队列高级特性降价、链路追踪新品发布、应用高可用新品发布、CloudToolkit 新品发布等方面的最新进展。 基于Aliware构建企业级互联网架构 经过10余年的技术沉淀,阿里中间件已推出企业级分布式应用服务EDAS、性能测试PTS、消息队列MQ、性能监控ARMS、应用配置管理服务ACM、全局事务服务GTS、分布式数据库服务DRDS等10余款中间件产品,其PaaS产品和中台思想已在各行业快速复制,并从企业组织的数字化、企业应用的数字化、企业数据的数字化三个维度,帮助各行业客户进行业务创新,实现数字化转型。 Aliware出海全球化 随着阿里云逐步在海外新加
- 下一篇
(十八) 整合spring cloud云架构 -后台管理基础功能简介
项目介绍 鸿鹄云开发平台是一个大型分布式、微服务、云架构、面向企业的 JavaEE体系快速研发平台,基于模块化、服务化、原子化、热插拔的设计思想,使用成熟领先的无商业限制的主流开源技术构建。 采用服务化的组件开发模式,可实现复杂的业务功能。使用Maven进行项目的构建管理,采用Jenkins进行持续集成,主要定位于大型分布式企业系统或大型分布式互联网产品的架构。使用当前最流行最先进的Spring Cloud技术实现服务组件化及管理,真正为企业打造分布式微服务云架构平台。 使用技术(技术使用太多,这里只列了一部分) SOA服务框架:SpringCloud 、SpringBoot、RestFul等 分布式缓存:Redis 模块化管理:Maven 数据库连接池:Alibaba Druid 核心框架:Spring framework、SpringBoot 持久层框架:MyBatis 安全框架:Apache Shiro 服务端验证:Hibernate Validator 任务调度:quartz 日志管理:SLF4J 1.7、Log4j 客户端验证:JQuery Validation 动态页签:e...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- Mario游戏-低调大师作品
- CentOS关闭SELinux安全模块
- CentOS7,CentOS8安装Elasticsearch6.8.6
- CentOS8安装MyCat,轻松搞定数据库的读写分离、垂直分库、水平分库
- Docker安装Oracle12C,快速搭建Oracle学习环境
- Red5直播服务器,属于Java语言的直播服务器
- CentOS8编译安装MySQL8.0.19
- SpringBoot2整合MyBatis,连接MySql数据库做增删改查操作
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- CentOS6,CentOS7官方镜像安装Oracle11G