亲宝软件园·资讯

展开

Spring Cloud Stream

怪 咖@ 人气:0

本篇文章所涉及到的demo练习 使用的cloud 2021.0.3+ springboot2.6.8

一、概述简介

官网:https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/

官网概述:https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html#spring-cloud-stream-preface-notable-deprecations

1.1. cloud Stream是什么

官方定义:Spring Cloud Stream是一个用于构建 与 共享消息系统 连接的高度可扩展的事件驱动微服务

目前主流的消息框架有:

假设公司业务项目用了RabbitMQ,而大数据项目用了Kafka。这时候就会出现有两个消息框架,相对于程序员来说其实并不友好,还得两个都掌握,正常对于一个程序员来说熟练一个消息框架都不错了,何况还搞了两个,并且两个维护起来也不好维护。

RabbitMQ和Kafka是两个不同的框架,两个消息模型上也存在着差异,并且代码上用法也不一样。Spring Cloud Stream就是不再关注具体MQ的细节,可以在不改代码的基础上,来完成Rabbit和Kafka两个不同的消息中间件的切换(这里的切换指的是原本用的RabbitMQ,但是用着用着发现kafka比较符合,所以想要换框架)。

总结成一句话:屏蔽底层消息中间件的差异,降低切换成本,统一消息的编程模型

注意:遗憾的是目前仅支持RabbitMQ、Kafka

1.2. 设计思想

常规的MQ设计如下:

cloud Stream设计如下:

通过定义绑定器Binder作为中间层,实现了应用程序与消息中间件细节之间的隔离。

在没有绑定器这个概念的情况下,我们的SpringBoot应用要直接与消息中间件进行信息交互的时候,由于各消息中间件构建的初衷不同,它们的实现细节上会有较大的差异性,通过定义绑定器作为中间层,完美地实现了应用程序与消息中间件细节之间的隔离。Stream对消息中间件的进一步封装,可以做到代码层面对中间件的无感知,甚至于动态的切换中间件(rabbitmq切换为kafka),使得微服务开发的高度解耦,服务可以关注更多自己的业务流程

注意:左图是官网的架构图

Binder可以生成Binding,Binding用来绑定消息容器的生产者和消费者,它有两种类型,INPUT和OUTPUT,INPUT对应于消费者,OUTPUT对应于生产者。

stream为了屏蔽差异,抽象出来了一个Binder层,而目前为止,只提供了两个框架的实现,通过具体的实现来连接消息中间件。

假如想要通过stream连接RabbitMQ就使用:

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

假如想要通过stream连接Kafka就使用:

<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>

Stream中的消息通信方式遵循了发布-订阅模式,Topic主题进行广播,在RabbitMQ就是Exchange,在Kakfa中就是Topic

1.3. 标准流程

1.4. 注解

注解完全是基于官方给的模型而定的!通过stream使用消息中间件也是非常简单的,直接使用以下注解就可以使用。

注意:注解依然是能用的,但是官方明确表示注解已经被弃用,弃用并不是不能用,而是用了会画横杠不建议用。但是功能是没有问题的,低版本的cloud是没有被弃用的。针对于注解和函数式编程两种我都会进行使用。

题外话:学技术永远是这样,技术一直在不断的更新迭代,真正学习一个技术并不是要掌握编码使用,而是要掌握他到底是什么,能干什么,要去深入理解他,对于编码,我认为其实不是很重要。就算你今天掌握了官方最新用法,回头人家又改写法了。

二、基于注解代码练习

生产者就是消息发送者,消费者就是消息接受者。这里我就不用kafka了,我直接用的是RabbitMQ。

windows下安装RabbitMQ:https://blog.csdn.net/weixin_43888891/article/details/126514021

2.1. 消息驱动之生产者

1.创建项目(可以是聚合可以是普通springboot项目)
2.添加pom

因为是和RabbitMQ整合,所以就是引入的spring-cloud-starter-stream-rabbit启动器

<properties>
    <maven.compiler.source>8</maven.compiler.source>
    <maven.compiler.target>8</maven.compiler.target>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <springboot.version>2.6.8</springboot.version>
    <springcloud.version>2021.0.3</springcloud.version>
</properties>

<dependencyManagement>
    <dependencies>
        <dependency>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-dependencies</artifactId>
           <version>${springboot.version}</version>
           <type>pom</type>
           <scope>import</scope>
       </dependency>
        <dependency>
           <groupId>org.springframework.cloud</groupId>
           <artifactId>spring-cloud-dependencies</artifactId>
           <version>${springcloud.version}</version>
           <type>pom</type>
           <scope>import</scope>
       </dependency>
    </dependencies>
</dependencyManagement>
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>
</dependencies>

3.添加application配置

server:
  port: 8801

spring:
  application:
    name: cloud-stream-provider
  cloud:
    stream:
      binders: # 在此处配置要绑定的rabbitmq的服务信息;
        defaultRabbit: # 表示定义的名称,用于于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置rabbitmq的相关的环境配置
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings: # 服务的整合处理
        output: # 这个名字是一个通道的名称
          destination: studyExchange # 表示要使用的Exchange名称定义
          content-type: application/json # 设置消息类型,本次为json,文本则设置“text/plain”
          binder: defaultRabbit # 设置要绑定的消息服务的具体设置

