第十一章 Stream消息驱动

Stream消息驱动

gitee:springcloud_study: springcloud：服务集群、注册中心、配置中心（热更新）、服务网关（校验、路由、负载均衡）、分布式缓存、分布式搜索、消息队列（异步通信）、数据库集群、分布式日志、系统监控链路追踪。

1. 消息驱动概述

作用：屏蔽底层消息中间件的差异,降低切换成本，统—消息的编程模型。底层不管是什么中间件如kafka、rabbitmq，Stream可以解决不同中间件的通信。 官网：Spring Cloud Stream

官方定义Spring Cloud Stream是一个构建消息驱动微服务的框架。
​
应用程序通过 inputs 或者 outputsj来与Spring Cloud Stream中binder对象交互。通过我们配置来binding(绑定)，而Spring Cloud Stream的 binder对象负责与消息中间件交互。所以，我们只需要搞清楚如何与Spring Cloud Stream交互就可以方便使用消息驱动的方式。
​
通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动。
Spring Cloud Stream为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区的三个核心概念。
​
但是Stream只支持kafka、rabbitmq。

img

设计思想 标准的MQ：

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

Cloud Stream：

Stream利用Binder来绑定中间件的输入流和输出流。如果系统使用到了两个中间件（kafka、rabbitmq）：这些中间件的差异性导致我们实际项目开发给我们造成了一定的困扰，我们如果用了两个消息队列的其中一种，后面的业务需求，我想往另外一种消息队列进行迁移，这时候无疑就是一个灾难性的人—大堆东西都要重新推倒重新做，因为它跟我们的系统耦合了，这时候springcloud Stream给我们提供了一种解耦合的方式。

Stream中的消息通信方式遵循了发布-订阅模式：

Topic在Rabbitmq中是Exchange、在kafka中是Topic。

Spring Cloud Stream标准流程套路

img

Middleware：中间件，目前只支持RabbitMQ和Kafka
Binder：是应用与消息中间件之间的封装，目前实行了Kafka和RabbitMQ的Binder，通过Binder可以很方便的连接中间件，可以动态的改变消息类型(对应于Kafka的topic,RabbitMQ的exchange)，这些都可以通过配置文件来实现。
@Input：注解标识输入通道，通过该输入通道接收到的消息进入应用程序
@Output：注解标识输出通道，发布的消息将通过该通道离开应用程序
@StreamListener：监听队列。用于消费者的队列的消息接收
@EnableBinding：指信道channel和exchange绑定在一起

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

2. 消息驱动之生产者

创建cloud-stream-rabbitmq-provider8801：作为生产者进行发消息模块

  1. pom文件

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

  1. application.yaml

server:port: 8801
spring:application:name: cloud-stream-providercloud:stream:binders: #在此处配置要绑定的rabbitmq的服务信息;defaultRabbit:type: rabbitenvironment:spring:rabbitmq:host: 192.168.25.153port: 5672username: adminpassword: aaaaaabindings: #服务的整合处理output: #destination: studyExchange  #表示要使用的Exchange名称定义content-type: application/json #设置消息类型，本次为json,文本则设置“text/plain”binder: defaultRabbit #设置要绑定的消.息服务的具体设置
eureka:client:service-url:defaultZone: http://eureka7001.com:7001/eureka,http://eureka7002.com:7002/eurekaregister-with-eureka: truefetch-registry: trueinstance:lease-renewal-interval-in-seconds: 2lease-expiration-duration-in-seconds: 5instance-id: send-8801.comprefer-ip-address: true

  1. 主启动类

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

  1. service层

public interface IMessageProvider {String send();
}
@EnableBinding(Source.class)  //定义消息的推送管道
public class MessageProviderImpl implements IMessageProvider {
​@Resourceprivate MessageChannel output;  //消息发送管道
​@Overridepublic String send() {String serial = UUID.randomUUID().toString();output.send(MessageBuilder.withPayload(serial).build());System.out.println("********serial："+serial);return null;}
}

  1. controller层

@RestController
public class SendMessageController {
​@Resourceprivate IMessageProvider messageProvider;
​@GetMapping(value = "/sendMessage")public String sendMessage(){return messageProvider.send();}
}

测试：

3. 消息驱动之消费者

创建cloud-stream-rabbitmq-consumer8802，作为消息接收模块

  1. pom文件

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

  1. application.yml

