当前位置: 首页 > news >正文

spring-cloud-stream

系列文章目录

第一章 Java线程池技术应用
第二章 CountDownLatch和Semaphone的应用
第三章 Spring Cloud 简介
第四章 Spring Cloud Netflix 之 Eureka
第五章 Spring Cloud Netflix 之 Ribbon
第六章 Spring Cloud 之 OpenFeign
第七章 Spring Cloud 之 GateWay
第八章 Spring Cloud Netflix 之 Hystrix
第九章 代码管理gitlab 使用
第十章 SpringCloud Alibaba 之 Nacos discovery
第十一章 SpringCloud Alibaba 之 Nacos Config
第十二章 Spring Cloud Alibaba 之 Sentinel
第十三章 JWT
第十四章 RabbitMQ应用
第十五章 RabbitMQ 延迟队列
第十六章 spring-cloud-stream

在这里插入图片描述


文章目录

  • 系列文章目录
    • @[TOC](文章目录)
  • 前言
  • 1、stream设计思想
  • 2、编码常用的注解
  • 3、编码步骤
    • 3.1、添加依赖
    • 3.2、修改配置文件
    • 3.3、生产
    • 3.4、消费
    • 3.5、延迟队列
      • 3.5.1、修改配置文件
      • 3.5.2、生产端
      • 3.5.2、消息确认机制 消费端
  • 总结

前言

https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit
官方定义Spring Cloud Stream是一个构建消息驱动微服务的框架。应用程序通过inputs或者outputs来与Spring Cloud Stream中binder对象交互。通过我们配置来binding(绑定),而Spring Cloud Stream的binder对象负责与消息中间件交互。

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

Stream让我们不再关注具体MQ的细节我们只需要用一种适配绑定的方式,自动的给我们在各种MQ内切换,总的来说Stream能够屏蔽底层消息中间件的差异、降低切换成本,是统一消息的编程模型。

1、stream设计思想

在这里插入图片描述
在这里插入图片描述

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

2、编码常用的注解

在这里插入图片描述

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

3、编码步骤

3.1、添加依赖

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

3.2、修改配置文件

server:port: 8088spring:cloud:stream:binders: #需要绑定的rabbitmq的服务信息defaultRabbit:  #定义的名称,用于bidding整合type: rabbit  #消息组件类型environment:  #配置rabbimq连接环境spring:rabbitmq:host: localhost   #rabbitmq 服务器的地址port: 5672           #rabbitmq 服务器端口username: tiger       #rabbitmq 用户名password: tiger       #rabbitmq 密码virtual-host: tiger_vh  #虚拟路径bindings:        #服务的整合处理saveOrderOutput:    #这个是消息通道的名称 --->保存订单输出通道destination: exchange-saveOrder     #exchange名称,交换模式默认是topic;把SpringCloud stream的消息输出通道绑定到RabbitMQ的exchange-saveOrder交换器。content-type: application/json      #设置消息的类型,本次为jsondefault-binder: defaultRabbitgroup: saveOrderGroup               #分组saveOrderInput: #生产者绑定,这个是消息通道的名称---> 保存订单输入通道destination: exchange-saveOrder     #exchange名称,交换模式默认是topic;把SpringCloud stream的消息输出通道绑定到RabbitMQ的exchange-saveOrder交换器。content-type: application/json      #设置消息的类型,本次为jsondefault-binder: defaultRabbitgroup: saveOrderGroup               #分组

3.3、生产

/*** 订单消息输出通道处理器*/
@Component
public interface OrderOutputChannelProcesor {@Output("saveOrderOutput")MessageChannel saveOrderOutput();
}
@Slf4j
@EnableBinding(OrderOutputChannelProcesor.class)
public class OrderMessageProducer {@Autowired@Output("saveOrderOutput")private MessageChannel messageChannel;public void sentMsg(UserInfo userInfo){messageChannel.send(MessageBuilder.withPayload(userInfo).build());log.info("消息发送成功:" + userInfo);}
}

3.4、消费

