(十七) 整合spring cloud云架构 -消息驱动 Spring Cloud Stream

简介: springboot

在使用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等发送接收消息。

  1. 使用工具:

rabbit,具体的下载和安装细节我这里不做太多讲解,网上的实例太多了

  1. 创建commonservice-mq-producer消息的发送者项目,在pom里面配置stream-rabbit的依赖

    <dependency>  
       <groupId>org.springframework.cloud</groupId>  
       <artifactId>spring-cloud-starter-stream-rabbit</artifactId>  
    </dependency></span>  
    1. 在yml文件里面配置rabbit mq
     port: 5666  
    spring:  
     application:  
       name: commonservice-mq-producer  
     profiles:   
       active: dev  
     cloud:  
       config:  
         discovery:   
           enabled: true  
           service-id: commonservice-config-server  
     <span style="color: #ff0000;"># 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>  
    1. 定义接口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>  
  1. 定义绑定
<span style="font-size: 16px;">package com.honghu.cloud.producer;  
  
import org.springframework.cloud.stream.annotation.EnableBinding;  
  
@EnableBinding(ProducerService.class)  
public class SendServerConfig {  
  
}</span>  
  1. 定义发送消息业务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: #ff0000;">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>  
  1. 创建commonservice-mq-consumer1消息的消费者项目,在pom里面配置stream-rabbit的依赖

    <!-- 引入MQ消息驱动的微服务包,引入stream只需要进行配置化即可,是对rabbit、kafka很好的封装 -->  
    <dependency>  
       <groupId>org.springframework.cloud</groupId>  
       <artifactId>spring-cloud-starter-stream-rabbit</artifactId>  
    </dependency>  
  2. 在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: #ff0000;">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 
  1. 定义接口ConsumerService
package com.honghu.cloud.consumer;  
  
import org.springframework.cloud.stream.annotation.Input;  
import org.springframework.messaging.SubscribableChannel;  
  
public interface ConsumerService {  
      
    <span style="color: #ff0000;">String SCORE_INPUT = "mqScoreInput";  
  
    @Input(ConsumerService.SCORE_INPUT)  
    SubscribableChannel sendMessage();</span>  
  
}  
  1. 定义启动类和消息消费
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: #ff0000;">@StreamListener(ConsumerService.SCORE_INPUT)  
    public void onMessage(Object obj) {  
        System.out.println("消费者1,接收到的消息:" + obj);  
    }</span>  
  
}  
  1. 分别启动commonservice-mq-producer、commonservice-mq-consumer1
  2. 通过postman来验证消息的发送和接收

image.png
image.png

image.png
image.png
image.png
image.png
可以看到接收到了消息,下一章我们介绍mq的集群方案。

到此,整个消息中心方案集成完毕(企业架构源码可以加求球:叁五三陆二肆柒二伍玖)!!

欢迎大家和我一起学习spring cloud构建微服务云架构,我这边会将近期研发的spring cloud微服务云架构的搭建过程和精髓记录下来,帮助更多有兴趣研发spring cloud框架的朋友,大家来一起探讨spring cloud架构的搭建过程及如何运用于企业项目。

目录
相关文章
|
1月前
|
消息中间件 存储 Java
RabbitMQ 和 Spring Cloud Stream 实现异步通信
本文介绍了在微服务架构中,如何利用 RabbitMQ 作为消息代理,并结合 Spring Cloud Stream 实现高效的异步通信。内容涵盖异步通信的优势、RabbitMQ 的核心概念与特性、Spring Cloud Stream 的功能及其与 RabbitMQ 的集成方式。通过这种组合,开发者可以构建出具备高可用性、可扩展性和弹性的分布式系统,满足现代应用对快速响应和可靠消息传递的需求。
111 2
RabbitMQ 和 Spring Cloud Stream 实现异步通信
|
7月前
|
负载均衡 Dubbo Java
Spring Cloud Alibaba与Spring Cloud区别和联系?
Spring Cloud Alibaba与Spring Cloud区别和联系?
|
8月前
|
消息中间件 Java Kafka
【Azure Kafka】使用Spring Cloud Stream Binder Kafka 发送并接收 Event Hub 消息及解决并发报错
reactor.core.publisher.Sinks$EmissionException: Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially.
131 5
|
8月前
|
前端开发 Java Nacos
🛡️Spring Boot 3 整合 Spring Cloud Gateway 工程实践
本文介绍了如何使用Spring Cloud Alibaba 2023.0.0.0技术栈构建微服务网关,以应对微服务架构中流量治理与安全管控的复杂性。通过一个包含鉴权服务、文件服务和主服务的项目,详细讲解了网关的整合与功能开发。首先,通过统一路由配置,将所有请求集中到网关进行管理;其次,实现了限流防刷功能,防止恶意刷接口;最后,添加了登录鉴权机制,确保用户身份验证。整个过程结合Nacos注册中心,确保服务注册与配置管理的高效性。通过这些实践,帮助开发者更好地理解和应用微服务网关。
1360 0
🛡️Spring Boot 3 整合 Spring Cloud Gateway 工程实践
|
9月前
|
人工智能 安全 Java
AI 时代:从 Spring Cloud Alibaba 到 Spring AI Alibaba
本次分享由阿里云智能集团云原生微服务技术负责人李艳林主讲,主题为“AI时代:从Spring Cloud Alibaba到Spring AI Alibaba”。内容涵盖应用架构演进、AI agent框架发展趋势及Spring AI Alibaba的重磅发布。分享介绍了AI原生架构与传统架构的融合,强调了API优先、事件驱动和AI运维的重要性。同时,详细解析了Spring AI Alibaba的三层抽象设计,包括模型支持、工作流智能体编排及生产可用性构建能力,确保安全合规、高效部署与可观测性。最后,结合实际案例展示了如何利用私域数据优化AI应用,提升业务价值。
817 4
|
10月前
|
负载均衡 Java 开发者
深入探索Spring Cloud与Spring Boot:构建微服务架构的实践经验
深入探索Spring Cloud与Spring Boot:构建微服务架构的实践经验
542 5
|
消息中间件 Java Kafka
SpringCloud学习之SpringCloudStream&集成kafka
一、关于Spring-Cloud-Stream   Spring Cloud Stream本质上就是整合了Spring Boot和Spring Integration,实现了一套轻量级的消息驱动的微服务框架。
2892 0
|
3月前
|
Java Spring 容器
SpringBoot自动配置的原理是什么?
Spring Boot自动配置核心在于@EnableAutoConfiguration注解,它通过@Import导入配置选择器,加载META-INF/spring.factories中定义的自动配置类。这些类根据@Conditional系列注解判断是否生效。但Spring Boot 3.0后已弃用spring.factories,改用新格式的.imports文件进行配置。
775 0
|
8天前
|
JavaScript Java Maven
【SpringBoot(二)】带你认识Yaml配置文件类型、SpringMVC的资源访问路径 和 静态资源配置的原理!
SpringBoot专栏第二章,从本章开始正式进入SpringBoot的WEB阶段开发,本章先带你认识yaml配置文件和资源的路径配置原理,以方便在后面的文章中打下基础
112 3
|
8天前
|
Java 测试技术 数据库连接
【SpringBoot(四)】还不懂文件上传?JUnit使用?本文带你了解SpringBoot的文件上传、异常处理、组件注入等知识!并且带你领悟JUnit单元测试的使用!
Spring专栏第四章,本文带你上手 SpringBoot 的文件上传、异常处理、组件注入等功能 并且为你演示Junit5的基础上手体验
282 2

热门文章

最新文章