server:port: 8802
spring:application:name: cloud-stream-consumercloud:stream:binders: #在此处配置要绑定的rabbitmq的服务信息;defaultRabbit:type: rabbitenvironment:spring:rabbitmq:host: 192.168.25.153port: 5672username: adminpassword: aaaaaabindings: #服务的整合处理input: #destination: studyExchange  #表示要使用的Exchange名称定义content-type: application/json #设置消息类型，本次为json,文本则设置“text/plain”binder: defaultRabbit #设置要绑定的消.息服务的具体设置
eureka:client:service-url:defaultZone: http://eureka7001.com:7001/eureka,http://eureka7002.com:7002/eurekaregister-with-eureka: truefetch-registry: trueinstance:lease-renewal-interval-in-seconds: 2lease-expiration-duration-in-seconds: 5instance-id: receive-8802.comprefer-ip-address: true

  1. controller层

@Component
@EnableBinding(Sink.class)
public class ReceiveMessageController {
​@Value("${server.port}")private String serverPort;
​@StreamListener(Sink.INPUT)public void input(Message<String> message){System.out.println("消费者1号，------>接收到的消息："+message.getPayload()+"\t port："+serverPort);}
}

  1. 主启动类

@SpringBootApplication
@EnableEurekaClient
public class ConsumerMQMain8802 {public static void main(String[] args) {SpringApplication.run(ConsumerMQMain8802.class,args);}
}

测试：

启动loccalhost:8801/sendMessage就可以了，消费者就是一个监听器，有message就消费。

4. 分组消费与持久化

根据cloud-stream-rabbitmq-consumer8802创建8803项目，运行暴露问题：


消息重复消费和消息持久化问题，需要进行分组操作。注意在Stream中处于同一个group中的多个消费者是竞争关系，就能够保证消息只会被其中一个应用消费一次。不同组是可以全面消费的(重复消费)。

解决重复消费方法：加入同一个组（下图是不同分组的情况）

cloud-stream-rabbitmq-consumer8802和8803设置不同分组yicaiA/B

server:port: 8803
spring:application:name: cloud-stream-consumercloud:stream:binders: #在此处配置要绑定的rabbitmq的服务信息;defaultRabbit:type: rabbitenvironment:spring:rabbitmq:host: 192.168.25.153port: 5672username: adminpassword: aaaaaabindings: #服务的整合处理input: #destination: studyExchange  #表示要使用的Exchange名称定义content-type: application/json #设置消息类型，本次为json,文本则设置“text/plain”binder: defaultRabbit #设置要绑定的消.息服务的具体设置group: yicaiB
server:port: 8802
spring:application:name: cloud-stream-consumercloud:stream:binders: #在此处配置要绑定的rabbitmq的服务信息;defaultRabbit:type: rabbitenvironment:spring:rabbitmq:host: 192.168.25.153port: 5672username: adminpassword: aaaaaabindings: #服务的整合处理input: #destination: studyExchange  #表示要使用的Exchange名称定义content-type: application/json #设置消息类型，本次为json,文本则设置“text/plain”binder: defaultRabbit #设置要绑定的消.息服务的具体设置group: yicaiA

cloud-stream-rabbitmq-consumer8802和8803设置同一个组yicaiA

server:port: 8802
spring:application:name: cloud-stream-consumercloud:stream:binders: #在此处配置要绑定的rabbitmq的服务信息;defaultRabbit:type: rabbitenvironment:spring:rabbitmq:host: 192.168.25.153port: 5672username: adminpassword: aaaaaabindings: #服务的整合处理input: #destination: studyExchange  #表示要使用的Exchange名称定义content-type: application/json #设置消息类型，本次为json,文本则设置“text/plain”binder: defaultRabbit #设置要绑定的消.息服务的具体设置group: yicaiA

测试：

持久化 加上group就算实现类持久化。所谓的持久化就是如果没有分组，一个服务发送消息，其他服务由于没有分组，如果其他哪些服务断开，又继续重启，这样就会导致以前那些消息丢失。

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

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

相关文章

maven命令行安装依赖测试

maven命令行安装依赖测试

mvn dependency:get -DgroupIdorg.springframework -DartifactIdspring-core -Dversion5.3.9作用&#xff1a;可用于测试配置环境变量后&#xff0c;能否下载依赖到本地仓库
阅读更多...
前后端分离架构的特点以及优缺点

前后端分离架构的特点以及优缺点

