文档建议反馈控制台
首页
学习
活动
专区
圈层
工具
MCP广场
文章/答案/技术大牛
发布
首页
学习
活动
专区
圈层
工具
MCP广场
返回腾讯云官网
社区首页 >专栏 >SpringCloud Stream 消息驱动

SpringCloud Stream 消息驱动

作者头像
OY
发布于 2022-03-17 10:22:14
发布于 2022-03-17 10:22:14
32900
代码可运行
举报
文章被收录于专栏：OY_学习记录OY_学习记录
运行总次数：0
代码可运行
关联问题
换一批

博客学习参考视频

一、消息驱动概述

① 是什么

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

官网

  1. https://spring.io/projects/spring-cloud-stream#overview Spring Cloud Stream 是用于构建与共享 消息传递系统 连接的高度可伸缩的事件驱动微服务框架，该框架提供一个灵活的编程模型，它建立在 spring 的基础上，包括支持持久化的发布/订阅、消费以及分区 这三个核心概念
  1. https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.1.RELEASE/reference/html/
  2. Spring Cloud Stream 中 文 指 导 手 册 ：https://m.wang1314.com/doc/webapp/topic/20971999.html

② 设计思想

标准 MQ

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

为什么用 Cloud Stream

  1. 比方说我们用到的了 RabbitMQ 和 Kafka，由于这两个消息中间件的架构上的不同，像 RabbitMQ 有 exchange, kafka 有 Topic 和 Partitions 分区
  1. 这些中间件的差异性导致我们实际项目开发中会造成一定困扰， 我们如果用了两个消息队列的其中一种， 后面的业务需求，我们想往另外一种消息队列进行迁移， 这时候无疑就是一个灾难性的， 一大堆东西都要中心推倒重新做， 因为他跟我们的系统耦合了， 这时候 springcloud Stream 给我们提供了一种解耦合的方式。
  2. stream 凭什么可以统一底层差异 。

4.Binder

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

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

  1. Topic 主题进行广播
  • 在 RabbitMQ 就是 Exchange
  • 在 Kafka 中就是 Topic

③ Spring Cloud Stream 标准流程套路

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

④ 编码 API 和常用注解

二、案例说明

① RabbitaMQ 环境已经 OK

​ 请参考这篇博客： SpringBoot 与消息

② 工程中新建三个子模块

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

三、消息驱动之生产者

新建 Module： cloud-stream-rabbitmq-provider8801

POM

代码语言：javascript
代码运行次数：0
运行
复制
<?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>clould</artifactId>
        <groupId>com.oy</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

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

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</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>

</project>

YML

代码语言：javascript
代码运行次数：0
运行
复制
server:
  port: 8801

spring:
  application:
    name: cloud-stream-provider
  cloud:
    stream:
      binders: # 在此处配置要绑定的rabbitmq的服务信息；
        defaultRabbit: # 表示定义的名称，用于于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置rabbitmq的相关的环境配置
            spring:
              rabbitmq:
                host: 116.63.177.72
                port: 5672
                username: guest
                password: guest
      bindings: # 服务的整合处理
        output: # 这个名字是一个通道的名称
          destination: studyExchange # 表示要使用的Exchange名称定义
          content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”
          binder: defaultRabbit # 设置要绑定的消息服务的具体设置

eureka:
  client: # 客户端进行Eureka注册的配置
    service-url:
      defaultZone: http://localhost: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地址

主启动类

代码语言：javascript
代码运行次数：0
运行
复制
@SpringBootApplication
public class StreamMQMain8801 {
     public static void main(String[] args) {
           SpringApplication.run(StreamMQMain8801.class, args);
      }
}

业务类

发消息接口：

代码语言：javascript
代码运行次数：0
运行
复制
public interface IMessageProvider {
    public String send();
}

发送消息接口实现类

