SpringCloud Stream消息驱动

2021-09-10 23:25:03

SpringCloud Stream消息驱动

SpringCloud Stream消息驱动

简介

什么是SpringCloudStream
  • 官方定义 SpringCloud Stream 是一个构建消息驱动微服务对框架。
  • 应用程序通过 inputs 或者 outputs 来 与Springcloud Stream 中 binder 对象交互 通过我们配置来 binding（绑定），而 SpringCloud Stream 的 binder 对象负责与消息中间件交互，所以我们只需要搞清楚如何与 Spring Coud Stream 交互就可以方便使用消息驱动的方式。
  • 通过使用SpringIntegration 来连接消息代理中间件实现消息事件驱动。
  • Spring Cloud Stream 为一些供应商的消息中间件产品提供来个性化的自动化配置实现，引用来发布-订阅、消费组、分区的三个核心概念。
  • ==目前值支持 RabbitMQ、Kafka==
    SpringCloud Stream消息驱动
  • 官网 https://spring.io/projects/spring-cloud-stream#overview

Spring Cloud Stream 是用于构建与消息传递系统的高度可伸缩的事件驱动微服务架构，该框架提供来一个灵活的编程模型，它简历在已经建立和熟悉的Spring 熟语和最佳实践上，包括支持持久化的发布订阅、消费组以及消息分区这三个核心概念

SpringCloud Stream消息驱动
API：https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.1.RELEASE/reference/html/

中文指导手册：https://m.wang1314.com/doc/webapp/topic/20971999.html

设计思想

标准MQ

SpringCloud Stream消息驱动
SpringCloud Stream消息驱动
SpringCloud Stream消息驱动

为什么用Cloud Stream
  • 比方说我们用到了RabbitMQ 和 Kafka ，由于这两个消息中间件的架构上的不同，像RabbitMQ 有 exchange，Kafka有 Topic 和 Partitions 分区，
  • SpringCloud Stream消息驱动
    这些中间件的差异性导致我们实际项目开发给我们造成一定困扰，我们如果用了两个消息队列的其中一种，后面业务需求，我们想往另一个消息队列进行迁移，这时候无疑就是一个灾难性的，一大堆东西都需要推到重新做，因为它跟我们系统耦合度很高，这时候SpringCloud Stream 给我们提供了一种解耦合的方式
stream凭什么可以统一底层差异

Biinder

  • INPUT 对应于消费者
  • OUTPUT 对应于生产者
  • 在没有绑定器这个概念的情况下，我们springBoot 应用要直接与消息中间件进行交互的时候，由于消息中间件构造不同，他们的实现细节上会有较大的差异性，通过定义绑定器作为中间层，完美地实现了应用与消息中间件细节之间的隔离。通过向应用程序暴露统一的Channel 通道，使得应用程序不需要再考虑各种不同的消息中间件实现
  • 通过绑定器Binder作为中间层，实现了应用程序与消息中间件的隔离

SpringCloud Stream消息驱动
SpringCloud Stream消息驱动

  • 通过定义绑定器Binder 作为中间层，实现了应用程序与消息中间件之间细节的隔离。
  • Stream 中的消息通信方式遵循了发布订阅 Topic主题进行广播

Spring Cloud Stream标准流程套路
SpringCloud Stream消息驱动
SpringCloud Stream消息驱动

编码API和常用注解

SpringCloud Stream消息驱动

案例：
消息驱动生产者
  1. 新建模块 cloud-stream-rabbitmq-provider8801
  2. pom
<dependencies>
    <!--stream rabbit -->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>
    <!--eureka client-->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!--监控-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    <!--热部署-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-devtools</artifactId>
        <scope>runtime</scope>
        <optional>true</optional>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>
  1. yml
server:
  port: 8801

