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

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

2022-06-08 727
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 基于运维编排将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
云消息队列 MQ group
云消息队列 MQ实例
云消息队列 MQ topic group
相关实践学习
消息队列RocketMQ版：基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
洛小蒙
目录
相关文章
hhzz
|
1月前
|
消息中间件 存储 Serverless
【实践】快速学会使用阿里云消息队列RabbitMQ版
云消息队列 RabbitMQ 版是一款基于高可用分布式存储架构实现的 AMQP 0-9-1协议的消息产品。云消息队列 RabbitMQ 版兼容开源 RabbitMQ 客户端，解决开源各种稳定性痛点（例如消息堆积、脑裂等问题），同时具备高并发、分布式、灵活扩缩容等云消息服务优势。
hhzz
92 2
最好zzz
|
2月前
|
消息中间件 Java Apache
RocketMQ消息回溯实践与解析
在分布式系统和高并发应用的开发中，消息队列扮演着至关重要的角色，而RocketMQ作为阿里巴巴开源的一款高性能消息中间件，以其高吞吐量、高可用性和灵活的配置能力，在业界得到了广泛应用。本文将围绕RocketMQ的消息回溯功能进行实践与解析，分享工作学习中的技术干货。
最好zzz
78 3
郑小健
|
3月前
|
消息中间件 弹性计算 Kubernetes
RabbitMQ与容器化技术的集成实践
【8月更文第28天】RabbitMQ 是一个开源消息代理和队列服务器，用于在分布式系统中存储、转发消息。随着微服务架构的普及，容器化技术（如 Docker 和 Kubernetes）成为了部署和管理应用程序的标准方式。本文将探讨如何使用 Docker 和 Kubernetes 在生产环境中部署和管理 RabbitMQ 服务，同时保证高可用性和弹性伸缩能力。
郑小健
68 3
DreamSpark
|
1月前
|
消息中间件 安全 Java
云消息队列RabbitMQ实践解决方案评测
一文带你详细了解云消息队列RabbitMQ实践的解决方案优与劣
DreamSpark
71 9
开发者评测
|
28天前
|
消息中间件
解决方案 | 云消息队列RabbitMQ实践获奖名单公布！
云消息队列RabbitMQ实践获奖名单公布！
开发者评测
54 1
aliyun7689123603-22772
|
1月前
|
消息中间件 存储 弹性计算
云消息队列RabbitMQ实践
云消息队列RabbitMQ实践
aliyun7689123603-22772
68 4
不游泳的鱼鱼
|
1月前
|
消息中间件 存储 弹性计算
云消息队列 RabbitMQ 版实践解决方案评测
随着企业业务的增长，对消息队列的需求日益提升。阿里云的云消息队列 RabbitMQ 版通过架构优化，解决了消息积压、内存泄漏等问题，并支持弹性伸缩和按量计费，大幅降低资源和运维成本。本文从使用者角度详细评测这一解决方案，涵盖实践原理、部署体验、实际优势及应用场景。
不游泳的鱼鱼
76 2
叶一一
|
1月前
|
消息中间件 存储 监控
解决方案 | 云消息队列RabbitMQ实践
在实际业务中，网站因消息堆积和高流量脉冲导致系统故障。为解决这些问题，云消息队列 RabbitMQ 版提供高性能的消息处理和海量消息堆积能力，确保系统在流量高峰时仍能稳定运行。迁移前需进行技术能力和成本效益评估，包括功能、性能、限制值及费用等方面。迁移步骤包括元数据迁移、创建用户、网络打通和数据迁移。
叶一一
64 4
小宋1021
|
2月前
|
消息中间件 存储 缓存
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
小宋1021
245 7
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
猿java
|
1月前
|
消息中间件 负载均衡 算法
聊聊 RocketMQ中 Topic，Queue，Consumer，Consumer Group的关系
本文详细解析了RocketMQ中Topic、Queue、Consumer及Consumer Group之间的关系。文中通过图表展示了Topic可包含多个Queue，Queue分布在不同Broker上；Consumer组内多个消费者共享消息；并深入探讨了集群消费与广播消费模式下Queue与Consumer的关系，以及Rebalancing机制在实例增减时如何确保负载均衡。理解这些关系有助于更好地掌握RocketMQ的工作原理，提升系统运维效率。
猿java
125 2
弹性计算

云服务器ECS

热门文章

最新文章

  • 1
    ECS 系统盘支持一键扩容啦，无需更换系统盘
  • 2
    云服务器ECS安全组实践(一）
  • 3
    云服务器ECS，你真的懂吗？
  • 4
    安全组设置内网互通的方法
  • 5
    扫盲人工智能的计算力基石--异构计算
  • 6
    Mac神兵利器（四）时间管理工具
  • 7
    安全组规则批量添加授权对象
  • 8
    防止误删的神器-ECS实例删除保护
  • 9
    使用OpenApi弹性管理云服务器ECS
  • 10
    按量付费实例批量更改实例带宽
  • 1
    阿里云服务器计算型c8i、通用型g8i和计算型c8a、通用型g8a实例区别及选择参考
    11
  • 2
    阿里云上云礼包、区域补贴、算力补贴、上云抵扣金、无门槛等优惠券区别参考
    13
  • 3
    阿里云服务器、轻量应用服务器、gpu云服务器收费标准与实时活动价格参考
    19
  • 4
    ubuntu和debian 的安装包dpkg管理命令对安装包进行安装，查询，卸载
    23
  • 5
    预告 | 阿里云邀您参加2024中国生成式AI大会上海站，马上报名
    167
  • 6
    阿里云服务器计算型c8i实例解析：实例规格性能及使用场景和最新价格参考
    22
  • 7
    阿里云通用算力型U1实例性能、适用场景、与经济型e区别、收费标准参考
    27
  • 8
    阿里云GPU云服务器性能、应用场景及收费标准和活动价格参考
    33
  • 9
    阿里云服务器通用型g8i实例性能与使用场景介绍及最新收费标准参考
    30
  • 10
    阿里云服务器经济型e和通用算力型u1实例区别及选择参考
    26

    • 相关课程

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

    • 相关电子书

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

    • 相关实验场景

    更多
  • 消息队列RocketMQ版：消费异常运维排查体验
  • 消息队列RocketMQ版：定时消息通知功能体验
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 部署RabbitMQ
  • RocketMQ中使用Java客户端发送消息和消费的应用
  • RocketMQ一站式入门使用
    • 下一篇
    无影云桌面