代码语言：javascript
代码运行次数：0
运行
复制
@EnableBinding(Source.class) // 定义消息的推送管道
public class MessionProvicerImpl implements IMessageProvider {

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

    @Override
    public String send() {
        String serial = UUID.randomUUID().toString();
        output.send(MessageBuilder.withPayload(serial).build());
        System.out.println("******serial:" + serial);
        return null;
    }
}

Controller

代码语言：javascript
代码运行次数：0
运行
复制
@RestController
public class SendMessageController {

    @Resource
    private IMessageProvider messageProvider;

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

测试

  • 启动 7001eureka
  • 启动 rabbitmq http://localhost:15672/ （自己使用的是 Docker）
  • 启动 8801 访问： http://localhost:8801/sendMessage

注意：有个大坑,视频里的 application.yml 使用了 spring.cloud.stream.binders.defaultRabbit.environment.spring.rabbitmq.xx

来配置 rabbitmq 的环境如果你是用的其他服务器上的 rabbitmq，比如我使用的我自己的华为云服务器然后创建 docker 容器来运行

rabbitmq。按照视频中的配置方式的话，启动时会试图连接两次 rabbitmq 程序第一次试图连接访问的就是 application.yml 中配置

的地址，此时已经订阅成功了但是程序还会在之后进行第二次连接，此时访问的地址就是 localhost:5672，在我的环境中，我本地

没有 rabbitmq 环境，所以直接报 IOException 所以，如果是使用的自己的服务器来配置，则需要修改配置文件，将 rabbitmq 的配置

信息移动到 application.yml 中的 spring 节点下修改后的配置如下：

代码语言：javascript
代码运行次数：0
运行
复制
server:
  port: 8801

spring:
  application:
    name: cloud-stream-provider
  rabbitmq:
    host: 华为云ip
    port: 5672
    username: guest
    password: guest
  cloud:
    stream:
      binders:
        defaultRabbit:
          type: rabbit
      bindings:
        output:
          destination: studyExchange
          content-type: application/json
          binder: defaultRabbit

eureka:
  client: # 客户端进行Eureka注册的配置
    service-url:
      defaultZone: http://localhost: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地址

四、消息驱动之消费者

新建 Module

​ cloud-stream-rabbitmq-consumer8802

POM

代码语言：javascript
代码运行次数：0
运行
复制
<?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>clould</artifactId>
        <groupId>com.oy</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>
        <!--
        https://mvnrepository.com/artifact/org.springframework.cloud/spring-cloud-starter-eure
        ka-server -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>
        <dependency>
            <groupId>com.oy</groupId>
            <artifactId>cloud-api-commons</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <!--https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-web
        -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!--
        https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-web
        -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <!--
        https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-devtools -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId><optional>true</optional>
        </dependency>
        <!--
        https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-test
        -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>

YML

代码语言：javascript
代码运行次数：0
运行
复制
server:
  port: 8802

spring:
  application:
    name: cloud-stream-consumer
  rabbitmq:
    host: 华为云ip // localhost
    port: 5672
    username: guest
    password: guest
  cloud:
    stream:
      binders: # 在此处配置要绑定的rabbitmq的服务信息；
        defaultRabbit: # 表示定义的名称，用于于binding整合
          type: rabbit # 消息组件类型