spring:
  application:
    name: cloud-stream-provider
  cloud:
    stream:
      binders: #在此处配置要绑定的rabbitmq的服务信息
        defaultRabbit: #表示定义的名称，用于binding整合
          type: rabbit #消息组件类型
          environment: #设置rabbitmq的相关环境配置
            spring:
              rabbitmq:
                host: ip端口号  #RabbitMQ在本机的用localhost，在服务器的用服务器的ip地址
                port: 5672
                username: guest
                password: guest
      bindings: #服务的整合处理
        output: #这个名字是一个通道的名称
          destination: studyExchange #表示要使用的Exchange名称定义
          content-type: application/json #设置消息类型，本次为json
          binder: defaultRabbit #设置要绑定的消息服务的具体设置（爆红不影响使用，位置没错）

eureka:
  client:
    service-url:
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 #设置心跳的时间间隔（默认是30S)
    lease-expiration-duration-in-seconds: 5 #如果超过5S间隔就注销节点 默认是90s
    instance-id: send-8801.com #在信息列表时显示主机名称
    prefer-ip-address: true #访问的路径变为IP地址

启动类

@SpringBootApplication
public class StreamMQMain8801 {

    public static void main(String[] args) {
        SpringApplication.run(StreamMQMain8801.class, args);
    }

}
  1. 新建业务类

新建service包 --- IMessageProvider接口 与实现类

public interface IMessageProvider {
    public String send();
}
@EnableBinding(Source.class)    //定义消息的推送管道（Source是spring的）
public class IMessageProviderImpl implements IMessageProvider {

    @Resource
    private MessageChannel output;  //消息发送管道

    @Override
    public String send() {
        String serial = UUID.randomUUID().toString();
        output.send(MessageBuilder.withPayload(serial).build());     //MessageBuilder是spring的integration.support.MessageBuilder
        System.out.println("*******serial: " + serial);
        return null;
    }
}
@RestController
public class SendMessageController {

    @Resource
    private IMessageProvider iMessageProvider;

    @GetMapping("/sendMessage")
    public String sendMessage(){
        return iMessageProvider.send();
    }

}
消费者
  1. 新建模块cloud-stream-rabbitmq-consumer8802
  2. pom
<dependencies>
    <!--stream rabbit -->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>
    <!--eureka client-->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!--监控-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    <!--热部署-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-devtools</artifactId>
        <scope>runtime</scope>
        <optional>true</optional>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>
  1. yml
server:
  port: 8802

spring:
  application:
    name: cloud-stream-provider
  cloud:
    stream:
      binders: #在此处配置要绑定的rabbitmq的服务信息
        defaultRabbit: #表示定义的名称，用于binding整合
          type: rabbit #消息组件类型
          environment: #设置rabbitmq的相关环境配置
            spring:
              rabbitmq:
                host: ip端口  #RabbitMQ在本机的用localhost，在服务器的用服务器的ip地址
                port: 5672
                username: 用户名
                password: 密码
      bindings: #服务的整合处理
        input: #这个名字是一个通道的名称
          destination: studyExchange #表示要使用的Exchange名称定义
          content-type: application/json #设置消息类型，本次为json，本文要设置为“text/plain”
          binder: defaultRabbit #设置要绑定的消息服务的具体设置（爆红不影响使用，位置没错）

eureka:
  client:
    service-url:
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 #设置心跳的时间间隔（默认是30S)
    lease-expiration-duration-in-seconds: 5 #如果超过5S间隔就注销节点 默认是90s
    instance-id: receive-8802.com #在信息列表时显示主机名称
    prefer-ip-address: true #访问的路径变为IP地址
  1. 主启动类
@SpringBootApplication
public class StreamMQMain8802 {

    public static void main(String[] args) {
        SpringApplication.run(StreamMQMain8802.class, args);
    }

}
  1. 新建 Controller
@EnableBinding(Sink.class)
@Controller
public class ReceiveMessageListenerController {

    @Value("${server.port}")
    private String serverPort;

    @StreamListener(Sink.INPUT) //监听
    public void input(Message<String> message){
        System.out.println("消费者1号------>收到的消息：" + message.getPayload() + "\t port：" + serverPort);
    }

}

启动Eureka 消息驱动生产者，消费者

SpringCloud Stream消息驱动

SpringCloud Stream消息驱动

上一篇：QoS最佳实践，实现更好的带宽管理


下一篇：SQL中采用Newtonsoft.Json处理json字符串