【Springcloud篇】学习笔记七（十三章）：Stream消息驱动

第十三章_Stream消息驱动

image-20230923112030279

1.消息驱动Stream介绍

1.1Stream为什么被引入

常见MQ(消息中间件)：

  • ActiveMQ
  • RabbitMQ
  • RocketMQ
  • Kafka

有没有一种新的技术诞生，让我们不再关注具体MQ的细节，我们只需要用一种适配绑定的方式，自动的给我们在各种MQ内切换。（类似于Hibernate）

1.2是什么

屏蔽底层消息中间件的差异,降低切换成本，统一消息的编程模型。

官方定义Spring Cloud Stream是一个构建消息驱动微服务的框架。

应用程序通过inputs或者 outputs 来与Spring Cloud Stream中binder对象交互。通过我们配置来binding(绑定)，而Spring Cloud Stream 的binder对象负责与消息中间件交互。所以，我们只需要搞清楚如何与Spring Cloud Stream交互就可以方便使用消息驱动的方式。

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

目前仅支持RabbitMQ、 Kafka。

1.3设计思想

（1）标准MQ

image-20230923145029250

  • 生产者/消费者之间靠消息媒介传递信息内容
  • 消息必须走特定的通道 - 消息通道 Message Channel
  • 消息通道里的消息如何被消费呢，谁负责收发处理 - 消息通道MessageChannel的子接口SubscribableChannel，由MessageHandler消息处理器所订阅。
（2）为什么用Cloud Stream

比方说我们用到了RabbitMQ和Kafka，由于这两个消息中间件的架构上的不同，像RabbitMQ有exchange，kafka有Topic和Partitions分区。

image-20230923145056628

我们如果用了两个消息队列的其中一种，后面的业务需求，我想往另外一种消息队列进行迁移，这时候无疑就是一个灾难性的，一大堆东西都要重新推倒重新做，因为它跟我们的系统耦合了，这时候Spring Cloud Stream给我们提供了—种解耦合的方式。

（3）Stream凭什么可以统一底层差异

在没有绑定器这个概念的情况下，我们的SpringBoot应用要直接与消息中间件进行信息交互的时候，由于各消息中间件构建的初衷不同，它们的实现细节上会有较大的差异性。

通过定义绑定器作为中间层，完美地实现了应用程序与消息中间件细节之间的隔离。通过向应用程序暴露统一的Channel通道，使得应用程序不需要再考虑各种不同的消息中间件实现。

一句话：屏蔽底层消息中间件的差异，降低切换成本，统一消息的编程模型。

Stream对消息中间件的进一步封装，可以做到代码层面对中间件的无感知，甚至于动态的切换中间件(rabbitmq切换为kafka)，使得微服务开发的高度解耦，服务可以关注更多自己的业务流程

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

（4）Binder

Binder可以生成Binding，Binding用来绑定消息容器的生产者和消费者，它有两种类型

  • INPUT对应于消息消费者
  • OUTPUT对应于消息生产者

下图为SpringCloud Stream的处理架构

image-20230923145237564

（5）Stream中的消息通信方式遵循发布—订阅模式

在消息中间件系统中，主题（Topic）是一种消息发布和订阅模式，允许发布者将消息发送到一个主题，而订阅者可以选择订阅他们感兴趣的主题，以接收相关消息。RabbitMQ和Apache Kafka都支持主题的概念，但它们在如何理解和实现主题方面有一些不同。

  • RabbitMQ：

    RabbitMQ是一个消息队列系统，它使用Exchange来实现主题的概念。在RabbitMQ中，Exchange是消息的分发中心，它接收来自生产者的消息并将其路由到一个或多个队列中。

  • Kafka：

    Kafka是一个分布式事件流平台，它使用Topic来实现主题的概念。在Kafka中，一个主题是一个逻辑通道，用于组织和发布消息。Kafka的主题与传统消息队列不同，因为它的主题中的消息通常保留在日志中，供多个消费者进行订阅和消费。

1.4Stream的标准流程套路

image-20230923150053190

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

1.5编码API和常用注解

image-20230923150157519

2.案例说明