      bindings: # 服务的整合处理
        input: # 这个名字是一个通道的名称
          destination: studyExchange # 表示要使用的Exchange名称定义
          content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”
          binder: defaultRabbit # 设置要绑定的消息服务的具体设置
eureka:
  client: # 客户端进行 Eureka 注册的配置
    service-url:
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是 30 秒）
    lease-expiration-duration-in-seconds: 5 # 如果现在超过了 5 秒的间隔（默认是 90秒）
    instance-id: receive-8802.com # 在信息列表时显示主机名称
    prefer-ip-address: true # 访问的路径变为 IP 地址

主启动

代码语言：javascript
代码运行次数：0
运行
复制
@SpringBootApplication
public class StreamMQMain8802 {
     public static void main(String[] args) {
           SpringApplication.run(StreamMQMain8802.class, args);
      }
}

业务类

代码语言：javascript
代码运行次数：0
运行
复制
package com.oy.springcloud.controller;

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.messaging.Message;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Component;

/**
 * @Author OY
 * @Date 2020/10/22
 */
@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListenerController {

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

    @StreamListener(Sink.INPUT)
    public void input(Message<String> message){
        System.out.println(" 消 费 者 1 号 ， 接 受 ： "+message.getPayload()+"\t port:"+serverPort);
    }
}

测试 8801 发送 8802 接收消息：

​ http://localhost:8801/sendMessage

五、分组消费与持久化

① 配置 8803

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

cloud-stream-rabbitmq-consumer8803

POM

代码语言：javascript
代码运行次数：0
运行
复制
<?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>clould</artifactId>
        <groupId>com.oy</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

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

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</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>
</project>

YML

代码语言：javascript
代码运行次数：0
运行
复制
server:
  port: 8803

spring:
  application:
    name: cloud-stream-consumer
  rabbitmq:
    host: 116.63.177.72
    port: 5672
    username: guest
    password: guest
  cloud:
    stream:
      binders: # 在此处配置要绑定的rabbitmq的服务信息；
        defaultRabbit: # 表示定义的名称，用于于binding整合
          type: rabbit # 消息组件类型

      bindings: # 服务的整合处理
        input: # 这个名字是一个通道的名称
          destination: studyExchange # 表示要使用的Exchange名称定义
          content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”
          binder: defaultRabbit # 设置要绑定的消息服务的具体设置
eureka:
  client: # 客户端进行 Eureka 注册的配置
    service-url:
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是 30 秒）
    lease-expiration-duration-in-seconds: 5 # 如果现在超过了 5 秒的间隔（默认是 90秒）
    instance-id: receive-8803.com # 在信息列表时显示主机名称
    prefer-ip-address: true # 访问的路径变为 IP 地址

主启动类

代码语言：javascript
代码运行次数：0
运行
复制
@SpringBootApplication
public class StreamMQMain8803 {
     public static void main(String[] args) {
           SpringApplication.run(StreamMQMain8803.class, args);
      }
}

Controller

代码语言：javascript
代码运行次数：0
运行
复制
package com.oy.springcloud.controller;

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 OY
 * @Date 2020/10/22
 */
@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListenerController {

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

    @StreamListener(Sink.INPUT)
    public void input(Message<String> message){
        System.out.println(" 消 费 者 2 号 ， 接 受 ： "+message.getPayload()+"\t port:"+serverPort);
    }
}

② 启动

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

③ 运行后两个问题

  1. 有重复消费问题
  2. 消息持久化问题 、

④ 消费： 目前是 8802/8803 同时都收到了， 存在重复消费问题

  1. http://localhost:8801/sendMessage
  1. 如何解决： 分组和持久化属性 group
  2. 生产实际案例

⑤ 分组

原理

​ 微服务应用放置于同一个 group 中，就能保证消息只会被其中一个应用消费一次。不同的组是可以消费的。同一个组内发生的竞争的关系，只有一个可以消费。

8802/8803 都变成不同组， group 两个不同

  1. group: atguiguA、 atguiguB
  2. 8802 修改 YML
代码语言：javascript
代码运行次数：0
运行
复制
group: atguiguA
  1. 8803 修改 YML
代码语言：javascript
代码运行次数：0
运行
复制
group: atguiguB
  1. 我们自己配置

​ 分布式微服务应用为了实现高可用和负载均衡，实际上都会户数多个实例，本例启动了两个消费微服务（8802/8803）多数情况，生产者发送消息给某个具体微服务只希望被消费一次，按照上面我们启动两个应用的例子，虽然它们同属一个应用，但是这个消息出现了被重复消费两次的情况。为了解决这个问题，在 Spring Cloud Stream 中提供了消费组的概念。

