备案控制台
探索云世界
开发者社区 弹性计算 云服务器ECS 文章 正文

标签继承实践--RocketMQ实例标签继承Topic或者Group

2022-06-08 699
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
云服务器 ECS，每月免费额度200元 3个月
推荐场景：
部署并使用Docker 快速搭建云上博客 搭建微信小程序
云服务器ECS，u1 2核4GB 1个月
推荐场景：
搭建2048小游戏 搭建turtle画布 搭建wiki知识库
简介： 基于运维编排将RocketMQ实例上的标签同步到实例所包含的topic 和 group 上。

需求

需要将RocketMQ实例上的标签同步到实例所包含的topic 和 group 上


实现原理

  • 查询RocketMQ实例标签
  • 是否继承topic
  • 将指定实例标签继承到topic
  • 是否继承group
  • 将指定实例标签继承到group

image.png


操作步骤

  • 先创建子模版 模板名称为Ons_topicExtendInstanceTags，具体脚本参考下面子模板代码。
  • 再创建子模板 模板名称为Ons_groupExtendInstanceTags，具体脚本参考下面子模板代码。
  • 最后创建父模版，并将父模版中的子模版执行名称，改成第1、2步里自己创建模版时的名称，如果使用上述名称就不需要修改。具体脚本参考下面父模板代码。
  • 创建执行
  • 选择需要继承的地域
  • 选择需要继承的标签
  • 选择需要继承资源类型
  • 选择相关key是否覆盖
  • 在 消息队列 RocketMQ 版控制台查看执行效果。

image.png



模板

子模版：根据实例Id以及其他条件更新Topic 的标签

模板名称：Ons_topicExtendInstanceTags

FormatVersion: OOS-2019-06-01
Description: 标签继承
Parameters:  regionId:    Type: String
    Description:      en: The id of region
      zh-cn: 地域ID
    Label:      en: Region
      zh-cn: 地域
    AssociationProperty: RegionId
    Default: '{{ ACS::RegionId }}'  instanceId:    Type: String
    Description:      en: The id of OnsInstance
      zh-cn: RocketMQ实例ID
    Label:      en: InstanceId
      zh-cn: RocketMQ实例ID
  tagKeys:    Type: List
    Description: 标签Key的集合
  isUpdate:    Type: Boolean
    Description: 是否覆盖
    Default: falseRamRole: ''Tasks:  - Name: getOnsInstanceAndTags
    Action: ACS::ExecuteAPI
    Description: ListTagResources
    Properties:      Service: ONS
      API: ListTagResources
      Parameters:        RegionId: '{{ regionId }}'        ResourceId:          - '{{instanceId}}'        ResourceType: instance
    Outputs:      onsTags:        Type: Json
        ValueSelector: '.TagResources | map(select( .TagKey | test("^(?!acs).*"))) | map(select(.TagKey as $tagKey | {{tagKeys}} | index($tagKey) >=0)) '  - Name: getTopicAndTags
    Action: ACS::ExecuteAPI
    Description: OnsTopicList
    Properties:      Service: Ons
      API: OnsTopicList
      Parameters:        RegionId: '{{ regionId }}'        InstanceId: '{{instanceId}}'    Outputs:      updateTopicAndTags:        Type: List
        ValueSelector: .Data.PublishInfoDo[] | {"Topic":.Topic,"Tags":.Tags}  - Name: tagResources
    Action: ACS::ExecuteAPI
    Description: TagTopics
    Properties:      Service: Ons
      API: TagResources
      Parameters:        RegionId: '{{ regionId }}'        ResourceId:          Fn::Jq:            - All
            - .Topic
            - '{{ACS::TaskLoopItem}}'        Tag:          Fn::If:            - '{{isUpdate}}'            - Fn::Jq:                - All
                - '.[] |{"Key":.TagKey, "Value": .TagValue} '                - '{{getOnsInstanceAndTags.onsTags}}'            - Fn::Jq:                - All
                - '.Tags.Tag as $topicTags | {{getOnsInstanceAndTags.onsTags}} | map(select(.TagKey as $tagKey | [$topicTags[].Key] | index($tagKey) <0)) |.[] |{"Key":.TagKey, "Value": .TagValue}'                - '{{ACS::TaskLoopItem}}'        ResourceType: topic
        InstanceId: '{{instanceId}}'    Outputs:      reqResult:        Type: Json
        ValueSelector: .RequestId
    Loop:      RateControl:        Mode: Concurrency
        MaxErrors: 100        Concurrency: 1      Items: '{{getTopicAndTags.updateTopicAndTags}}'      Outputs:        tagResult:          AggregateType: Fn::ListJoin
          AggregateField: reqResult