/*** 订单消息输入通道处理器*/
@Component
public interface OrderInputChannelProcesor {@Input("saveOrderInput")SubscribableChannel saveOrderInput();
}
@Slf4j
@EnableBinding(OrderInputChannelProcesor.class)
public class OrderMessageConsumer {@StreamListener("saveOrderInput")public void receiveMsg(Message<UserInfo> userInfoMessage){log.info("接收消息成功:" + userInfoMessage.getPayload());}
}

3.5、延迟队列

安装延迟队列插件:
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/3.11.1/rabbitmq_delayed_message_exchange-3.11.1.ez
下载解压,到plugins目录,执行以下的命令:

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

3.5.1、修改配置文件

server:port: 8088spring:cloud:stream:binders: #需要绑定的rabbitmq的服务信息defaultRabbit:  #定义的名称,用于bidding整合type: rabbit  #消息组件类型environment:  #配置rabbimq连接环境spring:rabbitmq:host: localhost   #rabbitmq 服务器的地址port: 5672           #rabbitmq 服务器端口username: tiger       #rabbitmq 用户名password: tiger       #rabbitmq 密码virtual-host: tiger_vh  #虚拟路径bindings:        #服务的整合处理saveOrderOutput:    #这个是消息通道的名称 --->保存订单输出通道destination: exchange-saveOrder-delay     #exchange名称,交换模式默认是topic;把SpringCloud stream的消息输出通道绑定到RabbitMQ的exchange-saveOrder交换器。content-type: application/json      #设置消息的类型,本次为jsondefault-binder: defaultRabbitgroup: saveOrderGroup               #分组saveOrderInput: #生产者绑定,这个是消息通道的名称---> 保存订单输入通道destination: exchange-saveOrder-delay     #exchange名称,交换模式默认是topic;把SpringCloud stream的消息输出通道绑定到RabbitMQ的exchange-saveOrder交换器。content-type: application/json      #设置消息的类型,本次为jsondefault-binder: defaultRabbitgroup: saveOrderGroup               #分组rabbit:bindings: #服务的整合处理saveOrderOutput:    #这个是消息通道的名称 --->保存订单输出通道producer:delayed-exchange: truesaveOrderInput:consumer:delayed-exchange: true

3.5.2、生产端

@Slf4j
@EnableBinding(OrderOutputChannelProcesor.class)
public class OrderMessageProducer {@Autowired@Output("saveOrderOutput")private MessageChannel messageChannel;public void sentMsg(UserInfo userInfo){messageChannel.send(MessageBuilder.withPayload(userInfo).setHeader("x-delay", 5000).build());log.info("消息发送成功:" + userInfo);}
}

3.5.2、消息确认机制 消费端

rabbit:bindings: #服务的整合处理saveOrderInput:consumer:acknowledge-mode: MANUAL #手动确认
@StreamListener("saveOrderInput")
public void receiveMsg(Message<UserInfo> userInfoMessage){log.info("接收消息成功:" + userInfoMessage.getPayload());Channel channel = (Channel) userInfoMessage.getHeaders().get(AmqpHeaders.CHANNEL);Long delieverTag = (Long) userInfoMessage.getHeaders().get(AmqpHeaders.DELIVERY_TAG);/** deliveryTag:Channel的消息投递的唯一标识符。* multiple:是否否定应答多条消息。如果设置为true,则否定应答带指定deliveryTag的消息及该deliveryTag之前的多条消息;* 如果设置为false,则仅否定应答带指定deliveryTag的单条消息。* requeue:被否定应答的消息是否重入队列。如果设置为true,则消息重入队列;* 如果设置为false,则消息被丢弃或发送到死信Exchange。*/try {channel.basicAck(delieverTag,true);} catch (IOException e) {e.printStackTrace();}
}

定义交换机类型为direct

rabbit:bindings: #服务的整合处理saveOrderInput:consumer:bindingRoutingKey: orderRoutingKeybindQueue: trueexchangeType: directsaveOrderOutput:producer:routingKeyExpression: orderRoutingKeyexchangeType: direct

总结