  1. 结论： 还是重复消费

8802/8803 实现了轮询分组， 每次只有一个消费者 8801 模块的发的消息只能被 8802 或 8803 其中一个接收到， 这样避免了重复

消费 。

8802/8803 都变成相同组， group 两个相同 。

  1. group: atguiguA
  2. 8802 修改 YML
代码语言：javascript
代码运行次数：0
运行
复制
group: atguiguA
  1. 8803 修改 YML
代码语言：javascript
代码运行次数：0
运行
复制
group: atguiguA

结论： 同一个组的多个微服务实例， 每次只会有一个拿到

⑥ 持久化

通过上述， 解决了重复消费问题， 再看看持久化，停止 8802/8803 并去除掉 8802 的分组 group:atguiguA，8803 的分组

group:atguiguA 没有去掉。8801 先发送 4 条信息到 rabbitmq。

  • 先启动 8802， 无分组属性配置， 后台没有打出来消息
  • 先启动 8803， 有分组属性配置， 后台打出来了 MQ 上的消息
本文参与 腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。
原始发表：2020-10-21，如有侵权请联系 cloudcommunity@tencent.com 删除
微服务
消息队列 CMQ 版
kafka
spring cloud

本文分享自 作者个人站点/博客 前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ，欢迎热爱写作的你一起参与！

微服务
消息队列 CMQ 版
kafka
spring cloud
评论
登录后参与评论
暂无评论
登录 后参与评论
推荐阅读
SpringCloud之Stream
springspring cloudrabbitmq消息队列 CMQ 版kafka
比方说我们用到了RabbitMQ和Kafka，由于这两个消息中间件的架构上的不同，像RabbitMQ有exchange，kafka有Topic和Partitions分区。
shaoshaossm
2022/12/27
3860
SpringCloud之Stream
SpringCloud集成Stream
spring消息队列 CMQ 版spring cloudrabbitmqkafka
有没有一种新的技术诞生，让我们不再关注具体MQ的细节，我们只需要用一种适配绑定的方式，自动的给我们在各种MQ内切换。（类似于Hibernate）
大忽悠爱学习
2021/12/08
5180
SpringCloud集成Stream
消息驱动(SpringCloud Stream)
springspring cloudrabbitmqhttps网络安全
官网：https://spring.io/projects/spring-cloud-stream#overview https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.1.RELEASE/reference/html/ Spring Cloud Stream中文指导手册： https://m.wang1314.com/doc/webapp/topic/20971999.html
鱼找水需要时间
2023/02/16
4590
消息驱动(SpringCloud Stream)
SpringCloud Stream消息驱动
rabbitmqspringspring cloud
  官方定义 Spring Cloud Stream 是一个构建消息驱动微服务的框架。应用程序通过 inputs 或者 outputs 来与 Spring Cloud Stream中binder对象交互。通过我们配置来binding(绑定) ，而 Spring Cloud Stream 的 binder对象负责与消息中间件交互。
别团等shy哥发育
2023/02/25
4600
SpringCloud Stream消息驱动
Stream 消息驱动
springspring cloudrabbitmqkafka消息队列 CMQ 版
有没有一种新的技术诞生，让我们不再关注具体MQ的细节，我们只需要用一种适配绑定的方式，自动的给我们在各种MQ内切换。（类似于Hibernate）
用户9615083
2022/12/25
4430
Stream 消息驱动
SpringCloud Stream消息驱动
stream服务配置消息中间件中间件
cheese
2023/10/25
3730
SpringCloud Stream消息驱动
15-SpringCloud Stream
spring消息队列 CMQ 版rabbitmqspring cloudkafka
应用程序通过inputs或者 outputs 来与Spring Cloud Stream中binder对象交互。
彼岸舞
2021/09/09
5700
15-SpringCloud Stream
springCloud --- 中级篇(3)
rabbitmqspring cloudjar
本系列笔记涉及到的代码在GitHub上，地址：https://github.com/zsllsz/cloud
贪挽懒月
2020/06/08
8480
springCloud --- 中级篇(3)
springCloud - 第4篇 - 消费者调用服务 ( Feign )
微服务springspring cloud负载均衡
2. spring Initializr - module SDK 选择自己的 JDK ，其余的可以不用填写，next。
微风-- 轻许--
2019/08/01
5460
10SpringCloud Stream消息驱动
rabbitmqhttpmaven腾讯云测试服务
原因：rabbitmq从3.3.0开始禁止使用guest/guest权限通过除localhost外的访问
Remember_Ray
2020/11/04
3590
10SpringCloud Stream消息驱动
Spring cloud stream【入门介绍】
rabbitmqspringkafka微服务
案例代码:https://github.com/q279583842q/springcloud-e-book
用户4919348
2019/07/03
1.3K0
Spring cloud stream【入门介绍】
微服务（十二）——Steam消息驱动&Sleuth链路监控
spring消息队列 CMQ 版spring cloudrabbitmqhttp
有没有一种新的技术诞生，让我们不再关注具体MQ的细节，我们只需要用一种适配绑定的方式，自动的给我们在各种MQ内切换。（类似于Hibernate）
不愿意做鱼的小鲸鱼
2022/09/26
5050
微服务（十二）——Steam消息驱动&Sleuth链路监控
SpringCloud Stream消息驱动代码实战
rabbitmqhttp腾讯云测试服务json
cloud-stream-rabbitmq-provider8801， 作为生产者进行发消息模块
一个风轻云淡
2022/11/13
2450
SpringCloud Stream消息驱动代码实战
SpringCloud Stream消息驱动
springspring cloud消息队列 CMQ 版kafkarabbitmq
目前市面上常用的四种消息中间件：ActiveMQ、RabbitMQ、RocketMQ、Kafka。由于每个项目需求的不同，在消息中间件的选型上也就会不同。
乐心湖
2021/01/18
9180
SpringCloud Stream消息驱动
springcloud : Stream消息驱动
springspring cloud消息队列 CMQ 版kafkarabbitmq
官网 : https://spring.io/projects/spring-cloud-stream#overview
冷环渊
2021/10/19
7000
springcloud之Stream
服务客户端配置消息中间件stream
创建项目cloud-stream-rabbitmq-consumer8803与cloud-stream-rabbitmq-consumer8802是并列的消费者。 此时生产者发送消息，这两个消费者都会收到，因为默认情况下不同的微服务是不同组，是并列关系，只要改为竞争关系即可。
Java微观世界
2025/01/20
1030
springcloud之Stream
Spring Cloud 系列之消息驱动 Stream
springspring bootspring cloudrabbitmq微服务
  在一个系统中我们可能包含前端页面、接口服务、大数据层，可能在接口服务中使用的是 RabbitMQ 而在大数据层中使用的是 Kafka，那么我只会 RabbitMQ 不会 Kafka 岂不是还要去学习，白天 996 晚上 007 简直要命。那么有没有一个像 JDBC 一样的能够屏蔽细节让我们可以迅速切换。   Spring Cloud Stream 是一个构建消息驱动微服务应用的框架。它基于 Spring Boot 构建独立的、生产级的 Spring 应用，并使用 Spring Integration 为消息代理提供链接。应用程序通过 inputs 或者 outputs 来与 Spring Cloud Stream 中 binder 交互，通过我们配置来 binding ，而 Spring Cloud Stream 的 binder 负责与中间件交互。所以，我们只需要搞清楚如何与 Spring Cloud Stream 交互就可以方便使用消息驱动的方式。 Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区的三个核心概念。目前只实现了 Kafka 和 RabbitMQ 的 Binder。
Demo_Null
2020/11/24
1.5K0
Spring Cloud 系列之消息驱动 Stream
SpringCloud 分布式配置中心
httphttps网络安全微服务
​ 微服务意味着要将单体应用中的业务拆分成一个个子服务，每个服务的粒度相对小，因此系统中出现大量的服务，由于每个服务都需要配置信息才能运行，所以一套集中式的、动态的配置管理设施是必不可少的。
OY
2022/03/17
3490
SpringCloud 分布式配置中心
springCloud - 第3篇 - 消费者调用服务 ( RestTemplate + Ribbon )
微服务spring cloudspring负载均衡
2. spring Initializr - module SDK 选择自己的 JDK ，其余的可以不用填写，next。
微风-- 轻许--
2019/08/01
7300
RabbitMQ与Spring的框架整合之Spring Cloud Stream实战
rabbitmqspringspring cloud
1、RabbitMQ与Spring Cloud Stream整合实战。SpringCloud Stream整体结构核心概念图，如下所示：
别先生
2019/12/02
2K0
推荐阅读
编辑精选文章
万字详解高可用架构设计Go 开发者必备：Protocol Buffers 入门指南10分钟带你彻底搞懂分布式链路跟踪多租户的 4 种常用方案亿级月活的社交 APP，陌陌如何做到 3 分钟定位故障？60页PPT全解：DeepSeek系列论文技术要点整理
相关讨论
redis高可用使用springcloud连接不上？启动一个springcloud项目提示： WebMvcConfigurer.class找不到？添加stream 模块在配置文件里面配置stream后重启nginx提示不允许使用stream指令？
SpringCloud之Stream
3860
SpringCloud集成Stream
5180
消息驱动(SpringCloud Stream)
4590
SpringCloud Stream消息驱动
4600
Stream 消息驱动
4430
SpringCloud Stream消息驱动
3730
15-SpringCloud Stream
5700
springCloud --- 中级篇(3)
8480
springCloud - 第4篇 - 消费者调用服务 ( Feign )
5460
10SpringCloud Stream消息驱动
3590
Spring cloud stream【入门介绍】
1.3K0
微服务（十二）——Steam消息驱动&Sleuth链路监控
5050
SpringCloud Stream消息驱动代码实战
2450
SpringCloud Stream消息驱动
9180
springcloud : Stream消息驱动
7000
springcloud之Stream
1030
Spring Cloud 系列之消息驱动 Stream
1.5K0
SpringCloud 分布式配置中心
3490
springCloud - 第3篇 - 消费者调用服务 ( RestTemplate + Ribbon )
7300
RabbitMQ与Spring的框架整合之Spring Cloud Stream实战
2K0
相关推荐
SpringCloud之Stream
更多 >
目录
  • 一、消息驱动概述
    • ① 是什么
    • ② 设计思想
    • ③ Spring Cloud Stream 标准流程套路
    • ④ 编码 API 和常用注解
  • 二、案例说明
    • ① RabbitaMQ 环境已经 OK
    • ② 工程中新建三个子模块
  • 三、消息驱动之生产者
  • 四、消息驱动之消费者
  • 五、分组消费与持久化
    • ① 配置 8803
    • ② 启动
    • ③ 运行后两个问题
    • ④ 消费： 目前是 8802/8803 同时都收到了， 存在重复消费问题
    • ⑤ 分组
    • ⑥ 持久化
加入讨论
的问答专区 >
领券

腾讯云开发者

扫码关注腾讯云开发者

扫码关注腾讯云开发者

领取腾讯云代金券

热门产品

热门推荐

更多推荐

Copyright © 2013 - 2025 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 深公网安备号 44030502008569

腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号 | 京公网安备号11010802020287

问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档

Copyright © 2013 - 2025 Tencent Cloud.

All Rights Reserved. 腾讯云 版权所有

登录 后参与评论
2