springcloud(十二)--SpringCloud Stream消息驱动

SpringCloud Stream消息总线

一、消息驱动概述

官网: https://spring.io/projects/spring-cloud-stream#overview
文档: https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.1.RELEASE/reference/html/
SpringCloud Stream中文知道手册:https://m.wang1314.com/doc/webapp/topic/20971999.html

1、消息驱动是什么

在实际应用中有很多消息中间件,比如现在企业里常用的有ActiveMQ、RabbitMQ、RocketMQ、Kafka等,学习所有这些消息中间件无疑需要大量时间经历成本,那有没有一种技术,使我们不再需要关注具体的消息中间件的细节,而只需要用一种适配绑定的方式,自动的在各种消息中间件内切换呢?消息驱动就是这样的技术,它能屏蔽底层消息中间件的差异,降低切换成本,统一消息的编程模型

什么是SpringCloud Stream
官方定义 SpringCloud Stream是一个构件消息驱动微服务的框架。
通过 inputs 和 outputs 来与 Spring Cloud Stream中的binder(绑定器)对象交互。
通过配置来binding(绑定),而 Spring Cloud Stream 的binder对象负责与消息中间件交互。
所以,我们只需要搞清楚如何与 Spring Cloud Stream交互就可以方便使用消息驱动的方式。

通话Spring Integration来连接消息代理中间件以实现消息事件驱动。
Spring Cloud Stream 为一些供应商的消息中间件产品通过了个性化的自动化配置,引用了发布-订阅、消费组、分区的三个核心概念。

但是截至到目前时间,SpringCloud Stream目前仅支持RabbitMQ和Kafka

在这里插入图片描述

2、设计思想

标准MQ

在这里插入图片描述

  • 生产者/消费者之间靠消息媒介传递信息内容(Message)
  • 消息必须走特定的通道(消息通道MessageChannel)
  • 消息通道里的消息如何被消费呢,谁负责收发处理(消息通道MessageChannel的子接口SubscribableChannel,由MessageHandler消息处理器所订阅)

为什么用Cloud Stream
比方说我们使用到了RibbitMQ和Kafka,由于这两个消息中间件的架构上的不同,
像RibbitMQ有exchange,kafka有Topic和Paritions分区,

在这里插入图片描述

这些中间件的差异性导致我们实际项目开发给我们造成一定的困扰,我们如果用了两个消息队列的其中一种,后面的业务需求,我们想另外一种消息队列进行迁移,这时候无疑就是一个灾难性的,一大堆东西都要重新推到重新做,因为它跟我们的系统耦合了,这时候springcloud stream给我们提供了一种解耦合的方式。

在没有绑定器这个概念的情况下,我们的SpringBoot应用要直接与中间件进行信息交互的时候,
由于个消息中间件构建的初衷不同,它们的实现细节上会有较大的差异性
通过定义绑定器作为中间层,完美实现了应用程序与消息中间件细节之间的隔离
通过向应用程序暴露统一的Channel通道,使得应用程序不需要再考虑各种不同的消息中间件实现。

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

在这里插入图片描述

Binder(绑定器):

  • input对应于消费者
  • output对应于生产者
    可以做到代码层面对中间件的无感知,甚至于动态的切换中间件(rabbitmq切换为kafka),使得微服务开发的高速解耦,服务可以关注更多自己的业务流程

在这里插入图片描述


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

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

3、Spring Cloud Stream标准流程

在这里插入图片描述


在这里插入图片描述

  • Binder (连接中间件,屏蔽差异)
  • Channel(通道,是队列Queue的一种抽象,在消息通讯系统中就是实现存储和转发的媒介,通过Channel对队列进行配置)
  • Source和Sink(简单的可理解为参照对象是Spring Cloud Stream自身,从Stream发布消息就是输出,接受消息就是输入)

4、编码API和常用注解

组成说明
Middleware中间件,目前只支持RabbitMQ和Kaf
BinderBinder是应用与消息中间件之间的封装,目前实行了RabbitMQ和Kafka的Binder,通过Binder可以很方便的连接中间件,可以动态的改变消息类型(对应于Kafka的topic,RabbitMQ的exchange),这些都可以通过配置文件来实现
@Input注解标识输入通道,通过该输入通道接收到的消息进入应用程序
@Output注解标识输出通道,发布的消息将通过该通道离开应用程序
@StreamListner监听队列,用于消费者的队列的消息接收
@EnableBinding使信道Channel和交换机/主题(Exchange/Topic)绑定在一起

