AI 助理
文档备案控制台
开发者社区 弹性计算 云服务器ECS 文章 正文

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

2022-06-08 1125
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 基于运维编排将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
云消息队列 MQ实例topic
相关实践学习
快速体验阿里云云消息队列RocketMQ版
本实验将带您快速体验使用云消息队列RocketMQ版Serverless系列实例进行获取接入点、创建Topic、创建订阅组、收发消息、查看消息轨迹和仪表盘。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
洛小蒙
目录
相关文章
三分钟热度的鱼
|
消息中间件 Kubernetes RocketMQ
消息队列 MQ产品使用合集之topic是怎么选择分布在哪里brocker上面的
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
三分钟热度的鱼
250 0
小宋1021
|
消息中间件 存储 缓存
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
小宋1021
2197 8
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
猿java
|
消息中间件 负载均衡 算法
聊聊 RocketMQ中 Topic，Queue，Consumer，Consumer Group的关系
本文详细解析了RocketMQ中Topic、Queue、Consumer及Consumer Group之间的关系。文中通过图表展示了Topic可包含多个Queue，Queue分布在不同Broker上；Consumer组内多个消费者共享消息；并深入探讨了集群消费与广播消费模式下Queue与Consumer的关系，以及Rebalancing机制在实例增减时如何确保负载均衡。理解这些关系有助于更好地掌握RocketMQ的工作原理，提升系统运维效率。
猿java
3709 2
土木林森
|
消息中间件 开发者
【RabbitMQ深度解析】Topic交换器与模式匹配：掌握消息路由的艺术！
【8月更文挑战第24天】在消息队列（MQ）体系中，交换器作为核心组件之一负责消息路由。特别是`topic`类型的交换器，它通过模式匹配实现消息的精准分发，适用于发布-订阅模式。不同于直接交换器和扇形交换器，`topic`交换器支持更复杂的路由策略，通过带有通配符（如 * 和 #）的模式字符串来定义队列与交换器间的绑定关系。
土木林森
641 2
三分钟热度的鱼
|
消息中间件 存储 Java
消息队列 MQ使用问题之如何将RocketMQ中某个集群的topic迁移到另一个集群
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
三分钟热度的鱼
961 0
三分钟热度的鱼
|
消息中间件 Java API
消息队列 MQ产品使用合集之遇到"No topic route info in name server for the topic"错误，该如何处理
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
三分钟热度的鱼
2110 0
三分钟热度的鱼
|
消息中间件 Java 开发工具
消息队列 MQ产品使用合集之topic相同，但是tag不同，这个类不能放入map中，该如何处理
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
三分钟热度的鱼
261 0
君哥聊技术
|
消息中间件 算法 Java
弥补延时消息的不足，RocketMQ 基于时间轮算法实现了定时消息！
弥补延时消息的不足，RocketMQ 基于时间轮算法实现了定时消息！
君哥聊技术
1450 1
弥补延时消息的不足，RocketMQ 基于时间轮算法实现了定时消息！
君哥聊技术
|
消息中间件 uml RocketMQ
3 张图带你彻底理解 RocketMQ 事务消息
3 张图带你彻底理解 RocketMQ 事务消息
君哥聊技术
68260 2
3 张图带你彻底理解 RocketMQ 事务消息
君哥聊技术
|
消息中间件 Java uml
5张图带你理解 RocketMQ 顺序消息实现机制
5张图带你理解 RocketMQ 顺序消息实现机制
君哥聊技术
1344 1
5张图带你理解 RocketMQ 顺序消息实现机制
弹性计算

云服务器ECS

热门文章

最新文章

  • 1
    惊呼！阿里云云服务器ECS最低3折售卖啦
  • 2
    ECS控制台使用小贴士
  • 3
    玩转ECS云盘 — 按量付费升级到包年包月云盘
  • 4
    ECS 支持 IPv6 啦，快来尝鲜吧~
  • 5
    ECS 系统盘支持一键扩容啦，无需更换系统盘
  • 6
    VPC实例公网IP转弹性公网IP功能
  • 7
    【降价信息】弹性计算好“任性”，ECS又降价了~
  • 8
    云服务器ECS，你真的懂吗？
  • 9
    ECS 新版移动端购买发布啦！！！
  • 10
    阿里云基础产品技术月刊 2019年4月
  • 1
    阿里云38元轻量应用服务器详细介绍：配置、抢购规则、适用场景与选购攻略
    115
  • 2
    阿里云99元云服务器详细介绍：实例规格和配置、购买和续费规则、适用场景解析
    164
  • 3
    阿里云热门活动：云资源产品直降90%，新客首单38元起，“99 计划”新购续费同价
    134
  • 4
    阿里云组合购活动最新优惠：精选“云+AI”产品组合，轻量服务器+ TokenPlan 套餐68元起
    113
  • 5
    阿里云服务器租赁费用：新版租赁收费标准及活动报价参考
    1865
  • 6
    阿里云服务器2核8G/4核16G/8核32G如何选择实例？经济型e、通用算力型u2i与通用型g9i选哪个？
    184
  • 7
    2核4G/4核8G/8核16G阿里云服务器如何选择实例？经济型e、通用算力型u2i与计算型c9i选哪个？
    555
  • 8
    阿里云活动中心优惠权益解析：QoderWork CN首月0元，秒悟新注送1万积分，开通阿里云百炼享1亿+tokens
    148
  • 9
    阿里云AI产品与云产品最新组合套餐：Token Plan、AI coding及云服务器和建站等组合优惠价
    763
  • 10
    阿里云服务器实例规格怎么选？经济型e、通用算力型u1/u2i/u2a、高性能c9i/g9i/r9i对比与选择参考
    163

    • 相关课程

    更多
  • 消息队列 RocketMQ 全类型业务消息学习课程
  • 消息队列 RocketMQ 消息集成
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • 行业实践：RocketMQ 业务集成典型行业应用和实践
  • 技术揭秘：RocketMQ 5.0 云原生架构升级之路
  • RocketMQ Summit 2022 开源生态发展
    • 下一篇
    一条命令迁移，帮你实现 OpenClaw 与 Hermes Agent 记忆互通！