子模版：根据实例Id 以及输入条件更新Group 的标签

模板名称：Ons_groupExtendInstanceTags

FormatVersion: OOS-2019-06-01
Description: 标签继承
Parameters:  regionId:    Type: String
    Description:      en: The id of region
      zh-cn: 地域ID
    Label:      en: Region
      zh-cn: 地域
    AssociationProperty: RegionId
    Default: '{{ ACS::RegionId }}'  instanceId:    Type: String
    Description:      en: The id of OnsInstance
      zh-cn: 实例ID
    Label:      en: InstanceId
      zh-cn: 实例ID
  tagKeys:    Type: List
    Description: 标签Key的集合
  isUpdate:    Type: Boolean
    Description: 是否覆盖
    Default: falseRamRole: ''Tasks:  - Name: getOnsInstanceAndTags
    Action: ACS::ExecuteAPI
    Description: ListTagResources
    Properties:      Service: ONS
      API: ListTagResources
      Parameters:        RegionId: '{{ regionId }}'        ResourceId:          - '{{instanceId}}'        ResourceType: instance
    Outputs:      onsTags:        Type: Json
        ValueSelector: '.TagResources | map(select( .TagKey | test("^(?!acs).*"))) | map(select(.TagKey as $tagKey | {{tagKeys}} | index($tagKey) >=0)) '  - Name: getGroupAndTags
    Action: ACS::ExecuteAPI
    Description: OnsGroupList
    Properties:      Service: Ons
      API: OnsGroupList
      Parameters:        RegionId: '{{ regionId }}'        InstanceId: '{{instanceId}}'    Outputs:      updateGroupAndTags:        Type: List
        ValueSelector: .Data.SubscribeInfoDo[] | {"Group":.GroupId,"Tags":.Tags}  - Name: tagResources
    Action: ACS::ExecuteAPI
    Description: TagGroups
    Properties:      Service: Ons
      API: TagResources
      Parameters:        RegionId: '{{ regionId }}'        ResourceId:          Fn::Jq:            - All
            - .Group
            - '{{ACS::TaskLoopItem}}'        Tag:          Fn::If:            - '{{isUpdate}}'            - Fn::Jq:                - All
                - '.[] |{"Key":.TagKey, "Value": .TagValue} '                - '{{getOnsInstanceAndTags.onsTags}}'            - Fn::Jq:                - All
                - '.Tags.Tag as $groupTags | {{getOnsInstanceAndTags.onsTags}} | map(select(.TagKey as $tagKey | [$groupTags[].Key] | index($tagKey) <0)) |.[] |{"Key":.TagKey, "Value": .TagValue}'                - '{{ACS::TaskLoopItem}}'        ResourceType: group
        InstanceId: '{{instanceId}}'    Outputs:      reqResult:        Type: Json
        ValueSelector: .RequestId
    Loop:      RateControl:        Mode: Concurrency
        MaxErrors: 100        Concurrency: 1      Items: '{{getGroupAndTags.updateGroupAndTags}}'      Outputs:        tagResult:          AggregateType: Fn::ListJoin
          AggregateField: reqResult

父模版：按照输入条件同步某个地域下所有Ons 实例的Topic 和Group 的标签

模板名称：RocketMQInstanceTagInherit