RabbitMQ环境已经OK，工程中新建三个子模块

  • cloud-stream-rabbitmq-provider8801，作为生产者进行发消息模块
  • cloud-stream-rabbitmq-consumer8802，作为消息接收模块
  • cloud-stream-rabbitmq-consumer8803 作为消息接收模块

2.1消息生产者provider8801

（1）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><dependency><!-- 引入自己定义的api通用包，可以使用Payment支付Entity --><groupId>springcloud</groupId><artifactId>cloud-api-commons</artifactId><version>${project.version}</version></dependency></dependencies>
（2）yml

image-20230923151608331

（3）主启动

image-20230923151553678

（4）业务类
  • service层

image-20230923151545950

  • controller层

image-20230923153445058

（5）测试
  • 启动Eureka7001
  • 再启动8801
  • 登录rabbit 15672，查看发送消息的主题名称

image-20230923152025096

2.2消息消费者consumer8802

cloud-stream-rabbitmq-consumer8802

（1）pom
<dependencies><!-- stream-rabbit --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId></dependency><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><dependency><!-- 引入自己定义的api通用包，可以使用Payment支付Entity --><groupId>springcloud</groupId><artifactId>cloud-api-commons</artifactId><version>${project.version}</version></dependency></dependencies>
（2）yml

image-20230923152517867

（3）主启动类

image-20230923153137374

（4）controller层

image-20230923153150565

（5）测试
  • 发送消息

image-20230923153532346

  • 8801后台

image-20230923153558127

  • 8802后台

image-20230923153614353

2.3分组消费与持久化

依照8802，clone出来一份运行8803

启动

  • 分组消费RabbitMQ
  • 7001：服务注册
  • 8801：消息生产
  • 8802：消息消费
  • 8803：消息消费

运行后有两个问题：

  • 有重复消费问题
  • 消息持久化问题，即消息消费方不能接收未曾被接收的消息
（1）消息重复消费

8001发送一条消息，发结果8802/8803同时都收到了，存在重复消费问题

生产实际案例：

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

image-20220322175234895

（2）解决消息重复消费

原理：

微服务应用放置于同一个group中，就能够保证消息只会被其中一个应用消费一次。

不同组是可以全面消费的(重复消费)，同一组内会发生竞争关系，只有其中一个可以消费。

分组

8002，8003添加到同一分组即可

application.yml中添加分组groupPeng

image-20230923161815875

  • 测试

image-20230923161756283

（3）解决消息持久化

通过分组可以实现，消息消费方重启的时候，可以接收，其在宕机的时候，消息发送方发送的消息。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/452538.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Codeforces Beta Round 8 C. Looking for Order 【状压DP】

Codeforces Beta Round 8 C. Looking for Order 【状压DP】

C. Looking for Order 题意 平面直角坐标系上有 n n n 个物品&#xff0c;还有一个初始背包位置 ( x 0 , y 0 ) (x_0, y_0) (x0​,y0​)&#xff0c;从背包位置出发&#xff0c;每次最多携带两个物品回来背包&#xff0c;求把所有物品带回背包位置要走的最短距离&#xff0c…
阅读更多...
[python]基于Ultra-Fast-Lane-Detection-v2车道线实时检测onnx部署

[python]基于Ultra-Fast-Lane-Detection-v2车道线实时检测onnx部署

【论文地址】 https://arxiv.org/pdf/2206.07389.pdf 【框架地址】 https://github.com/cfzd/Ultra-Fast-Lane-Detection-v2 【框架介绍】 Ultra-Fast-Lane-Detection-v2&#xff08;UFL-D-v2&#xff09;算法是一种高效的车道线检测算法&#xff0c;它旨在快速准确地识别…
阅读更多...
ElementUI Form：Form表单

ElementUI Form：Form表单

ElementUI安装与使用指南 Form表单 点击下载learnelementuispringboot项目源码 效果图 el-form.vue&#xff08;Form表单&#xff09;页面效果图 项目里 el-form.vue代码 <script> export default {name: el_form,data() {var checkAge (rule, value, callback…
阅读更多...
论文阅读-一种用于大规模分布式文件系统中基于深度强化学习的自适应元数据管理方案

