SpringCloud-Stream

news2025/1/24 22:43:22

一、介绍

（1）提供统一接口操作不同厂商的消息队列组件，降低学习成本
（2）生产者和消费者只需操作binder对象即可与消息队列交互，生产者output，消费者input
（3）核心概念：发布订阅、消费组、分区
（4）使用topic模式
在这里插入图片描述
在这里插入图片描述

二、项目搭建

（1）生产者
a、编写pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>demo20220821</artifactId>
        <groupId>com.wsh.springcloud</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>cloud-stream-rabbitmq-provider8801</artifactId>

    <dependencies>
        <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>com.wsh.springcloud</groupId>
            <artifactId>cloud-api-common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </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-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

</project>

b、编写application.yml

server:
  port: 8801

spring:
  application:
    name: cloud-stream-rabbit-provider
  cloud:
    stream:
      binders:
        defaultRabbit:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: 192.168.0.166
                port: 5672
                username: guest
                password: guest
      bindings:
        output:
          destination: testExchange
          content-type: application/json
          binder: defaultRabbit


eureka:
  client:
    #    客户端设置为true
    register-with-eureka: true
    #    客户端设置为true
    fetch-registry: true
    service-url:
      #      defaultZone: http://localhost:7001/eureka
      defaultZone: http://eureka1.com:7001/eureka, http://eureka2.com:7002/eureka
  instance:
    instance-id: cloudSreamRabbitProvider8801
    prefer-ip-address: true

management:
  endpoints:
    web:
      exposure:
        include: "*"

c、编写启动类

package com.wsh.springcloud;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.netflix.eureka.EnableEurekaClient;

/**
 * @ClassName ConfigMain3344
 * @Description: TODO
 * @Author wshaha
 * @Date 2023/10/15
 * @Version V1.0
 **/
@SpringBootApplication
@EnableEurekaClient
public class StreamRabbitMqProvider8801 {
    public static void main(String[] args) {
        SpringApplication.run(StreamRabbitMqProvider8801.class, args);
    }
}

d、编写接口及实现类

package com.wsh.springcloud.service;

public interface IMessageProvider {

    public String send();
}


package com.wsh.springcloud.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
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;

/**
 * @ClassName MessageProviderImpl
 * @Description: TODO
 * @Author wshaha
 * @Date 2023/10/15
 * @Version V1.0
 **/
@EnableBinding(Source.class)
public class MessageProviderImpl implements IMessageProvider {

    @Autowired
    @Qualifier("output")
    private MessageChannel messageChannel;

    @Override
    public String send() {
        messageChannel.send(MessageBuilder.withPayload("hello").build());
        return null;
    }
}

e、编写Controller

package com.wsh.springcloud.controller;

import com.wsh.springcloud.service.IMessageProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @ClassName MessageController
 * @Description: TODO
 * @Author wshaha
 * @Date 2023/10/15
 * @Version V1.0
 **/
@RestController
public class MessageController {

    @Autowired
    private IMessageProvider messageProvider;

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

（2）编写消费者
a、编写pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>demo20220821</artifactId>
        <groupId>com.wsh.springcloud</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>cloud-stream-rabbitmq-consumer8802</artifactId>

    <dependencies>
        <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>com.wsh.springcloud</groupId>
            <artifactId>cloud-api-common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </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-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>

b、编写application.yml

server:
  port: 8802

spring:
  application:
    name: cloud-stream-rabbit-provider
  cloud:
    stream:
      binders:
        defaultRabbit:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: 192.168.0.166
                port: 5672
                username: guest
                password: guest
      bindings:
        input:
          destination: testExchange
          content-type: application/json
          binder: defaultRabbit


eureka:
  client:
    #    客户端设置为true
    register-with-eureka: true
    #    客户端设置为true
    fetch-registry: true
    service-url:
      #      defaultZone: http://localhost:7001/eureka
      defaultZone: http://eureka1.com:7001/eureka, http://eureka2.com:7002/eureka
  instance:
    instance-id: cloudSreamRabbitProvider8801
    prefer-ip-address: true

management:
  endpoints:
    web:
      exposure:
        include: "*"

c、编写启动类

package com.wsh.springcloud;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.netflix.eureka.EnableEurekaClient;

/**
 * @ClassName ConfigMain3344
 * @Description: TODO
 * @Author wshaha
 * @Date 2023/10/15
 * @Version V1.0
 **/
@SpringBootApplication
@EnableEurekaClient
public class StreamRabbitMqConsumer8802 {
    public static void main(String[] args) {
        SpringApplication.run(StreamRabbitMqConsumer8802.class, args);
    }
}

d、编写Controller

package com.wsh.springcloud.controller;

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.web.bind.annotation.RestController;

/**
 * @ClassName ConsumerController
 * @Description: TODO
 * @Author wshaha
 * @Date 2023/10/15
 * @Version V1.0
 **/
@RestController
@EnableBinding(Sink.class)
public class ConsumerController {