二、案列说明

  • 本地安装RabbitMQ环境OK
  • 工程中新建三个子模块
模块名称模块功能
stream-rabbitmq-provider-8801生产者,发送消息模块
stream-rabbitmq-consumer-8802消费者,接受消息模块
stream-rabbitmq-consumer-8803消费者,接受消息模块

三、消息驱动之生产者

新建Module:stream-rabbitmq-provider-8801作为消息的生产者用来发送消息
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>
        <!-- 引入自定义的API通用包 实体类 -->
        <dependency>
            <groupId>com.zzp.springcloud</groupId>
            <artifactId>cloud-api-commons</artifactId>
            <version>${project.version}</version>
        </dependency>
        <!--web actuator-->
        <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>

application.yml配置文件:

server:
  port: 8801

spring:
  application:
    name: 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: zzpStudyExchange # 表示要使用的Exchange名称定义
          content-type: application/json # 设置消息类型,本次为json,文本则设置“text/plain”
          binder: defaultRabbit  # 设置要绑定的消息服务的具体设置

eureka:
  client: # 客户端进行Eureka注册的配置
    service-url:
      defaultZone: http://eureka7001.com:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔(默认是30秒)
    lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔(默认是90秒)
    instance-id: send-8801.com  # 在信息列表时显示主机名称
    prefer-ip-address: true     # 访问的路径变为IP地址

启动类:

@SpringBootApplication
public class StreamMQProviderMain8801{

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

业务类:

package com.zzp.springcloud.service;

/**
 * @author zzp
 * @create
 */
public interface IMessageProvider {

    //发送消息
    String send();
}

这里不需要添加@Service注解,这个业务是需要跟MQ交互的,不再是传统的业务。添加@EnableBinding注解 用来绑定消息的发送管道,消息生产者绑定的消息推送管道为org.springframework.cloud.stream.messaging.Source:(源)
实现类:

package com.zzp.springcloud.service.impl;

import com.zzp.springcloud.service.IMessageProvider;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;

import java.util.UUID;

@Slf4j
@EnableBinding(Source.class) //定义消息的推送管道
public class IMessageProviderImpl implements IMessageProvider {

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

    @Override
    public String send() {
        String serial = UUID.randomUUID().toString().replace("-","");
        output.send(MessageBuilder.withPayload(serial).build());
        log.info(">>>>>>>>>>>>serial: " + serial);
        return serial;
    }
}

MessageBuilder.withPayload(serial).build()消息构建器–参考文档

在这里插入图片描述

控制类:

@RestController
public class SendMessageController {

    @Autowired
    private IMessageProvider messageProvider;

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

启动7001,8801测试:http://localhost:8801/sendMessage

在这里插入图片描述


查看日记:

在这里插入图片描述


查看RabbitMQ控制面板中的交换机上多出了一个名为zzpStudyExchange的交换机,这就是我们yml配置的定义的Exchange名称

在这里插入图片描述


在RabbitMQ的控制面板中我们也看到了确实发送了消息

在这里插入图片描述

四、消息驱动之消费者

新建Module:stream-rabbitmq-consumer-8802/8803作为消息的生产者用来接收消息
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>
        <!-- 引入自定义的API通用包 实体类 -->
        <dependency>
            <groupId>com.zzp.springcloud</groupId>
            <artifactId>cloud-api-commons</artifactId>
            <version>${project.version}</version>
        </dependency>
        <!--web actuator-->
        <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>

yml配置文件:

server:
  port: 8802

spring:
  application:
    name: 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: zzpStudyExchange # 表示要使用的Exchange名称定义
          content-type: application/json # 设置消息类型,本次为json,文本则设置“text/plain”
          binder: defaultRabbit  # 设置要绑定的消息服务的具体设置

eureka:
  client: # 客户端进行Eureka注册的配置
    service-url:
      defaultZone: http://eureka7001.com:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔(默认是30秒)
    lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔(默认是90秒)
    instance-id: receiver-8802.com  # 在信息列表时显示主机名称
    prefer-ip-address: true     # 访问的路径变为IP地址

其中:input表示消息的消费者
8801的yml文件中:output表示消息的生产者