spring-cloud-stream目前支持RabbitMQ和Kafka,与spring-cloud无缝集成,非常方便。

相关文章:

spring-cloud-stream

系列文章目录 第一章 Java线程池技术应用 第二章 CountDownLatch和Semaphone的应用 第三章 Spring Cloud 简介 第四章 Spring Cloud Netflix 之 Eureka 第五章 Spring Cloud Netflix 之 Ribbon 第六章 Spring Cloud 之 OpenFeign 第七章 Spring Cloud 之 GateWay 第八章 Sprin…...

2.0 熟悉CheatEngine修改器

Cheat Engine 一般简称为CE&#xff0c;它是一款功能强大的开源内存修改工具&#xff0c;其主要功能包括、内存扫描、十六进制编辑器、动态调试功能于一体&#xff0c;且该工具自身附带了脚本工具&#xff0c;可以用它很方便的生成自己的脚本窗体&#xff0c;CE工具可以帮助用户…...

微信小程序数据交互和缓存

目录 前言&#xff1a; 数据交互 1. 发起网络请求 2. WebSocket 2.1实时数据库 3. 微信支付 数据缓存 1. 页面级缓存 2. 内存级缓存 3. 数据缓存策略 优化用户体验 总结 前言&#xff1a; 在开发微信小程序时&#xff0c;数据交互和缓存是非常重要的方面。本文将介…...

kubernetes集群编排——k8s认证授权

pod绑定sa [rootk8s2 ~]# kubectl create sa admin [rootk8s2 secret]# vim pod5.yaml apiVersion: v1 kind: Pod metadata:name: mypod spec:serviceAccountName: admincontainers:- name: nginximage: nginxkubectl apply -f pod5.yamlkubectl get pod -o yaml 认证 [rootk8s…...

rabbitmq下载安装教程

1.首先需要下载erlang和rabbitmq安装包&#xff1a; 官网下载比较慢&#xff0c;通过网盘下载&#xff1a; 链接&#xff1a;https://pan.baidu.com/s/1fM2BrJqefyzUDZD4tfZLIg 提取码&#xff1a;5hsu 2.安装&#xff0c;傻瓜式安装就可以&#xff0c;可以自定义自己要安装的目…...

数据分析实战 | SVM算法——病例自动诊断分析

目录 一、数据分析及对象 二、目的及分析任务 三、方法及工具 四、数据读入 五、数据理解 六、数据准备 七、模型训练 八、模型应用及评价 一、数据分析及对象 CSV文件——“bc_data.csv” 数据集链接&#xff1a;https://download.csdn.net/download/m0_70452407/88…...

Splunk Connect for Kafka – Connecting Apache Kafka with Splunk

1: 背景: 1: splunk 有时要去拉取kafka 上的数据: 下面要用的有用的插件:Splunk Connect for Kafka 先说一下这个Splunk connect for kafka 是什么: What is Splunk Connect for Kafka? Spunk Connect for Kafka is a “sink connector” built on the Kafka Connect…...

Unity | Shader(着色器)和material(材质)的关系

一、前言 在上一篇文章中 【精选】Unity | Shader基础知识&#xff08;什么是shader&#xff09;_unity shader_菌菌巧乐兹的博客-CSDN博客 我们讲了什么是shader&#xff0c;今天我们讲一下shder和material的关系 二、在unity中shader的本质 unity中&#xff0c;shader就…...

Leetcode—69.x的平方根【简单】

2023每日刷题&#xff08;二十七&#xff09; Leetcode—69.x的平方根 直接法实现代码 int mySqrt(int x) {long long i 0;while(i * i < x) {i;}if(i * i > x) {return i - 1;}return i; }运行结果 二分法实现代码 int mySqrt(int x) {long long left 0, right (l…...

再探单例模式

再探单例模式 一&#xff1a;故事背景二&#xff1a;单例重点三&#xff1a;总结提升 一&#xff1a;故事背景 最近在进行单例模式的复习&#xff0c;今天进行一下对应的总结&#xff0c;分析一下各个设计模式。今天从最简单的单例模式开始。 二&#xff1a;单例重点 概念 一…...