FormatVersion: OOS-2019-06-01
Description: 标签继承
Parameters:  regionIds:    Type: List
    Description:      en: The id of region
      zh-cn: 地域ID
    Label:      en: Region
      zh-cn: 地域
    AssociationProperty: RegionId
  tags:    Type: List
    AssociationProperty: Tags
    AssociationPropertyMetadata:      ResourceType: ALIYUN::MQ::INSTANCE
    Description: 消息队列RocketMQ实例上的标签，选择需要继承到topic/group上的标签键及标签值
  resourceTypes:    Type: List
    Description: 所选择继承的资源类型
    AllowedValues:      - topic
      - group
  isUpdate:    Type: Boolean
    Description: 如果被继承的资源存在相同key但是value不相同时是否覆盖value
    Default: falseRamRole: ''Tasks:  - Name: getOnsInstance
    Action: ACS::ExecuteAPI
    Description: OnsInstanceInServiceList
    Properties:      Service: ONS
      API: OnsInstanceInServiceList
      Parameters:        RegionId: '{{ACS::TaskLoopItem}}'    Outputs:      onsInstance:        Type: List
        ValueSelector: '.Data.InstanceVO[] |. + {"RegionId": "{{ACS::TaskLoopItem}}" }'    Loop:      RateControl:        Mode: Concurrency
        MaxErrors: 100        Concurrency: 1      Items: '{{regionIds}}'      Outputs:        allOnsInstances:          AggregateType: Fn::ListJoin
          AggregateField: onsInstance
  - Name: isTagTopic
    Action: ACS::Choice
    Description:      en: Choose next task by resource type
      zh-cn: 根据资源类型选择下一个任务
    Properties:      DefaultTask: isTagGroup
      Choices:        - When:            Fn::Equals:              - true
              - Fn::Jq:                  - First
                  - contains(["topic"])
                  - '{{resourceTypes}}'          NextTask: tagTopic
  - Name: tagTopic
    Action: ACS::Template
    Description:      en: Update topic tags by ons instance tags
      zh-cn: 通过RocketMQ实例标签更新Topic的标签
    Properties:      TemplateName: Ons_topicExtendInstanceTags
      Parameters:        regionId:          Fn::Jq:            - First
            - .RegionId
            - '{{ACS::TaskLoopItem}}'        instanceId:          Fn::Jq:            - First
            - .InstanceId
            - '{{ACS::TaskLoopItem}}'        tagKeys:          Fn::Jq:            - All
            - .[].Key
            - '{{tags}}'        isUpdate: '{{isUpdate}}'    Loop:      RateControl:        Mode: Concurrency
        MaxErrors: 100        Concurrency: 1      Items:        Fn::Jq:          - All
          - .[][]          - '{{getOnsInstance.allOnsInstances}}'      Outputs:        tagResult:          AggregateType: Fn::ListJoin
          AggregateField: reqResult
  - Name: isTagGroup
    Action: ACS::Choice
    Description:      en: Choose next task by resource type
      zh-cn: 根据资源类型选择下一个任务
    Properties:      DefaultTask: ACS::END
      Choices:        - When:            Fn::Equals:              - true
              - Fn::Jq:                  - First
                  - contains(["group"])
                  - '{{resourceTypes}}'          NextTask: tagGroup
  - Name: tagGroup
    Action: ACS::Template
    Description:      en: Update snapshot tags by ecs instance tags
      zh-cn: 通过RocketMQ实例标签更新Group的标签
    Properties:      TemplateName: Ons_groupExtendInstanceTags
      Parameters:        regionId:          Fn::Jq:            - First
            - .RegionId
            - '{{ACS::TaskLoopItem}}'        instanceId:          Fn::Jq:            - First
            - .InstanceId
            - '{{ACS::TaskLoopItem}}'        tagKeys:          Fn::Jq:            - All
            - .[].Key
            - '{{tags}}'        isUpdate: '{{isUpdate}}'    Loop:      RateControl:        Mode: Concurrency
        MaxErrors: 100        Concurrency: 1      Items:        Fn::Jq:          - All
          - .[][]          - '{{getOnsInstance.allOnsInstances}}'      Outputs:        tagResult:          AggregateType: Fn::ListJoin
          AggregateField: reqResult