文章目录 一、前后端不分离架构(传统单体结构)1.1 什么是前后端不分离1.2 工作原理1.3 前后端不分离的优缺点1.4 应用场景 二、前后端分离架构2.1 为什么要前后端分离2.2 什么是前后端分离2.3 工作原理2.4 前后端分离的优缺点 参考资料 一、前后端不分离架构(传统单体结构) 首…
阅读更多...
【linux】cut的基本使用

【linux】cut的基本使用

cut主要用于按列切分文本行&#xff0c;并输出指定的字段&#xff0c;这是类unix系统中常用的文本处理工具。 基本使用 首先随便去网上找个文本或者列表文件 如果使用cat看文本的话就是这样的 sh-3.2# cat data.csv Name,Age,City,Salary Alice,30,New York,70000 Bob,25,L…
阅读更多...
C++ 侯捷 内存管理

C++ 侯捷 内存管理

C 的内存获取机制&#xff1a; void* p1 malloc(512); free(p1);complex<int>* p2 new complex<int>; delete p2;void* p3 ::operator new(512); ::operator delete(p3);//GNUC void* p4 alloc::allocate(512); alloc::deallocate(p4, 512);//GNUC4.9 void* p5…
阅读更多...
Hbase详解

Hbase详解

Hbase 概念 base 是分布式、面向列的开源数据库&#xff08;其实准确的说是面向列族&#xff09;。HDFS 为 Hbase 提供可靠的底层数据存储服务&#xff0c;MapReduce 为 Hbase 提供高性能的计算能力&#xff0c;Zookeeper 为 Hbase 提供稳定服务和 Failover 机制&#xff0c;…
阅读更多...
OR-NeRF论文笔记

OR-NeRF论文笔记

OR-NeRF论文笔记 文章目录 OR-NeRF论文笔记论文概述Abstract1 Introduction2 Related Work3 Background4 Method4.1 Multiview Segmentation4.2 Scene Object Removal 5 ExperimentsDatasetsMetricsMultiview SegmentationScene Object Removal 6 Conclusion 论文概述 目的&am…
阅读更多...
【软件工程】漫谈增量过程模型：软件开发的逐步之道

【软件工程】漫谈增量过程模型：软件开发的逐步之道

&#x1f34e;个人博客&#xff1a;个人主页 &#x1f3c6;个人专栏&#xff1a; 软件工程 ⛳️ 功不唐捐&#xff0c;玉汝于成 目录 前言&#xff1a; 正文 增量过程模型&#xff08;Incremental Process Model&#xff09; 主要特点和阶段&#xff1a; 优点&#xff1…
阅读更多...
TCP 协议为什么要设计三次握手 TCP 协议，是一种可靠的，基于字节流的，面向连接的传输层协议。

TCP 协议为什么要设计三次握手 TCP 协议，是一种可靠的，基于字节流的，面向连接的传输层协议。

文章目录 TCP 协议为什么要设计三次握手TCP 协议&#xff0c;是一种可靠的&#xff0c;基于字节流的&#xff0c;面向连接的传输层协议。TCP 通信双方的数据传输是通过字节流来实现传输的客户端向服务端发送连接请求并携带同步序列号 SYN。 今天我们来谈谈tcp的三次握手 TCP 协…
阅读更多...
Vue2+element-ui 实现select选择器结合Tree树形控件实现下拉树效果

Vue2+element-ui 实现select选择器结合Tree树形控件实现下拉树效果

效果&#xff1a; DOM部分 &#xff1a; // 设置el-option隐藏的下拉选项&#xff0c;选项显示的是汉字label&#xff0c;值是value // 如果不设置一个下拉选项&#xff0c;下面的树形组件将无法正常使用 <el-form-item label"报警区域" prop"monitorId"…
阅读更多...
如何让python在手机上运行,python程序在手机上运行

如何让python在手机上运行,python程序在手机上运行

大家好&#xff0c;给大家分享一下python怎么在手机上运行爱心代码&#xff0c;很多人还不知道这一点。下面详细解释一下。现在让我们来看看&#xff01; 1. 写在前面的话 天天都在PC端运行Python代码的我&#xff0c;今天突然灵光一现&#xff0c;想着是不是能够在移动端运行P…
阅读更多...
CSS 缩减顶部

CSS 缩减顶部

<template><!-- mouseenter"startAnimation" 表示在鼠标进入元素时触发 startAnimation 方法。mouseleave"stopAnimation" 表示在鼠标离开元素时触发 stopAnimation 方法。 --><!-- 容器元素 --><div class"container" mou…
阅读更多...
通用搜索的工作原理

