在使用spring cloud云架构的时候,我们不得不使用Spring cloud Stream,因为消息中间件的使用在项目中无处不在,我们公司后面做了娱乐方面的APP,在使用spring cloud做架构的时候,其中消息的异步通知,业务的异步处理都需要使用消息中间件机制。spring cloud的官方给出的集成建议(使用rabbit mq和kafka),我看了一下源码和配置,只要把rabbit mq集成,kafka只是换了一个pom配置jar包而已,闲话少说,我们就直接进入配置实施:
1. 简介:
Spring cloud Stream 数据流操作开发包,封装了与Redis,Rabbit、Kafka等发送接收消息。
2. 使用工具:
rabbit,具体的下载和安装细节我这里不做太多讲解,网上的实例太多了
3. 创建commonservice-mq-producer消息的发送者项目,在pom里面配置stream-rabbit的依赖
<!-- 引入MQ消息驱动的微服务包,引入stream只需要进行配置化即可,是对rabbit、kafka很好的封装 -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
4. 在yml文件里面配置rabbit mq
server:
port: 5666
spring:
application:
name: commonservice-mq-producer
profiles:
active: dev
cloud:
config:
discovery:
enabled: true
service-id: commonservice-config-server
# rabbitmq和kafka都有相关配置的默认值,如果修改,可以再次进行配置
stream:
bindings:
mqScoreOutput:
destination: honghu_exchange
contentType: application/json
rabbitmq:
host: localhost
port: 5672
username: honghu
password: honghu
eureka:
client:
service-url:
defaultZone: http://honghu:123456@localhost:8761/eureka
instance:
prefer-ip-address: true
5. 定义接口ProducerService
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();
}
6. 定义绑定
package com.honghu.cloud.producer;
import org.springframework.cloud.stream.annotation.EnableBinding;
@EnableBinding(ProducerService.class)
public class SendServerConfig {
}
7. 定义发送消息业务ProducerController
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方式发送对象 * @param name 路径参数 * @return 成功|失败 */ @RequestMapping(value = "/sendObj", method = RequestMethod.GET) public ResponseVO sendObj() { User user = new User(1, "hello User"); Message<User> msg = MessageBuilder.withPayload(user).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); } /** * 通过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方式发送json对象 * @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); } }
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
stream:
bindings:
mqScoreInput:
group: honghu_queue
destination: honghu_exchange
contentType: application/json
rabbitmq:
host: localhost
port: 5672
username: honghu
password: honghu
eureka:
client:
service-url:
defaultZone: http://honghu:123456@localhost:8761/eureka
instance:
prefer-ip-address: true
9. 定义接口ConsumerService
package com.honghu.cloud.consumer;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
public interface ConsumerService {
String SCORE_INPUT = "mqScoreInput";
@Input(ConsumerService.SCORE_INPUT)
SubscribableChannel sendMessage();
}
10. 定义启动类和消息消费
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);
}
@StreamListener(ConsumerService.SCORE_INPUT)
public void onMessage(Object obj) {
System.out.println("消费者1,接收到的消息:" + obj);
}
}
11. 分别启动commonservice-mq-producer、commonservice-mq-consumer1
12. 通过postman来验证消息的发送和接收
可以看到接收到了消息,下一章我们介绍mq的集群方案。
到此,整个消息中心方案集成完毕(需要源码可以加qq:2147775633)!!
欢迎大家和我一起学习spring cloud构建微服务云架构,我这边会将近期研发的spring cloud微服务云架构的搭建过程和精髓记录下来,帮助更多有兴趣研发spring cloud框架的朋友,大家来一起探讨spring cloud架构的搭建过程及如何运用于企业项目。
相关推荐
各个微服务统一使用spring-cloud-starter-stream-rabbit 为规范各个微服务之间消息队列的交换机、通道、队列、消息投递更加便于识别、扩展和维护特进行以下要求
springcloud-stream-demo-master.zip
spring-cloud-stream-samples, spring 云流示例 spring Cloud示例应用程序这个库包含使用 spring 云流编写的应用程序的集合。 所有的应用程序都是自包含的。 它们可以针对 Kafka 或者RabbitMQ中间件技术运行。 你...
介绍Spring Cloud Stream与RabbitMQ集成的代码示例。Spring Cloud Stream是一个建立在Spring Boot和Spring Integration之上的框架,有助于创建事件驱动或消息驱动的微服务。
spring-cloud-stream-app-descriptor-Celsius.SR3.stream-apps-kafka-10-docker
spring-cloud
spring cloud stream 整合 rabbitmq , 自定义消息通道,既能发消息,也能收消息
spring-cloud之spring-cloud-stream,底层为rabbitmq实现,简单栗子,仅供参考。生产者、消费者配置及生产者、消费者代码实现。
springcloud生产者与消费者项目实战案例 Spring Cloud 中断路器 Circuit Breaker的应用 配置 Spring Cloud Config Server Spring Cloud Config使用Oracle数据库作为后端配置存储 Spring Cloud Config + Spring Cloud...
spring-cloud-sleuth-master/spring-cloud-security-master/spring-cloud-cluster-master/spring-cloud-stream-master/ribbon-master/zuul-master/eureka-master/Hystrix-master/feign-master/pring-cloud-stream-...
基于微服务架构(spring cloud)的微信聊天界面,发送消息基于stream流 服务治理:spring-boot-admin,config,eureka,gateway 文件说明:startall.sh为一键启动 所有.sh文件都是shell脚本,文件迁移部署为docker技术...
spring cloud整合MQTT简单示例,分为三个项目eureka-server、service-hi、service-ribbon,MQTT环境需要自己提前搭好,我这搭的环境是apache-apollo-1.7.1
官方定义SpringCloudStream是一个构建消息驱动微服务的框架。SpringCloudStreamApplication应用程序通过inputs或者outputs来与SpringCloudStream中binder交互,通过我们配置来binding,而SpringCloudStream的binder...
Spring云流侦探用于将Spring Cloud Stream与Spring Cloud Sleuth结合使用的示例回购。 包含两个Spring Boot项目,每个项目都有一个简单的Spring Boot微服务。 使用mvnw spring-boot:run运行它们在运行应用程序和发送...
春天云流演示 探索Spring Cloud Stream Spring-Cloud-Stream是底层消息传递系统的抽象。 它有助于以最少的精力来建立消息驱动的系统,以建立代理并连接生产者和消费者。
SpringCloud Stream 消息驱动 SpringCloud Sleuth 分布式请求链路跟踪 SpringCloud Alibaba入门简介 SpringCloud Alibaba Nacos 服务注册和配置中心 SpringCloud Alibaba Sentinel 实现熔断与限流 SpringCloud ...
春云流兔Spring Cloud Stream示例展示了各种消息交换模式。 该项目基于Baeldung构建的SCS示例,并进行了修改以与Elmhurst版本或更高版本一起使用。 示例包括: MessageConverter:样本处理器应用程序,它接收一条...
ddd-springcloud-stream.zip
springCloud-分享版: 基于Spring Cloud、OAuth2.0的前后端分离的系统。 通用RBAC权限设计及其数据权限和分库分表 支持服务限流、动态路由、灰度发布、 支持常见登录方式, 多系统SSO登录 基于 Spring Cloud Hoxton ...
spring-cloud-stream-kafka:Spring Cloud Streams Kafka Avro