论文阅读-一种用于大规模分布式文件系统中基于深度强化学习的自适应元数据管理方案

名称&#xff1a; An Adaptive Metadata Management Scheme Based on Deep Reinforcement Learning for Large-Scale Distributed File Systems I. 引言 如今&#xff0c;大型集群文件系统的规模已达到PB甚至EB级别&#xff0c;由此产生的数据呈指数级增长。系统架构师不断设…
阅读更多...
ai创作软件有哪些？这5个软件了解一下

ai创作软件有哪些？这5个软件了解一下

ai创作软件有哪些&#xff1f;随着科技的飞速发展&#xff0c;人工智能&#xff08;AI&#xff09;在各个领域都展现出了惊人的实力。特别是在内容创作领域&#xff0c;AI技术已经成为了助力创作者们提高效率、释放创意的得力助手。今天&#xff0c;我们将为大家介绍五款AI创作…
阅读更多...
指针的学习2

指针的学习2

目录 数组名的理解 使用指针访问数组 一维数组传参的本质 冒泡排序 二级指针 指针数组 指针数组模拟二维数组 数组名的理解 数组名是数组首元素的地址 例外&#xff1a; sizeof(数组名),sizeof中单独放数组名&#xff0c;这里的数组名表示整个数组&#xff0c;计算的…
阅读更多...
Mybatis Plus 结合 Mybatis X 插件快速生成CRUD代码

Mybatis Plus 结合 Mybatis X 插件快速生成CRUD代码

Mybatis Plus 结合 Mybatis X 插件快速生成CRUD代码 本文介绍在 IDEA 中如何使用 Mybatis X 插件快速生成 Mybatis Plus 的 CRUD 代码。包括实体类 model 代码、持久层 Mapper 代码和与之对应的Mapper.xml 代码、服务层 service 接口与impl接口实现类代码 1.安装Mybatis X 插…
阅读更多...
蓝桥杯---生日蜡烛

蓝桥杯---生日蜡烛

某君从某年开始每年都举办一次生日party&#xff0c;并且每次都要吹熄与年龄相同根数的蜡烛&#xff0c;现在算起来&#xff0c;他一共吹熄了236根蜡烛。请问,他从多少岁开始过生日party的? 请填写他开始过生日 party的年龄数。 注意:你提交的应该是一个整数&#xff0c;不要…
阅读更多...
【JavaScript 漫游】【006】数据类型 array

【JavaScript 漫游】【006】数据类型 array

文章简介 本文为【JavaScript 漫游】专栏的第 006 篇文章&#xff0c;记录笔者在了解 JS 数据类型 array 中摘录的知识点。 数组的本质是对象属组的 length 属性for ... in 循环和数组的遍历数组的空位类数组对象 除了上述 5 个重要知识点&#xff0c;学习数组更为重要的是掌…
阅读更多...
golang开源的可嵌入应用程序高性能的MQTT服务

golang开源的可嵌入应用程序高性能的MQTT服务

golang开源的可嵌入应用程序高性能的MQTT服务 什么是MQTT&#xff1f; MQTT&#xff08;Message Queuing Telemetry Transport&#xff09;是一种轻量级的、开放的消息传输协议&#xff0c;设计用于在低带宽、高延迟或不可靠的网络环境中进行通信。MQTT最初由IBM开发&#xf…
阅读更多...
fastjson 导致的OOM

fastjson 导致的OOM

fastjson 导致的OOM 示例代码 public static void main(String[] args) throws Exception {try {List<Integer> list JSONObject.parseArray("[2023,2024", Integer.class);}catch (Exception e){System.err.println("error");}System.out.println…
阅读更多...
【Java集合框架ArrayList、LinkedList、HashSet之间的区别】

【Java集合框架ArrayList、LinkedList、HashSet之间的区别】

Java集合框架ArrayList、LinkedList、HashSet之间的区别 1. 实现方式&#xff1a;2. 插入和删除操作的效率&#xff1a;3. 随机访问的效率&#xff1a;4. 内存占用&#xff1a;综上所述&#xff0c;选择ArrayList还是LinkedList或HashSet取决于具体的使用场景。如果需要频繁的插…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023