文章标签：
云消息队列 MQ
消息中间件
运维
RocketMQ
关键词：
云消息队列 MQ topic
云消息队列 MQ实例
云消息队列 MQ实践
云消息队列 MQ topic group
云消息队列 MQ group
相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv，使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
洛小蒙
目录
相关文章
三分钟热度的鱼
|
2月前
|
物联网
MQTT常见问题之用单片机接入阿里MQTT实例失败如何解决
MQTT（Message Queuing Telemetry Transport）是一个轻量级的、基于发布/订阅模式的消息协议，广泛用于物联网（IoT）中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总：
三分钟热度的鱼
52 1
讓丄帝愛伱
|
6月前
|
消息中间件 RocketMQ
RocketMQ报错：MQClientException：no route info of this topic的解决
RocketMQ报错：MQClientException：no route info of this topic的解决
讓丄帝愛伱
129 0
ftw2fzqaoykua
|
8月前
|
消息中间件 弹性计算 Java
Rocketmq-spring入门与实践
本场景带您体验如何在 Spring 生态中优雅地使用 Apache RocketMQ，感受最受欢迎业务开发框架与最受欢迎消息平台结合的魅力。
ftw2fzqaoykua
406 0
一百减一是零
|
20天前
|
消息中间件 监控 负载均衡
RocketMQ实践问题收集
RocketMQ实践问题收集
一百减一是零
27 0
三分钟热度的鱼
|
2月前
|
消息中间件 物联网 网络性能优化
MQTT常见问题之MQTT的topic超出上限25个如何解决
MQTT（Message Queuing Telemetry Transport）是一个轻量级的、基于发布/订阅模式的消息协议，广泛用于物联网（IoT）中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总：
三分钟热度的鱼
99 3
ftw2fzqaoykua
|
8月前
|
消息中间件 运维 Java
RocketMQ的常规运维实践应用
RocketMQ的常规运维实践应用
ftw2fzqaoykua
489 1
爱吃糖的范同学
|
4月前
|
消息中间件 网络架构
【面试问题】什么是 MQ topic 交换器（模式匹配） ？
【1月更文挑战第27天】【面试问题】什么是 MQ topic 交换器（模式匹配） ？
爱吃糖的范同学
65 0
极客李华
|
4月前
|
消息中间件 Java Spring
RabbitMQ各种模式的含义与Spring Boot实例详解
RabbitMQ各种模式的含义与Spring Boot实例详解
极客李华
39 0
sunrr
|
4月前
|
消息中间件 API RocketMQ
你的RocketMQ消费者组（Consumer Group）在查看时显示为离线，这可能是由于消费者组的状态没有被正确更新
你的RocketMQ消费者组（Consumer Group）在查看时显示为离线，这可能是由于消费者组的状态没有被正确更新【1月更文挑战第10天】【1月更文挑战第49篇】
sunrr
240 5
游客3bfkcht77micg
|
5月前
|
消息中间件 Java Kafka
RabbitMQ安装和5种不同的消息模型（BasicQueue,WorkQueue,Fanout Exchange,Direct Exchange,Topic Exchange）与SpringAMQP
RabbitMQ安装和5种不同的消息模型（BasicQueue,WorkQueue,Fanout Exchange,Direct Exchange,Topic Exchange）与SpringAMQP
游客3bfkcht77micg
48 1
弹性计算

云服务器ECS

热门文章

最新文章

  • 1
    惊呼！阿里云云服务器ECS最低3折售卖啦
  • 2
    ECS控制台使用小贴士
  • 3
    ECS 支持 IPv6 啦，快来尝鲜吧~
  • 4
    扫盲人工智能的计算力基石--异构计算
  • 5
    Mac神兵利器（四）时间管理工具
  • 6
    防止误删的神器-ECS实例删除保护
  • 7
    Mac 神兵利器（二） 极简软件清单
  • 8
    阿里公开核心技术：如何摘下4项世界冠军，推理性能比第二名快5倍
  • 9
    创建 ECS 支持使用镜像预设密码
  • 10
    双11技术攻略：企业云架构的正确姿势
  • 1
    阿里云目前活动内各云服务器实例规格适用场景与价格参考
    24
  • 2
    芯片研发设计单月算力成本降了50%怎么做到的？
    18
  • 3
    阿里云服务器香港地域租用收费标准参考
    26
  • 4
    阿里云服务器系统盘与数据盘ESSD AutoPL、ESSD云盘、SSD云盘区别及选择参考
    30
  • 5
    阿里云服务器租用价格参考，2核16G、4核32G、8核64G配置收费标准
    53
  • 6
    阿里云服务器租用价格参考，2核8G、4核16G、8核32G最新收费标准
    70
  • 7
    阿里云购买云服务器、注册域名、备案及绑定图文教程参考
    324
  • 8
    阿里云服务器租用价格参考，2核4G、4核8G、8核16G最新收费标准
    96
  • 9
    【技术分享】虚拟化
    80
  • 10
    阿里云服务器实例哪些属于入门级，哪些是企业级？有何区别？
    122

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列 MNS 入门课程
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • RocketMQ Client-GO 介绍
  • RocketMQ Prometheus Exporter 打造定制化 DevOps 平台
  • 基于 RocketMQ Prometheus Exporter 打造定制化 DevOps 平台

    • 相关实验场景

    更多
  • 消息队列RocketMQ版：消费异常运维排查体验
  • 部署RabbitMQ
  • RocketMQ运维自我实践
  • RocketMQ的常规运维实践应用
  • RocketMQ中使用Java客户端发送消息和消费的应用
  • RocketMQ一站式入门使用
    • 下一篇
    部署LAMP环境（Alibaba Cloud Linux 3）