    @StreamListener(Sink.INPUT)
    public void receiveMessage(Message<String> message){
        System.out.println(message.getPayload());
    }
}

（3）运行
在这里插入图片描述

三、解决消息重复消费

（1）绑定同一交换机且不同组的消费者会收到相同消息
（2）解决方式，同一组的消费者只有一个消费者会收到消息，故配置这群消费者为同一个组即可
（3）配置
在这里插入图片描述
在这里插入图片描述

四、消息持久化

（1）定义分组后会实现消息持久化，原理：没定义分组时，服务对应的队列是autodelete，服务停止后就删除队列，手续发送的消息无法收到

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

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

相关文章

RobotFramework用户关键字（一）

RobotFramework用户关键字（一）

文章目录 一 用户关键字1. 创建用户关键字2. 用户关键字的使用3. 用户关键字参数4. 用户关键字返回参数 一 用户关键字 Robotframework是一个用python写的基于关键字驱动的自动化测试框架。既然是基于关键字驱动的&#xff0c;那么关键字的编写和使用就是平时工作的核心。这篇文…
阅读更多...
单链表经典OJ题

单链表经典OJ题

目录 ​编辑 题目&#xff1a; 一、移除链表元素&#xff1a; 本质&#xff1a; 解题思路&#xff1a; 本题分为两种解法&#xff1a; 我们使用解法二&#xff1a; 注意事项&#xff1a; 完整代码&#xff1a; 题目&#xff1a; 一、移除链表元素&#xff1a; 本质&…
阅读更多...
论文阅读：Segment Any Point Cloud Sequences by Distilling Vision Foundation Models

论文阅读：Segment Any Point Cloud Sequences by Distilling Vision Foundation Models

论文地址&#xff1a;[2306.09347] Segment Any Point Cloud Sequences by Distilling Vision Foundation Models (arxiv.org) 代码地址&#xff1a;GitHub - youquanl/Segment-Any-Point-Cloud: [NeurIPS23 Spotlight] Segment Any Point Cloud Sequences by Distilling Visi…
阅读更多...
计网第六章（应用层）（二）（域名系统DNS）

计网第六章（应用层）（二）（域名系统DNS）

一、基本概述 域名和IP地址&#xff1a;在1983年&#xff0c;因特网就开始采用层次结构的命名树作为主机的名字&#xff0c;即域名。因为域名比IP地址更加方便人们记忆&#xff0c;人们可以通过域名更加方便地访问互联网。但实际上网络识别的仍然是IP地址&#xff0c;所以就有…
阅读更多...
文件操作(打开关闭文件、文件顺序以及随机读写)

文件操作(打开关闭文件、文件顺序以及随机读写)

文章目录 写在前面1. 文件的打开与关闭1.1 文件指针1.2 文件的打开(fopen)与关闭(fclose)1.2.1 fopen函数1.2.2 fclose函数 2. 文件的顺序读写2.1. fgetc 和 fputc函数2.1.1 fputc函数2.1.2 fgetc函数 2.2 fgets 和 fputs函数2.2.1 fputs函数2.2.2 fgets函数 2.3 fscanf和fprin…
阅读更多...
密码学技术总结

密码学技术总结

前言 本文内容主要摘抄网络规划设计师的教材和腾讯-SUMMER课堂&#xff0c;主要对网络安全进行简单梳理和总结 对称密码体制 密码分为私钥和公钥密码两种&#xff0c;而介于私钥和公钥只见密码成为混合密码。 私钥密码又称为对称密码&#xff0c;该体制的特点是加密和解密…
阅读更多...
软件开发项目文档系列之二如何撰写项目建设方案

软件开发项目文档系列之二如何撰写项目建设方案

目录 前言1 项目概况2 需求分析3 建设目标4 建设内容4.1 建设原则4.2 设计依据4.3 总体框架4.4 建设任务4.5 技术方案 5 实施保障措施6 验收和考核指标7 投资预算7.1 资金来源7.2 预算表7.3 预算编制说明 结语 前言 建设方案或解决方案是在任何新项目或计划启动之前&#xff0…
阅读更多...
程序员如何运营好博客平台

程序员如何运营好博客平台

shigen坚持日更的博客写手&#xff0c;擅长Java、python、vue、shell等编程语言和各种应用程序、脚本的开发。坚持记录和分享从业两年以来的技术积累和思考&#xff0c;不断沉淀和成长。 不知不觉间&#xff0c;我发现从我的日更博客和公众号目标确定到现在&#xff0c;我已经坚…
阅读更多...
数据库实验一：学生信息管理系统数据库结构搭建和表的创建

数据库实验一：学生信息管理系统数据库结构搭建和表的创建

实验项目名称&#xff1a;学生信息管理系统数据库结构搭建和表的创建 实验目的与要求实验原理与内容1. 数据库的组织结构2. 数据库的分离和附加3. 数据库表的创建&#xff0c;修改和删除 实验过程与结果1. 根据学生信息管理系统创建相关的数据库2. 数据库表初步设计及实现3. 实…
阅读更多...
【算法训练-排序算法 二】【手撕排序】快速排序、堆排序、归并排序

【算法训练-排序算法 二】【手撕排序】快速排序、堆排序、归并排序

废话不多说&#xff0c;喊一句号子鼓励自己&#xff1a;程序员永不失业&#xff0c;程序员走向架构&#xff01;本篇Blog的主题是【手撕排序系列】&#xff0c;使用【数组】这个基本的数据结构来实现&#xff0c;这个高频题的站点是&#xff1a;CodeTop&#xff0c;筛选条件为&…
阅读更多...
ffmpeg+安卓+yolo+RK3399部署

ffmpeg+安卓+yolo+RK3399部署

一次满足多项需求. 首先, 思路是, 使用ffmpeg解码本地mp4文件, 在无需任何其他改动的情况下, 就可以直接播放rtsp流, 这个是使用ffmpeg的好处. ffmpeg本身是c语言的, 所以需要编译成jni的库, https://note.youdao.com/s/6XeYftc 具体过程在这里, 用windows/macOS, Ubuntu应该都…
阅读更多...
Python 字典

Python 字典

目录 1 字典介绍2 字典的创建3 字典元素的访问4 字典元素添加、修改、删除5 序列解包6 表格数据使用字典和列表存储&#xff0c;并实现访问7 字典核心底层原理(重要)7.1 将一个键值对放进字典的底层过程7.2 扩容7.3 根据键查找“键值对”的底层过程7.4 用法总结&#xff1a; 声…
阅读更多...
【多媒体技术与实践】使用OpenCV处理图像（实验三.上）

【多媒体技术与实践】使用OpenCV处理图像（实验三.上）

1&#xff1a;图像直方图 将原彩色图像转成灰度图像&#xff0c;得到该灰度图像的灰度直方图&#xff0c;并对灰度直方图进行直方图均衡化&#xff0c;将原图、灰度图、直方图及均衡化后的直方图一起拼接为一张图片 import cv2 import numpy as np import matplotlib.pyp…
阅读更多...
数据结构与算法 | 第一章：概论

数据结构与算法 | 第一章：概论

本文参考网课为 数据结构与算法 1 第一章概论&#xff0c;主讲人 张铭 、王腾蛟 、赵海燕 、宋国杰 、邹磊 、黄群。 本文使用IDE为 Clion&#xff0c;开发环境 C14。 更新&#xff1a;2023 / 10 / 15 数据结构与算法 | 第一章&#xff1a;概论 数据结构概念逻辑存储运算抽象…
阅读更多...
Java基础--阳光总在风雨后，请相信彩虹

Java基础--阳光总在风雨后，请相信彩虹

1、今日任务 JAVA SE-韩顺平视频教程–30p以上&#xff08;今天得50p以上因为是基础&#xff09;计算机基础八股记忆总结刷题&#xff08;两题&#xff09;可以先用python 1、SSM ssm->Spring&#xff08;轻量级的文本开发框架&#xff09;/SpringMVC&#xff08;分层的w…
阅读更多...
【string题解 C++】翻转字符串II：区间部分翻转 | 验证回文串

【string题解 C++】翻转字符串II：区间部分翻转 | 验证回文串

翻转字符串II&#xff1a;区间部分翻转 力扣&#xff08;LeetCode&#xff09;官网 - 全球极客挚爱的技术成长平台 难度&#xff1a;简单 给定一个字符串 s 和一个整数 k&#xff0c;从字符串开头算起&#xff0c;每计数至 2k 个字符&#xff0c;就反转这 2k 字符中的前 k 个…
阅读更多...
【Mybatis】基于TypeHandler实现敏感数据加密

【Mybatis】基于TypeHandler实现敏感数据加密

一、介绍 业务场景中经常会遇到诸如用户手机号&#xff0c;身份证号&#xff0c;银行卡号&#xff0c;邮箱&#xff0c;地址&#xff0c;密码等等信息&#xff0c;属于敏感信息&#xff0c;需要保存在数据库中。而很多公司会会要求对数据库中的此类数据进行加密存储。 敏感数据…
阅读更多...
【JY】求？减隔震元件的滞回面积~

【JY】求？减隔震元件的滞回面积~

写在前文 为了求解等效阻尼比&#xff0c;通常我们需要求解滞回圈的面积&#xff0c;由于形状不是常见规整可直接套用长乘宽的形状&#xff0c;因此大多数人经常无从下手&#xff0c;或者分割分割再分割&#xff0c;一顿操作猛如虎&#xff0c;然后再得出结果。 计算方法 不论是…
阅读更多...
私仓库Nexus

私仓库Nexus

私仓库Nexus搭建 没有命令提示符tab键无效 #安装Linux命令补全工具 yum -y install bash-completion #执行bash或者reboot重启系统 bash#如果上述的命令执行了有问题可以执行功能下面的命令 yum install epel-release -y yum install bash-completion bash-completion-extras…
阅读更多...
[资源推荐]langchain、LLM相关

[资源推荐]langchain、LLM相关

之前很多次逛github或者去B站看东西或者说各种浏览资讯的情况&#xff0c;都会先看两眼然后收藏然后就吃灰的情况&#xff0c;那既然这样&#xff0c;不如多看几眼&#xff0c;看看是否真的能用得上&#xff0c;能用在哪&#xff0c;然后用几句话总结出来&#xff0c;分享出来&…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023