通用搜索的工作原理

了解 Google 的通用搜索结果为何如此重要&#xff0c;通用搜索的发展方向&#xff0c;以及它对您意味着什么。 让我们从回答一个显而易见的问题开始&#xff1a; 什么是通用搜索&#xff1f; 网络上有一些通用搜索的定义&#xff0c;但我更喜欢从马的嘴里听到这样的事情。 …
阅读更多...
与擎创科技共建一体化“数智”运维体系，实现数字化转型

与擎创科技共建一体化“数智”运维体系，实现数字化转型

小窗滴滴小编获取最新版公司简介 前言&#xff1a; 哈喽大家好&#xff0c;最近分享的互联网IT热讯大家都挺喜欢&#xff0c;小编看着数据着实开心&#xff0c;感谢大家支持&#xff0c;小编会继续给大家推送。 新岁即将启封&#xff0c;我们一年一期的运维干货年末大讲也要…
阅读更多...
【halcon深度学习】dev_display_dl_data 移植到C# 上篇

【halcon深度学习】dev_display_dl_data 移植到C# 上篇

效果展示 前言 在研究halcon深度学习的时候，会发现halcon的例程里面用到了大量的二次封装库函数。这些库函数内部也是由基础的算子组成。我们在halcon的开发环境里面用的很爽，但是一旦要在C#中使用，就会报错。 一开始，我想避开这个移植过程，直接使用halcon引擎(HDevEngi…
阅读更多...
深入Mybatis数据源

深入Mybatis数据源

数据源是持久层框架中最核心的组件之一&#xff0c;在实际工作中比较常见的数据源有 C3P0、Apache Common DBCP、Proxool 等。作为一款成熟的持久化框架&#xff0c;MyBatis 不仅自己提供了一套数据源实现&#xff0c;而且还能够方便地集成第三方数据源。 javax.sql.DataSourc…
阅读更多...
C++系列-第1章顺序结构-3-输出类cout

C++系列-第1章顺序结构-3-输出类cout

C系列-第1章顺序结构-3-输出类cout 在线练习&#xff1a; http://noi.openjudge.cn/ https://www.luogu.com.cn/ 总结 本文是C系列博客&#xff0c;主要讲述输出类cout的用法 cout介绍与基本用法 在C中&#xff0c;cout 是用于输出&#xff08;打印&#xff09;数据的工具&…
阅读更多...
文华财经鬼谷数据期货量化分析系统指标公式

文华财经鬼谷数据期货量化分析系统指标公式

鬼谷数据量化主图源码 安装流程——打开文华财经软件——右上角指标管理器——新建指标——复制源码——安装成功 鬼谷数据主图源码 安装流程——打开文华财经软件——右上角指标管理器——新建指标——复制源码——安装成功文华财经期货软件擒龙量化主图指标公式HH:HHV(HIGH…
阅读更多...
系列四、Eureka自我保护

系列四、Eureka自我保护

一、Eureka自我保护 1.1、故障现象 保护模式主要用于一组客户端和Eureka Server之间存在网络分区场景下的保护。一旦进入保护模式&#xff0c;Eureka Server将会尝试保护其服务注册表中的信息&#xff0c;不再删除服务注册表中的数据&#xff0c;也就是不会注销任何微服务。如…
阅读更多...
交叉编译aarch64架构支持openssl的curl、libcurl

交叉编译aarch64架构支持openssl的curl、libcurl

本文档旨在指导读者在x86_64平台上交叉编译curl和openssl库以支持aarch64架构。在开始之前&#xff0c;请确保您的系统环境已正确配置。 1. 系统环境准备 系统是基于Ubuntu 20.04 LTS&#xff0c;高版本可能会有问题。首页&#xff0c;安装必要的开发工具和库文件。打开终端并…
阅读更多...
【C++拷贝构造函数】动态分配与成员对象

【C++拷贝构造函数】动态分配与成员对象

系列文章目录 &#x1f308;座右铭&#x1f308;&#xff1a;人的一生这么长、你凭什么用短短的几年去衡量自己的一生&#xff01; &#x1f495;个人主页:清灵白羽 漾情天殇_计算机底层原理,深度解析C,自顶向下看Java-CSDN博客 目录 系列文章目录 一、拷贝构造函数是什么&am…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023