4.添加接口

public interface IMessageProvider {
    public String send();
}

5.添加实现类

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageChannel;

import javax.annotation.Resource;
import java.util.UUID;

// 可以理解为是一个消息的发送管道的定义
@EnableBinding(Source.class)
public class MessageProviderImpl implements IMessageProvider {

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

    @Override
    public String send() {
        String serial = UUID.randomUUID().toString();
        // 创建并发送消息
        this.output.send(MessageBuilder.withPayload(serial).build());
        System.out.println("***serial: " + serial);
        return serial;
    }
}

6.添加controller控制器

@RestController
public class SendMessageController {

    @Autowired
    private IMessageProvider iMessageProvider;

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

7.测试

(1)首先要保证RabbitMQ是可以访问的:http://localhost:15672

(2)启动项目访问:http://localhost:8801/send

下图波峰代表发送消息成功

启动后会创建交换机,名称就是application.yml当中的destination属性设置的

注意:停止服务后并没有删除交换机!!!

2.2. 消息驱动之消费者

1.创建项目
2.添加pom(pom和发送者依赖一模一样)

3.添加application配置

server:
  port: 8802

spring:
  application:
    name: cloud-stream-consumer
  cloud:
    stream:
      binders: # 在此处配置要绑定的rabbitmq的服务信息;
        defaultRabbit: # 表示定义的名称,用于于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置rabbitmq的相关的环境配置
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings: # 服务的整合处理
        input: # 这个名字是一个通道的名称
          destination: studyExchange # 表示要使用的Exchange名称定义
          content-type: application/json # 设置消息类型,本次为对象json,如果是文本则设置“text/plain”
          binder: defaultRabbit # 设置要绑定的消息服务的具体设置

4.添加监听(消费者只负责接受消息)

import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListener {
    @Value("${server.port}")
    private String serverPort;

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

5.测试

(1)启动RabbitMQ
(2)启动发送消息端服务
(3)启动消费者服务,启动后会发现,他自动会向这个交换机当中添加一个队列。

发送消息:http://localhost:8801/send
接受消息:

注意:当停止服务后消息队列会被自动删除!!!

2.3. 目前存在的问题

1.依照8802, clone出来一份运行8803,主要用来演示多个消费者的场景
2.启动8801生产者
3.启动8802消费者
4.启动8803消费者

当三个服务都启动后通过RabbitMQ界面会发现,一个交换机绑定了两个队列

运行后会发现存在两个问题:

有重复消费问题消息持久化问题

(1)重复消费问题:

发送消息后两个消费者都收到了消息:http://localhost:8801/send

比如在如下场景中,订单系统我们做集群部署,都会从RabbitMQ中获取订单信息,那如果一个订单同时被两个服务获取到,那么就会造成数据错误,我们得避免这种情况。这时我们就可以使用Stream中的消息分组来解决

注意在Stream中处于同一个group中的多个消费者是竞争关系,就能够保证消息只会被其中一个应用消费一次。不同组是可以全面消费的(重复消费),同一组内会发生竞争关系,只有其中一个可以消费。

(2)消息持久化问题:

当生产者发送消息的时候,消费者恰好宕机了,但是过一会消费者恢复了,但是消息却没收到。那也就是意味着消息队列是临时消息队列。针对于这一点,大家也可以测试一下,加深一下印象。

2.4. 分组解决重复消费问题

原理: 微服务应用放置于同一个group中,就能够保证消息只会被其中一个应用消费一次。同一个组内会发生竞争关系,只有其中一个可以消费。

接下来直接调整两个消费者为同一个组:添加如下配置

当两个消费者都设置好后启动,会发现一个问题: 实际上分到一个组对于RabbitMQ来说就是两个消费者监听了一个队列。一个队列那也就意味着,当队列收到一条消息,哪个消费者谁先消费就是谁的,消费完队列里面就没有了,也就是只有一个消费者能消费到消息!

注意:假如不设置group属性的时候,默认是启动一个消费者,就会创建一个消费队列,启动多个服务就会创建多个队列。stream默认使用的是RabbitMQ的topic交换机。当发送者向这个交换机发送消息的时候,两个队列就都会接收到。关于RabbitMQ相关知识本篇不记录,后续会专门写RabbitMQ相关文章。

最终测试:8802/8803实现了轮询分组,每次只有一个消费者8801模块的发的消息只能被8802或8803其中一个接收到,这样避免了重复消费。

2.5. 消息持久化

当三个项目都启动着的时候,现在我们要做几件事:

停止8802和8803并去除掉8802的分组group: gxs,8803不去分组信息,停止掉项目的时候会发现消息队列并没有删除,说明一旦设置分组信息,消息队列就不再是临时队列。

2.8801发送4条消息启动8802然后消息并没有打印,没有收到消息(注意8802是去掉分组信息的)再启动8803,有分组属性配置,后台打出来了MQ上的消息

原因就是:当两个项目都停止的时候,队列并未删除,而8803还绑定了这个队列,所以他就算宕机了,又重启了,依然可以收到消息。而8802没有设置分组信息,他再启动后系统会给他创建一个临时队列,自然而然收不到之前的消息了。

三、函数式编程练习

官网介绍:https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html#spring_cloud_function

加载全部内容

相关教程
猜你喜欢
用户评论