前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Stream 消息驱动

Stream 消息驱动

作者头像
BUG弄潮儿
发布2021-09-10 15:35:57
3490
发布2021-09-10 15:35:57
举报
文章被收录于专栏:JAVA乐园

一、什么是Spring Cloud Stream?

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

二、Stream的设计思想

1、标准MQ

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

2、为什么用Cloud Stream?

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

这些中间件的差异性导致我们实际项目开发给我们造成了一定的困扰,我们如果用了两个消息队列的其中一种,后面的业务需求,我想往另外一种消息队列进行迁移,这时候无疑就是一个灾难性的,一大堆东西都要重新推倒重新做,因为它跟我们的系统耦合了,这时候Spring Cloud Stream给我们提供了—种解耦合的方式。

3、Stream凭什么可以统一底层差异?

在没有绑定器这个概念的情况下,我们的SpringBoot应用要直接与消息中间件进行信息交互的时候,由于各消息中间件构建的初衷不同,它们的实现细节上会有较大的差异性通过定义绑定器作为中间层,完美地实现了应用程序与消息中间件细节之间的隔离。通过向应用程序暴露统一的Channel通道,使得应用程序不需要再考虑各种不同的消息中间件实现。

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

Binder

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

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

Topic主题进行广播

  • 在RabbitMQ就是Exchange
  • 在Kakfa中就是Topic

三、Stream编码常用注解简介

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

四、Stream之消息重复消费

运行后有两个问题

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

消费

  • http://localhost:8801/sendMessage
  • 目前是8802/8803同时都收到了,存在重复消费问题
  • 如何解决:分组和持久化属性group(重要)
生产实际案例

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

注意在Stream中处于同一个group中的多个消费者是竞争关系,就能够保证消息只会被其中一个应用消费一次。不同组是可以全面消费的(重复消费)。

五、Stream之group解决消息重复消费

1. 原理

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

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

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

group: A_Group、B_Group

六、Stream之消息持久化

  • 通过上述,解决了重复消费问题,再看看持久化。
  • 停止8802/8803并去除掉8802的分组group: A_Group,8803的分组group: A_Group没有去掉。
  • 8801先发送4条消息到RabbitMq。
  • 先启动8802,无分组属性配置,后台没有打出来消息。
  • 再启动8803,有分组属性配置,后台打出来了MQ上的消息。(消息持久化体现)

source:https://www.yuque.com/yanzipang-wf7ur/hkyrfw/vbkxz8

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2021-08-17,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 BUG弄潮儿 微信公众号,前往查看

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

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

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 一、什么是Spring Cloud Stream?
  • 二、Stream的设计思想
    • 1、标准MQ
      • 2、为什么用Cloud Stream?
        • 3、Stream凭什么可以统一底层差异?
          • 4、通过定义绑定器Binder作为中间层,实现了应用程序与消息中间件细节之间的隔离。
            • 1. 原理
        • 三、Stream编码常用注解简介
        • 四、Stream之消息重复消费
        • 五、Stream之group解决消息重复消费
        • 六、Stream之消息持久化
        相关产品与服务
        消息队列 CMQ 版
        消息队列 CMQ 版(TDMQ for CMQ,简称 TDMQ CMQ 版)是一款分布式高可用的消息队列服务,它能够提供可靠的,基于消息的异步通信机制,能够将分布式部署的不同应用(或同一应用的不同组件)中的信息传递,存储在可靠有效的 CMQ 队列中,防止消息丢失。TDMQ CMQ 版支持多进程同时读写,收发互不干扰,无需各应用或组件始终处于运行状态。
        领券
        问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档