 bindings: # 服务的整合处理
        output: # 这个名字是一个通道的名称

启动类:

@SpringBootApplication
public class StreamMQConsumer8802 {

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

业务类:
编写消息消费者的业务类,消费者也是消费mq中的消息,添加@EnableBinding注解用来绑定消息的推送管道,消息消费者绑定的消息推送管道为import org.springframework.cloud.stream.messaging.Sink,在接收消息的方法中需要使用注解@StreamListner来监听其绑定的消息推送管道:

package com.zzp.springcloud.listener;

import lombok.extern.slf4j.Slf4j;
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;

/**
 * @author zzp
 * @create
 */
@Component
@EnableBinding(Sink.class) // 定义消息的接受
@Slf4j
public class ReceiverMessageListener {

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

    @StreamListener(Sink.INPUT)
    public void input(Message<String> message) {
        log.info("serverPort: "+serverPort+ "   消费者1号,接受:"+message.getPayload());
    }

}

启动8802/8803,然后在8801生成者发送消息,查看8802/8803消费者微服务的日记:

在这里插入图片描述

在这里插入图片描述


在这里插入图片描述


查看RabbitMQ的控制面板中exchanges

在这里插入图片描述


出现波峰,说明这个交互机有消费

在这里插入图片描述


交换机有两台机器(默认分组)

在这里插入图片描述

五、分组消费与持久化

1、重复消费问题

当生产者发送消息后,此时的我们的消费者都接受了消息并进行了消费,也就是说同一条消息被多个消息消费者所消费
生产者推送三条消息:

在这里插入图片描述


两台消费者都收到三条消息:

在这里插入图片描述


在这里插入图片描述


上述的问题就是消息的重复消费问题,那么这个问题为什么如此重要呢?其实重复消费这个问题本身不可怕,可怕的是没考虑到重复消费之后,怎么保证幂等性。(幂等性,通俗的说,就一个数据,或者一个请求,重复很多次,需要确保对应的数据是不会改变的,不能出错)。分布式微服务应用为了实现高可用和负载均衡,实际上同一功能的服务都会部署多个具体的服务实例。举个例子,假设有一个系统,有一条消息要求往数据库里插入一条数据,要是这个消息重复消费两次,结果就是向数据库里插入了两条数据,这样数据就错了,就违背了幂等性原则,但是要是该消息消费到第二次的时候,可以判断一下该消息已经消费过了,然后直接将该消息丢弃,这就实现了只插入一条数据,一条消息重复出现了两次,但是只有第一次真正被消费了,数据库里也就只插入了一条数据,这就保证了系统的幂等性。

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

在RabbitMQ中,查看zzpStudyExchange交换机,可以发现8802和8803两个消息消费者处于不同的组,所以8801消息生产者发送的消息可以被这两个消费者重复消费:

在这里插入图片描述

2、分组解决重复消费问题

面在RabbitMQ控制面板中我们看到的组流水号是系统随机分配的,这样无疑不好控制,所以我们应该自定义配置分组,将8802/8803两个消息消费者微服务分为同一个组,以此来解决消息的重复消费问题,先来演示如何自定义分组。
8802/8803消费者微服务的yml配置文件中分别添加组名属性(不同的组名):

group: zzpA/B # 分组名称

在这里插入图片描述


先在RabbitMQ中控制面板中,查看zzpStudyExchange交换机;

在这里插入图片描述


此时由于8802/8803位于两个不同分组下,所以没有竞争关系,消息生产者发送消息后,仍然可以重复消费。

在这里插入图片描述


在这里插入图片描述


在这里插入图片描述

下面我们将这两个消息消费方微服务分到相同的消费组中,这样每次就只有一个消费者,消息生产者发送的消息只能被8802或8803其中一个接受到,这样就避免了重复消费,将8802和8803的分组名都改为zzpA,再次重启两个消息消费方微服务,此时我们可以看到在分组A下已经有了两个消费者

在这里插入图片描述

生产者推送四条消息:

在这里插入图片描述


消费者服务接收消息:我们发现8802/8803分别消费了各2条不同的消息,而没有出现重复消费的问题:

在这里插入图片描述


在这里插入图片描述

3、持久化

我们把8803消费微服务的yml文件配置分组名称去掉

在这里插入图片描述


停掉8802和8803消费者微服务:

在这里插入图片描述

在8801生产者服务推送四条消息:

在这里插入图片描述

启动8803消费者服务:

在这里插入图片描述

8803启动后,日记没有打印消费日记,说明8803并没有去mq上拿未曾消费的消息

启动8802消费者服务:

在这里插入图片描述


8802服务消费了从8801生产者服务推送四条消息

Stream的group分组属性,在我们的消息重复消费和消息持久化消费、避免消息丢失,是一个非常重要的属性。

原文地址:https://blog.csdn.net/weixin_42469070/article/details/114375274

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。

相关推荐


Nacos 中的参数有很多,如:命名空间、分组名、服务名、保护阈值、服务路由类型、临时实例等,那这些参数都是什么意思?又该如何设置?接下来我们一起来盘它。 1.命名空间 在 Nacos 中通过命名空间(Namespace)+ 分组(Group)+服务名(Name)可以定位到一个唯一的服务实例。 命名
Nacos 支持两种 HTTP 服务请求,一个是 REST Template,另一个是 Feign Client。之前的文章咱们介绍过 Rest Template 的调用方式,主要是通过 Ribbon(负载均衡) + RestTemplate 实现 HTTP 服务调用的,请求的核心代码是这样的: @
Nacos 是 Spring Cloud Alibaba 中一个重要的组成部分,它提供了两个重要的功能:服务注册与发现和统一的配置中心功能。 服务注册与发现功能解决了微服务集群中,调用者和服务提供者连接管理和请求转发的功能,让程序的开发者无需过多的关注服务提供者的稳定性和健康程度以及调用地址,因为这
Spring Cloud Alibaba 是阿里巴巴提供的一站式微服务开发解决方案,目前已被 Spring Cloud 官方收录。而 Nacos 作为 Spring Cloud Alibaba 的核心组件之一,提供了两个非常重要的功能:服务注册中心(服务注册和发现)功能,和统一配置中心功能。 Nac
在 Nacos 的路由策略中有 3 个比较重要的内容:权重、保护阈值和就近访问。因为这 3 个内容都是彼此独立的,所以今天我们就单独拎出“保护阈值”来详细聊聊。 保护阈值 保护阈值(ProtectThreshold):为了防止因过多实例故障,导致所有流量全部流入剩余健康实例,继而造成流量压力将剩余健
前两天遇到了一个问题,Nacos 中的永久服务删除不了,折腾了一番,最后还是顺利解决了。以下是原因分析和解决方案,建议先收藏,以备不时之需。 临时实例和持久化实例是 Nacos 1.0.0 中新增了一个特性。临时实例和持久化实例最大的区别是健康检查的方式:临时实例使用客户端主动上报的健康检查模式,而
Spring Cloud Alibaba 技术体系中的 Nacos,提供了两个重要的功能:注册中心(服务注册与发现)功能和配置中心功能。 其中注册中心解决了微服务调用中,服务提供者和服务调用者的解耦,让程序开发者可以无需过多的关注服务提供者和调用者的运行细节,只需要通过 Nacos 的注册中心就可以
负载均衡通器常有两种实现手段,一种是服务端负载均衡器,另一种是客户端负载均衡器,而我们今天的主角 Ribbon 就属于后者——客户端负载均衡器。 服务端负载均衡器的问题是,它提供了更强的流量控制权,但无法满足不同的消费者希望使用不同负载均衡策略的需求,而使用不同负载均衡策略的场景确实是存在的,所以客
本篇文章为大家展示了如何解决Spring Cloud 服务冲突问题,内容简明扼要并且容易理解,绝对能使你眼前一亮,通过这篇文章的详细介绍希望你能有所收获。一、背景...
本篇内容主要讲解“spring cloud服务的注册与发现怎么实现”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“spri...
本篇内容介绍了“Dubbo怎么实现Spring Cloud服务治理 ”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处...
本篇内容主要讲解“SpringCloud相关面试题有哪些”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“SpringCloud相...
如何分析Spring Cloud Ribbon、Spring Cloud Feign以及断路器,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希
这篇文章主要讲解了“springcloud微服务的组成部分有哪些”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“s...
这篇文章主要讲解了“SpringCloud的OpenFeign项目怎么创建”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习...
本篇内容主要讲解“spring cloud oauth3整合JWT后获取用户信息不全怎么办”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带...
怎样解析微服务架构SpringCloud,针对这个问题,这篇文章详细介绍了相对应的分析和解答,希望可以帮助更多想解决这个问题的小伙伴找到更简单易行的方法。...
这篇文章主要介绍spring cloud中API网关的示例分析,文中介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们一定要看完!一、服务网关简介1、外观模式客户端...
本篇内容介绍了“Spring Cloud微服务的相关问题有哪些”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处...
本文小编为大家详细介绍“spring cloud config整合gitlab如何搭建分布式的配置中心”,内容详细,步骤清晰,细节处理妥当,希望这篇“spring cloud config整合gi...