Postman使用json提取器和正则表达式实现接口的关联

近期在复习Postman的基础知识&#xff0c;在小破站上跟着百里老师系统复习了一遍&#xff0c;也做了一些笔记&#xff0c;希望可以给大家一点点启发。 一&#xff09;使用json提取器实现接口关联 实际项目场景&#xff0c;在财务信息页面&#xff0c;需要上传一个营业执照&…...

【11.10】现代密码学1——密码学发展史:密码学概述、安全服务、香农理论、现代密码学

密码学发展史 写在最前面密码学概述现代密码学量子密码学基本术语加解密的通信模型对称加密PKI通信工作流程 古典密码与分析古代密码的加密古典密码的分析 安全服务香农理论现代密码学乘积密码方案代换-置换网络安全性概念可证明安全性——规约&#xff08;*规约证明的方案——…...

时间序列预测实战(九)PyTorch实现LSTM-ARIMA融合移动平均进行长期预测

一、本文介绍 本文带来的是利用传统时间序列预测模型ARIMA(注意&#xff1a;ARIMA模型不属于机器学习)和利用PyTorch实现深度学习模型LSTM进行融合进行预测&#xff0c;主要思想是->先利用ARIMA先和移动平均结合处理数据的线性部分&#xff08;例如趋势和季节性&#xff09…...

由日期计算当天是星期几

题目 输入&#xff1a;一个合法的公历日期&#xff0c;格式为“XXXXXXXX”&#xff0c;分别代表年&#xff08;4 位&#xff09;、月&#xff08;2 位&#xff09;、日&#xff08;2 位&#xff09;。 输出&#xff1a;当日对应星期几的英语缩写&#xff08;3 个字母&#xff…...

springboot模板引擎

1.服务端渲染时相比与前后端分离开发 原理是 跳过前端这一层 直接到服务端 通过数据和模板 生成页面返回前端 springboot包含如下模板引擎 典型如thymeleaf 1>导入依赖 2>查看路径 模板页面在 public static final String DEFAULT_PREFIX “classpath:/templates/”; 即…...

如何判断从本机上传到服务器的文件数据内容是一致的?用md5加密算法!

问题场景 最近在帮导师做横向&#xff0c;我想把整个项目环境放到服务器中&#xff0c;需要把一个很大的数据文件传到服务器&#xff0c;传上去很方便&#xff0c;但是涉及到文件的压缩上传和服务器内解压环节&#xff0c;不是太确定文件在本机和服务器的数据内容是否一致。 解…...

Ubuntu 20.04 DNS解析原理, 解决resolv.conf被覆盖问题

------------------------------------------------------------------ author: hjjdebug date: 2023年 11月 09日 星期四 14:01:11 CST description: Ubuntu 20.04 DNS解析原理, 解决resolv.conf被覆盖问题 ----------------------------------------------------------------…...

探索经典算法:贪心、分治、动态规划等

1.贪心算法 贪心算法是一种常见的算法范式&#xff0c;通常在解决最优化问题中使用。 贪心算法是一种在每一步选择中都采取当前状态下最优决策的算法范式。其核心思想是选择每一步的最佳解决方案&#xff0c;以期望达到最终的全局最优解。这种算法特点在于只考虑局部最优解&am…...

【Linux】编译Linux内核

之所以编译内核&#xff0c;是因为gem5全系统仿真需要vmlinux文件&#xff0c;在此记录一下以备后面需要。 此过程编译之后会获得vmlinux和bzImage两个文件&#xff1b; 主要参考知行大佬的编译内核与gem5官方教程 文章目录 一、Linux源码下载二、安装编译依赖三、编译1. 内核编…...

网页判断版本更新

一、需求解析 为什么我会想到这个技术呢&#xff0c;是因为我有一次发现&#xff0c;我司的用户在使用网页的时候&#xff0c;经常会出现一个页面放很久&#xff0c;下班也不关这个页面&#xff0c;这样就会导致页面的代码长时间处于不更新的状态。 在使用到一个功能出了bug&a…...

ESP32读取DHT11温湿度数据

芯片&#xff1a;ESP32 环境&#xff1a;Arduino 一、安装DHT11传感器库 红框的库&#xff0c;别安装错了 二、代码 注意&#xff0c;DATA口要连接在D15上 #include "DHT.h" // 包含DHT库#define DHTPIN 15 // 定义DHT11数据引脚连接到ESP32的GPIO15 #define D…...

屋顶变身“发电站” ,中天合创屋面分布式光伏发电项目顺利并网!

5月28日&#xff0c;中天合创屋面分布式光伏发电项目顺利并网发电&#xff0c;该项目位于内蒙古自治区鄂尔多斯市乌审旗&#xff0c;项目利用中天合创聚乙烯、聚丙烯仓库屋面作为场地建设光伏电站&#xff0c;总装机容量为9.96MWp。 项目投运后&#xff0c;每年可节约标煤3670…...

linux 下常用变更-8

1、删除普通用户 查询用户初始UID和GIDls -l /home/ ###家目录中查看UID cat /etc/group ###此文件查看GID删除用户1.编辑文件 /etc/passwd 找到对应的行&#xff0c;YW343:x:0:0::/home/YW343:/bin/bash 2.将标红的位置修改为用户对应初始UID和GID&#xff1a; YW3…...

ios苹果系统,js 滑动屏幕、锚定无效

现象&#xff1a;window.addEventListener监听touch无效&#xff0c;划不动屏幕&#xff0c;但是代码逻辑都有执行到。 scrollIntoView也无效。 原因&#xff1a;这是因为 iOS 的触摸事件处理机制和 touch-action: none 的设置有关。ios有太多得交互动作&#xff0c;从而会影响…...

MySQL账号权限管理指南:安全创建账户与精细授权技巧

在MySQL数据库管理中&#xff0c;合理创建用户账号并分配精确权限是保障数据安全的核心环节。直接使用root账号进行所有操作不仅危险且难以审计操作行为。今天我们来全面解析MySQL账号创建与权限分配的专业方法。 一、为何需要创建独立账号&#xff1f; 最小权限原则&#xf…...

Android第十三次面试总结(四大 组件基础)

Activity生命周期和四大启动模式详解 一、Activity 生命周期 Activity 的生命周期由一系列回调方法组成&#xff0c;用于管理其创建、可见性、焦点和销毁过程。以下是核心方法及其调用时机&#xff1a; ​onCreate()​​ ​调用时机​&#xff1a;Activity 首次创建时调用。​…...

基于 TAPD 进行项目管理

起因 自己写了个小工具&#xff0c;仓库用的Github。之前在用markdown进行需求管理&#xff0c;现在随着功能的增加&#xff0c;感觉有点难以管理了&#xff0c;所以用TAPD这个工具进行需求、Bug管理。 操作流程 注册 TAPD&#xff0c;需要提供一个企业名新建一个项目&#…...

深入浅出深度学习基础:从感知机到全连接神经网络的核心原理与应用

文章目录 前言一、感知机 (Perceptron)1.1 基础介绍1.1.1 感知机是什么&#xff1f;1.1.2 感知机的工作原理 1.2 感知机的简单应用&#xff1a;基本逻辑门1.2.1 逻辑与 (Logic AND)1.2.2 逻辑或 (Logic OR)1.2.3 逻辑与非 (Logic NAND) 1.3 感知机的实现1.3.1 简单实现 (基于阈…...

MySQL 部分重点知识篇

一、数据库对象 1. 主键 定义 &#xff1a;主键是用于唯一标识表中每一行记录的字段或字段组合。它具有唯一性和非空性特点。 作用 &#xff1a;确保数据的完整性&#xff0c;便于数据的查询和管理。 示例 &#xff1a;在学生信息表中&#xff0c;学号可以作为主键&#xff…...

Oracle11g安装包

Oracle 11g安装包 适用于windows系统&#xff0c;64位 下载路径 oracle 11g 安装包...