云服务分类 云服务名称 产品控制台创建资源时是否支持绑定标签 产品控制台列表是否支持绑定和解绑标签 标签字符长度限制 标签字符内容限制 单实例默认可绑定标签个数 计算 弹性云主机 是 是 128位 首尾不包含空格 50 计算 GPU云主机 是 是 128位 首尾不包含空格 50 计算 物理机 是 是 128位 首尾不包含空格 50 计算 镜像服务 是 是 128位 首尾不包含空格 50 计算 弹性伸缩服务 是 是 128位 首尾不包含空格 50 计算 云主机快照 是 是 128位 首尾不包含空格 50 计算 SSH秘钥对 是 是 128位 首尾不包含空格 50 存储 云硬盘 是 是 128位 开头不包含空格 50 存储 弹性文件服务 否 是 128位 首尾不包含空格 50 存储 对象存储 是 是 128位 首尾不包含空格 50 存储 并行文件服务HPFS 是 是 128位 首尾不包含空格 50 存储 海量文件服务OceanFS 是 是 128位 首尾不包含空格 50 存储 云硬盘备份 是 是 128位 开头不包含空格 50 存储 云主机备份 是 是 128位 开头不包含空格 50 网络 弹性负载均衡 是 是 128位 首尾不包含空格 50 网络 共享流量包 否 是 128位 首尾不包含空格 50 网络 VPC终端节点 是 是 128位 首尾不包含空格 50 网络 NAT网关 是 是 128位 首尾不包含空格 50 网络 网关负载均衡 是 是 128位 首尾不包含空格 50 网络 内网DNS 否 是 128位 首尾不包含空格 50 网络 弹性IP 否 是 128位 首尾不包含空格 50 网络 共享带宽 否 是 128位 首尾不包含空格 50 网络 虚拟私有云 否 是 128位 首尾不包含空格 50 网络 流量镜像 否 是 128位 首尾不包含空格 50 网络 对等连接 是 是 128位 首尾不包含空格 50 网络 云间高速（标准版） 否 是 128位 首尾不包含空格 50 网络 VPN连接 否 是 128位 首尾不包含空格 50 网络 云专线CDA 是 是 128位 首尾不包含空格 50 专属云 专属云（计算独享型） 是 是 128位 首尾不包括空格 50 云原生 云容器引擎 是 是 128位 首尾不包含空格 50 云原生 容器镜像服务 否 是 128位 首尾不包含空格 50 云原生 云日志服务 否 是 128位 首尾不包含空格 50 云原生 应用性能监控 否 是 128位 首尾不包含空格 20 云原生 微服务云应用平台MSAP 否 是 128位 首尾不包含空格 50 云原生 微服务引擎API网关 否 是 128位 首尾不包含空格 20 云原生 微服务引擎微服务治理 否 是 128位 首尾不包含空格 20 云原生 微服务引擎注册配置中心 否 是 128位 首尾不包含空格 20 云原生 服务网格 否 是 128位 首尾不包含空格 50 云原生 分布式缓存服务Redis版 是 是 128位 首尾不包含空格 50 云原生 分布式消息队列RocketMQ 是 是 128位 首尾不包含空格 50 云原生 分布式消息队列RabbitMQ 是 是 128位 首尾不包含空格 50 云原生 分布式消息队列Kafka 是 是 128位 首尾不包含空格 50 云原生 分布式消息队列MQTT 是 是 128位 首尾不包含空格 50 云原生 弹性容器实例ECI 否 是 128位 首尾不包含空格 50 云原生 分布式容器云平台CCE ONE 是 是 128位 首尾不包含空格 20 云原生 Serverless容器引擎 是 是 128位 首尾不包含空格 50 云原生 函数计算 否 是 128位 首尾不包含空格 50 安全及管理 Web应用防火墙（原生版） 否 是 无限制 首尾不包含空格 50 安全及管理 服务器安全卫士（原生版） 否 是 128位 首尾不包含空格 10 安全及管理 云等保专区 否 是 128位 首尾不包含空格 50 安全及管理 数据库审计 否 是 128位 首尾不包含空格 10 安全及管理 云堡垒机（原生版） 否 是 128位 首尾不包含空格 10 安全及管理 日志审计（原生版） 否 是 128位 首尾不包含空格 10 安全及管理 秘钥管理 否 是 128位 首尾不包含空格 10 安全及管理 云密评专区 否 是 128位 首尾不包含空格 10 数据库 文档数据库服务 是 是 128位 首尾不包含空格 50 数据库 分布式关系型数据库 是 是 128位 首尾不包含空格 10 数据库 关系数据库PostgreSQL版 是 是 128位 首尾不包含空格 50 数据库 关系数据库MySQL版 是 是 128位 首尾不包含空格 50 数据库 云数据库ClickHouse版 是 是 128位 首尾不包含空格 50 数据库 关系数据库SQL Server版 是 是 128位 首尾不包含空格 50 数据库 数据传输服务DTS 否 是 128位 首尾不包含空格 50

开源对比 相较于开源自建Kafka，分布式消息服务Kafka在低成本运维、分区规模、消息查询、ACL访问控制、可视化配置、运维监控、集群巡检、稳定可靠、安全保证、简单易用等方面更具优势。更多信息请参见开源对比。 产品优势 分布式消息服务Kafka具备高可用性、高安全性、可靠性、全托管等优势，使其成为大规模数据处理和实时流处理的理想选择。更多信息请参见产品优势。 功能特性 分布式消息服务Kafka的功能特性主要体现在以下几个方面： 消息能力 广播消息：在同一个消费组内对所有消费者投递相同消息。 消息回溯：支持根据时间重置消费进度。 消息数据自动删除功能：在磁盘满后，在保护期外的数据，能自动删除，保证服务可用性。 自动故障切换功能：生产消费自动负载均衡，消息节点故障时自动主备切换，保证服务的连续性。 队列能力 高吞吐，消息多副本异步复制。 高可靠，消息多副本同步复制。 可视化管理 应用用户管理：多个应用可调用同一个消息服务，通过应用用户，对消息服务下的应用接入权限进行管理。 主题管理：支持对实例下的主题进行管理，执行创建删除等操作。 消费组管理：支持对实例下的消费组进行管理。 Broker监控：提供Broker详细信息以及多维度的监控指标查看。 Topic监控：提供Topic详细信息以及多维度的监控指标查看。

本文介绍如何应用事件总线EventBridge的事件流功能实现分布式消息服务Kafka的消息路由。 前提条件 开通事件总线EventBridge并授权。 开通分布式消息服务Kafka并创建最少两个主题。 背景信息 事件流作为更轻量、实时端到端的流式事件通道，提供轻量级的流式数据的过滤和转换的能力，在不同的数据仓库之间、数据处理程序之间、数据分析和处理系统之间进行数据同步。源端分布式消息服务Kafka生产的消息可以通过事件流这个通道被路由到目标端的分布式消息服务Kafka。 步骤一：创建事件流 1. 登录事件总线EventBridge控制台。 2. 在左侧导航栏，单击事件流。 3. 在事件流页面，单击创建事件流。 4. 在创建事件流面板，设置任务名称和描述，配置以下参数，然后单击保存。 a.在Source（源）配置向导，选择数据提供方为分布式消息服务Kafka ，设置以下参数，然后单击下一步。 参数 说明 示例 Kafka实例 选择Kafka实例。 instancexxx Kafka Topic 选择Kafka topic。 topicxxx Group 选择消费组。 快速创建 消费位点 选择消费位点。 最新位点 b.在Filtering（过滤）配置向导，设置事件过滤规则，单击下一步。 c.在Sink（目标）配置向导，选择服务类型为分布式消息服务Kafka，配置参数，单击保存，如图1所示。 参数 说明 示例 Kafka实例 选择Kafka实例。 instancexxx Kafka Topic 选择Kafka topic。 topicxxx 消息体（value） 提取事件中的数据。 $.data.value 消息键值（key） 提前事件中的key到目标。 $.data.key 图1 创建事件流时目标类型选择分布式消息服务Kafka 5. 创建事件流后，会有30秒~60秒的延迟时间，您可以在事件流页面的状态栏查看启动进度。

本文介绍如何在事件总线EventBridge管理控制台添加分布式消息Kafka类型的自定义事件源。 前提条件 事件总线EventBridge 开通事件总线EventBridge并委托授权。 创建自定义总线。 分布式消息服务Kafka 开通分布式消息服务Kafka。 创建实例、主题与消费组。 操作步骤 1. 登录事件总线管理控制台。 2. 在左侧导航栏，单击事件总线。 3. 在事件总线页面，单击目标总线名称。 4. 在左侧导航栏，单击事件源。 5. 在事件源 页面，单击添加事件源。 6. 在添加自定义事件源 面板，输入名称 和描述 ，事件提供方 选择分布式消息服务Kafka，选择实例、主题等配置然后单击确认，详见图1。 图1 添加事件提供方为分布式消息服务Kafka的事件源 参数说明 参数 说明 示例 名称 事件源名。 source1 实例名称 选择前提条件中已创建的分布式消息服务Kafka版实例。 KAFKAxxx Topic 选择当前实例中的Topic。 topic1 Group 快速创建：自动创建以GIDEVENTBRIDGExxx命名的Group ID。 使用已有：选择当前实例中已创建的Group，请不要与已有业务的Group混用，以免影响已有的消息收发。 group1 消费位点 开始消费的位置。 最新位点：从最新位点开始消费。 最早位点：从最初位点开始消费。 最新位点

枚举参数 无 请求示例 请求url 请求头header { "regionId":"bb9fdb42056f11eda1610242ac110002" } 请求体body 无 响应示例 { "statusCode": 800, "returnObj": { "clientid": "mqtt1", "keepalive": 60, "mqueuelen": 0, "maxinflight": 10, "createdat": "20250101 00:00:00", "connectedat": "20250101 00:00:00", "sendmsg": 10, "mqueuedropped": 0, "recvmsg": 10, "ipaddress": "33.0.0.1", "port": 30000, "username": "test", "cleanstart": true } } 状态码 请参考 状态码 错误码 请参考 错误码

本节介绍客户端首次接入分布式消息服务Kafka时出现异常的排查方法 客户端首次接入分布式消息服务Kafka时如果出现异常，您可以通过以下三个方面进行排查： 网络连通问题：通常有如下几种情况可能会导致网络不通，客户端无法连接分布式消息服务Kafka的服务端。 （1）ECS实例（客户端）与分布式消息服务Kafka实例不在同一个VPC内。关于如何通过VPC网络接入服务端，请参见VPC接入。 （2）本地客户端访问非公网的分布式消息服务Kafka实例。您可以通过公网接入服务端。 客户端版本问题：客户端版本太低可能会出现异常，建议升级到与服务端匹配的客户端版本。 配置问题：默认接入点配置错误等都可能导致接入失败。您可以参见快速入门进行配置检查。

本文主要介绍 产品定义。 Kafka是一个拥有高吞吐、可持久化、可水平扩展，支持流式数据处理等多种特性的分布式消息流处理中间件，采用分布式消息发布与订阅机制，在日志收集、流式数据传输、在线/离线系统分析、实时监控等领域有广泛的应用。 分布式消息服务Kafka是一款基于开源社区版Kafka提供的消息队列服务，向用户提供计算、存储和带宽资源独占式的Kafka专享实例。使用分布式消息服务Kafka，资源按需申请，即买即用，您将有更多精力专注于业务快速开发，不用考虑部署和运维。 关于Kafka的帮助手册阅读指引 受限于篇幅，我们提供的Kafka帮助手册重点描述产品相关的内容，以及与开源社区版Kafka的差异部分，例如Kafka的产品规格、控制台操作、客户端对接等。 如果您需要了解Kafka入门知识或消息生产、消费等方面的技术细节，请查阅Kafka官网资料。

本指南主要介绍RabbitMQ实例连接信息的收集，如获取RabbitMQ实例连接地址与端口、访问实例的用户名和密码，然后提供Python语言和Spring Boot的连接示例。 RabbitMQ实例完全兼容开源RabbitMQ协议，Python以外的语言，请参考RabbitMQ官网提供的不同语言的连接和使用向导。 开源SDK列表 分布式消息服务RabbitMQ版支持所有开源版本的SDK，常见的开源SDK如下表所示。 开源SDK列表 编程语言 SDK :: Java rabbitmqjavaclient Spring Framework SpringAMQP .Net rabbitmqdotnetclient Python pika PHP phpamqplib C rabbitmqc Go amqp091go 客户端可以通过以下方式访问RabbitMQ实例： 1. VPC内子网地址访问 如果客户端与RabbitMQ实例处于同region同VPC，则可以直接访问RabbitMQ实例提供的VPC内子网地址。 2. VPC对等连接方式访问 如果客户端与RabbitMQ实例处于相同region但不同VPC，则可以通过建立VPC对等连接后，访问RabbitMQ实例提供的VPC内子网地址。 关于创建和使用VPC对等连接，可参考对等连接。 3. 公网访问 客户端在其他网络环境，或者与RabbitMQ实例处于不同region，则访问实例的公网地址。

事件目标是事件规则的一部分，负责消费经事件规则过滤与转换后的事件。 事件目标 事件总线EventBridge支持以下事件目标： 函数计算 分布式消息服务RocketMQ 分布式消息服务RabbitMQ 分布式消息服务Kafka HTTP、HTTPS地址

模块 参数 参数说明 配置示例 备注 配置订阅渠道 网络通道 公网/vpc内网 公网 配置订阅渠道 订阅类型 公网：分布式消息服务/remotewriteAPI/API vpc内网:分布式消息服务 分布式消息服务 配置订阅渠道 类型 对应订阅类型下选择支持的类型 分布式消息服务Kafka 配置订阅渠道 地址 填写分布式消息服务/remotewriteAPI/API目标地址 配置订阅渠道 实例选择 下拉选择所创建的kafka实例（前提：“网络通道”选择“vpc内网”） 配置订阅渠道 用户名 目标地址用户名信息 test 配置订阅渠道 密码 目标地址用户名对应密码信息 配置订阅渠道 TOPIC 目标地址主题（TOPIC） 配置订阅渠道 kafka版本 选择目标客户端kafka版本，2.0.0/2.1.0 2.1.0 2.0.0及以下版本，请选择2.0.0 2.0.0以上版本，请选择2.1.0 基本信息 名称 填写订阅渠道名称 test 基本信息 描述 填写订阅渠道描述信息

模块 参数 参数说明 配置示例 备注 配置订阅渠道 网络通道 公网/vpc内网 公网 配置订阅渠道 订阅类型 公网：分布式消息服务/remotewriteAPI/API vpc内网:分布式消息服务 分布式消息服务 配置订阅渠道 类型 对应订阅类型下选择支持的类型 分布式消息服务Kafka 配置订阅渠道 地址 填写分布式消息服务/remotewriteAPI/API目标地址 配置订阅渠道 实例选择 下拉选择所创建的kafka实例（前提：“网络通道”选择“vpc内网”） 配置订阅渠道 用户名 目标地址用户名信息 test 配置订阅渠道 密码 目标地址用户名对应密码信息 配置订阅渠道 TOPIC 目标地址主题（TOPIC） 配置订阅渠道 kafka版本 选择目标客户端kafka版本，2.0.0/2.1.0 2.1.0 2.0.0及以下版本，请选择2.0.0 2.0.0以上版本，请选择2.1.0 基本信息 名称 填写订阅渠道名称 test 基本信息 描述 填写订阅渠道描述信息

本章节主要介绍分布式消息服务RabbitMQ的自定义策略。 如果系统预置的分布式消息服务RabbitMQ权限，不满足您的授权要求，可以创建自定义策略。 目前天翼云支持以下两种方式创建自定义策略： 可视化视图创建自定义策略：无需了解策略语法，按可视化视图导航栏选择云服务、操作、资源、条件等策略内容，可自动生成策略。 JSON视图创建自定义策略：可以在选择策略模板后，根据具体需求编辑策略内容；也可以直接在编辑框内编写JSON格式的策略内容。 具体创建步骤请参见：《统一身份认证服务用户指南》的“创建自定义策略”章节。本章为您介绍常用的分布式消息服务RabbitMQ自定义策略样例。 说明 分布式消息服务RabbitMQ服务的权限与策略基于分布式消息服务DMS，因此在IAM服务中为RabbitMQ分配用户与权限时，请选择并使用“DMS”的权限与策略。 分布式消息服务RabbitMQ自定义策略样例 示例1：授权用户删除实例和重启实例 { "Version": "1.1", "Statement": [ { "Effect": "Allow", "Action": [ "dms:instance:modifyStatus", "dms:instance:delete" ] } ] } 示例2：拒绝用户删除实例 拒绝策略需要同时配合其他策略使用，否则没有实际作用。用户被授予的策略中，一个授权项的作用如果同时存在Allow和Deny，则遵循Deny优先。 如果您给用户授予DMS FullAccess的系统策略，但不希望用户拥有DMS FullAccess中定义的删除实例权限，您可以创建一条拒绝删除实例的自定义策略，然后同时将DMS FullAccess和拒绝策略授予用户，根据Deny优先原则，则用户可以对DMS for RabbitMQ执行除了删除实例外的所有操作。拒绝策略示例如下： { "Version": "1.1", "Statement": [ { "Effect": "Deny", "Action": [ "dms:instance:delete" ] } ] }

本节介绍Group不存在但能消费消息原因 我在分布式消息服务Kafka控制台上，未查看到对应的Group，但此Group下却有消费线程在消费消息。 可能原因 如果客户端使用assign方式消费消息，那么即使不创建Group，也可能消费消息。 如果客户端使用subscribe方式消费消息，删除Group后，消费线程未停止或者未发生Rebalance，那么消费线程还可以继续正常消费。 解决方案 如果客户端使用assign方式消费消息，请提前在分布式消息服务Kafka控制台创建Group。 请尽量复用Group，避免创建过多的Group而影响集群的稳定性。 在删除Group前，请确保已停止该Group下的所有消费线程。

本章节介绍了如何创建和运行一个分布式消息服务RocketMQ的实例。 本章节以在“广州4”创建一个开启SSL的RocketMQ 实例为例，介绍如何在控制台创建RocketMQ 实例。 前提条件 在创建RocketMQ实例前，需要保证存在可使用的虚拟私有云和安全组。如果还没有，请参考准备环境创建。 为RocketMQ授权 如果RocketMQ实例需要开启IPv6功能，在创建RocketMQ实例前，需要为RocketMQ授予VPC终端节点操作权限，否则会导致创建失败。 在IAM控制台创建委托，委托参数如下，其他参数保持默认，创建委托的具体步骤请参见《统一身份认证服务 用户指南》的“委托 > 委托其他云服务管理资源”章节。 委托名称：dmsadmintrust 委托类型：云服务 云服务：分布式消息服务 持续时间：永久 委托将拥有所选策略：DMS VPCEndpointAccess、DMSAgencyCheckAccessPolicy 选择授权范围方案如下： DMS VPCEndpointAccess：指定区域项目资源 DMSAgencyCheckAccessPolicy：所有资源 操作步骤 1. 进入购买RocketMQ实例页面。 2. “计费模式”选择“按需计费”。 3. 在“区域”下拉列表中，选择靠近您应用程序的区域，可降低网络延时、提高访问速度。 4. 在“项目”下拉列表中，选择项目。 5. 在“可用区”区域，根据实际情况选择1个或者3个及以上可用区。 6. 设置“实例名称”和“企业项目”。 7. 设置实例信息，配置详情请参考表1。 表1 设置实例信息 参数 配置说明 规格选择模式 选择“默认” 版本 可选择“5.x、4.8.0” RocketMQ实例创建后，版本号不支持修改。 实例类型 5.x版本提供基础版、专业版两种类型 部署架构 5.x版本支持“单机”和“集群” 4.8.0支持“集群” 鲲鹏实例 “创建鲲鹏架构实例”选框，默认不勾选是X86架构的实例，勾选之后是ARM架构的鲲鹏实例。 代理规格 选择“rocketmq.4u8g.cluster” 代理数量 选择“1” 代理存储空间 选择“超高I/O 300GB” 实例总存储空间 单个代理的存储空间 代理数量 8. 设置实例网络环境信息，配置详情请参考表2。 参数 配置说明 虚拟私有云 选择已经创建好的虚拟私有云和子网。 虚拟私有云和子网在RocketMQ实例创建完成后，不支持修改。 IPv6 仅当子网选择了支持IPv6地址后，才会显示该参数。开启IPv6后，客户端可以使用IPv6的内网连接地址连接实例。 实例创建成功后，不支持修改IPv6开关。 使用内网IPv6方式连接实例需要通过VPC终端节点实现，使用期间会产生VPC终端节点的费用。 安全组 选择已经创建好的安全组。 SSL 开启 ACL访问控制 开启 9. 单击“更多配置”，设置更多相关信息，配置详情请参考表3。 表3 更多配置 参数 配置说明 公网访问 不开启 标签 不设置 描述 不设置 10. 填写完上述信息后，单击“立即购买”，进入规格确认页面。 11. 确认实例信息无误后，提交请求。 12. 在实例列表页面，查看RocketMQ实例是否创建成功。 创建实例大约需要3到15分钟，此时实例的“状态”为“创建中”。 当实例的“状态”变为“运行中”时，说明实例创建成功。 当实例的“状态”变为“创建失败”，请删除创建失败的实例，然后重新创建。如果重新创建仍然失败，请联系客服。

Kafka是一个拥有高吞吐、可持久化、可水平扩展，支持流式数据处理等多种特性的分布式消息流处理中间件，采用分布式消息发布与订阅机制，在日志收集、流式数据传输、在线/离线系统分析、实时监控等领域有广泛的应用。 分布式消息服务Kafka是一款基于开源社区版Kafka提供的消息队列服务，向用户提供计算、存储和带宽资源独占式的Kafka专享实例。使用分布式消息服务Kafka，资源按需申请，按需配置Topic的分区与副本数量，即买即用，您将有更多精力专注于业务快速开发，不用考虑部署和运维。

请求示例 请求url 请求头header { "regionId":"bb9fdb42056f11eda1610242ac110002" } 请求体body 无 响应示例 { "statusCode": 800, "returnObj": [ { "clientId": "mqtt1", "time": "20250101 00:00:00", "action": "connected" } ] } 状态码 请参考 状态码 错误码 请参考 错误码

本章节介绍Kafka Broker节点主机宕机故障演练。 背景介绍 高性能高可靠的分布式消息服务 Kafka 在复杂分布式环境中仍可能因 Broker 节点宕机引发数据丢失、集群可用性下降、请求延迟升高、副本同步滞后等严重问题，本演练可测试业务系统应对此类核心组件故障的响应能力、高可用切换机制及数据一致性保障效果。 基本原理 指定或随机一个Broker节点进行关机。 故障注入 1、纳管实例资源 1. 导航至 故障演练 > 目标应用 > 应用资源页面。 2. 在资源类型页签中选择分布式消息服务Kafka，然后单击添加资源。 3. 在弹出的对话框中，勾选目标分布式消息服务Kafka实例，单击确定。 2、编排演练任务 1. 导航至 故障演练 > 目标应用 > 演练管理 页面，单击新建演练。 2. 在基本信息 页面，按提示填写演练名称和描述，然后单击下一步。 3. 在演练对象配置页面： 配置动作组 ：为动作组 命名，资源类型选择分布式消息服务Kafka。 添加实例 ：单击添加实例 ，勾选上一步中添加的分布式消息服务Kafka实例。 添加故障动作 ：单击立即添加 ，在列表中选择Broker宕机动作。 4. 在弹出的参数配置框中，配置所需参数，然后单击确定。 持续时间：故障动作持续时间。 故障节点：可选择随机一个节点或者特定的节点。

本文介绍如何在事件总线EventBridge管理控制台添加分布式消息服务Kafka作为事件流中的事件提供方。 前提条件 事件总线EventBridge 开通事件总线EventBridge并委托授权。 分布式消息服务Kafka 开通分布式消息服务Kafka。 操作步骤 1. 登录事件总线管理控制台。 2. 在左侧导航栏，单击事件流。 3. 在事件流页面，单击创建事件流。 4. 在事件源（Source） 配置面板，事件提供方 选择分布式消息服务Kafka，选择或填入主题等配置然后单击确认，如图1所示。 图1 创建事件流时选择事件提供方为分布式消息服务Kafka 参数说明 参数 说明 示例 实例名称 选择前提条件中已创建的分布式消息服务Kafka版实例。 KAFKAxxx Topic 选择当前实例中的Topic。 topic1 Group 快速创建：自动创建以GIDEVENTBRIDGExxx命名的Group ID。 使用已有：选择当前实例中已创建的Group，请不要与已有业务的Group混用，以免影响已有的消息收发。 group1 消费位点 开始消费的位置。 最新位点：从最新位点开始消费。 最早位点：从最初位点开始消费。 最新位点 事件示例 plaintext { "id": "c40c8fe289994177b5aed37466bae2e5", "source": "ctyun:kafka", "specversion": "1.0", "subject": "ctyun:kafka:b342b77ef26b11ecbxxxxxxxc110002:0b3a633836ef4xxxxxx33ab4da:topic:streamsource", "time": "20250206T03:44:55.026774053Z", "datacontenttype": "application/json", "type": "kafka:Topic:SendMessage", "ctyunregion": "b342b77ef26b11ecbxxxxxxxc110002", "ctyunaccountid": "0b3a633836ef4xxxxxx33ab4da", "data": { "offset": 1, "topic": "streamsource", "value": "Hello,Kafka!", "key": "test" } } data字段包含的参数解释如下表所示： 参数 类型 示例值 描述 topic String source1 Topic名称。 offset Integer 1 消费位点。 key String test 消息Key值。 value Object Hello,Kafka! 消息体，默认以JSON格式编码。

本章节介绍Kafka Broker节点磁盘IO高负载故障演练。 背景介绍 分布式消息服务 Kafka 集群的性能与稳定性高度依赖底层磁盘 IO 能力，高并发写入、集群数据复制、海量消息存储检索及磁盘故障恢复等场景易导致 Broker 节点磁盘 IO 触达瓶颈，引发消息持久化延迟等问题，本演练可测试业务系统的响应与恢复能力。 基本原理 指定或随机一个Broker节点先通过dd命令将数据写入文件中，然后再通过循环读写文件占用磁盘带宽。 故障注入 1、纳管实例资源 1. 导航至 故障演练 > 目标应用 > 应用资源页面。 2. 在资源类型页签中选择分布式消息服务Kafka，然后单击添加资源。 3. 在弹出的对话框中，勾选目标分布式消息服务Kafka实例，单击确定。 2、编排演练任务 1. 导航至 故障演练 > 目标应用 > 演练管理 页面，单击新建演练。 2. 在基本信息 页面，按提示填写演练名称和描述，然后单击下一步。 3. 在演练对象配置页面： 配置动作组 ：为动作组 命名，资源类型选择分布式消息服务Kafka。 添加实例 ：单击添加实例 ，勾选上一步中添加的分布式消息服务Kafka实例。 添加故障动作 ：单击立即添加 ，在列表中选择Broker 磁盘IO高负载动作。 4. 在弹出的参数配置框中，配置所需参数，然后单击确定。 持续时间：故障动作持续时间。 读负载：开启读压力模式，创建一个临时文件并对其进行持续的读取操作。 写负载：开启写压力模式，持续向一个临时文件写入数据。 块大小：控制单次读写操作的数据块大小，单位为MB。增大此值可以提升单次操作的 IO 压力。通常保持默认值即可。

介绍分布式消息服务Kafka续订操作内容。 场景描述 分布式消息服务Kafka为用户提供全面周到的服务，支持用户续订的需求。针对包周期消息实例服务，用户可在到期前进行服务周期延长操作，即续订操作。 操作步骤 （1）登录管理控制台。 （2）进入Kafka管理控制台。 （3）在实例列表页在操作列，目标实例行点击“更多”，点击“续订”。 （4）点击“续订”后，会打开新页面，选择续订时长，会计算出对应价格，点击“确定”，后可到我的订单中支付。

请求示例 请求url 请求头header { "regionId":"bb9fdb42056f11eda1610242ac110002" } 请求体body { "prodInstId": "445d30ce3d5d40a2a79dc444a8c83859", "data": [ { "topics": "t/a,t/b,t/c", "clientid": "mqtta", "qos": 0, "payload": "test messsage" } ] } 响应示例 { "statusCode":800 } 状态码 请参考 状态码 错误码 请参考 错误码

本文主要介绍分布式消息服务RabbitMQ的准备环境。 虚拟私有云 虚拟私有云（Virtual Private Cloud，以下简称VPC）为RabbitMQ专享版实例提供一个隔离的、用户自主配置和管理的虚拟网络环境。 1. 在创建RabbitMQ专享版实例前，确保已存在可用的虚拟私有云和子网。 创建方法，请参考创建虚拟私有云和子网。如果您已有虚拟私有云和子网，可重复使用，不需要多次创建。 在创建VPC和子网时应注意如下要求： 创建的VPC与使用的RabbitMQ服务应在相同的区域。 创建VPC和子网时，如无特殊需求，配置参数使用默认配置即可。 2. 在创建RabbitMQ专享版实例前，确保已存在可用的安全组。 创建方法，请参考创建安全组。如果您已有安全组，可重复使用，不需要多次创建。 使用RabbitMQ实例前，添加表1所示安全组规则，其他规则请根据实际需要添加。 表1 安全组规则 方向 协议 端口 源地址 说明 入方向 TCP 5672 0.0.0.0/0 访问RabbitMQ实例（关闭SSL加密） 入方向 TCP 5671 0.0.0.0/0 访问RabbitMQ实例（开启SSL加密） 入方向 TCP 15672 0.0.0.0/0 访问Web界面UI地址（关闭SSL加密） 入方向 TCP 15671 0.0.0.0/0 访问Web界面UI地址（开启SSL加密） 说明 创建安全组后，系统默认添加入方向“允许安全组内的弹性云主机彼此通信”规则和出方向“放通全部流量”规则，此时使用内网通过同一个VPC访问RabbitMQ实例，无需添加上表中的规则。

场景描述 消息轨迹记录了一条消息从生产端到消息队列RocketMQ服务端，再到消费端的整个过程，包括各阶段的时间、执行状态等。 使用说明 客户端SDK默认不开启消息轨迹功能，需要在消息收发时主动开启消息轨迹，以Java为例，如下所示： 生产者启用消息轨迹： DefaultMQProducer producer new DefaultMQProducer("ProducerGroupName",new AclClientRPCHook(newSessionCredentials("控制台角色AK","控制台角色CK")),true,null); 消费者启用消息轨迹： DefaultMQPushConsumer consumer new DefaultMQPushConsumer("ConsumerGroupName",new AclClientRPCHook( new SessionCredentials( "控制台角色AK", "控制台角色CK") ),newAllocateMessageQueueAveragely(),true,null); 消息轨迹的查询方式 1. 天翼云官网点击控制中心，选择产品分布式消息服务RocketMQ。 2. 登录分布式消息服务RocketMQ控制台，点击右上角地域选择对应资源池。 3. 进入实例列表，点击【管理】按钮进入管理菜单。 4. 根据查询结果，查看消息轨迹。

介绍分布式消息服务Kafka退订操作内容。 场景描述 分布式消息服务kafka为用户提供全面周到的服务，支持用户退订的需求。用户如不需要继续使用该分布式消息服务Kafka实例，可进行删除实例操作，即退订操作。 操作步骤 （1）登录管理控制台。 （2）进入Kafka管理控制台。 （3）在实例列表页在操作列，目标实例行点击“更多”，点击“退订”。 （4）点击“退订”后，会打开新页面，点击“确定” 注意 退订的实例处于冻结状态，请务必在实例退订前停止全部的应用。 在申请退订前，请做好数据备份工作，退订后数据将保留15个自然日，15天后相关数据将不予保留，且不会进行备份，务必谨慎操作。

前提条件 RabbitMQ 版本 ≥ 3.8 启用官方插件： rabbitmqplugins enable rabbitmqdelayedmessageexchange 天翼云RabbitMQ默认启用延迟交换器交换器插件，无需操作 使用步骤 1. 声明 xdelayedmessage 类型交换机 2. 指定底层路由类型（xdelayedtype） 3. 发送消息时在 Header 中设置 xdelay（毫秒） 代码示例 java //声明延时交换机 Map args new HashMap<>(); args.put("xdelayedtype","direct"); channel.exchangeDeclare("delayedexchange","xdelayedmessage",true,false,args); // 声明并绑定队列 channel.queueDeclare("targetqueue",true,false,false,null); channel.queueBind("targetqueue","delayedexchange","routingkey"); //发送延时消息（延迟 5 秒） Map headers new HashMap<>(); headers.put("xdelay",5000); AMQP.BasicProperties props new AMQP.BasicProperties.Builder() .headers(headers) .build(); channel.basicPublish("delayedexchange","routingkey",props,"延时消息".getBytes(StandardCharsets.UTF8)); 特点说明 延迟精度高，使用方式直观 延迟逻辑在 Exchange 层完成 依赖插件及 RabbitMQ 版本兼容性 注意事项 精度与可靠性 DLX + TTL 可能存在毫秒级到秒级误差 插件方案 延迟精度更高，更适合严格定时场景 资源消耗 延时消息在到期前会占用 Broker 内存 大量长延时消息需重点监控： 队列堆积 内存水位 磁盘使用率 消息顺序 延时消息可能打乱原始发送顺序 对顺序敏感的业务需在消费端自行处理

本章节主要介绍分布式消息服务RabbitMQ的实例类问题。 RabbitMQ使用的版本是多少？ 服务端RabbitMQ的版本是3.8.35、3.7.17。 RabbitMQ实例SSL连接的协议版本号是多少？ TLS v1.2版本。 创建实例时为什么无法查看子网和安全组等信息？ 创建实例时，如果无法查看虚拟私有云、子网、安全组、弹性IP，可能原因是该用户无Server Administrator和VPC Administrator权限，增加权限的详细步骤请参考《统一身份认证服务 用户指南》的“用户指南 > 用户组及授权 > 查看或修改或删除用户组”章节。 若其中一台RabbitMQ重启失败，需要会如何处理？ 重启RabbitMQ实例时，不会重启实例所在虚拟机，仅重启RabbitMQ进程。 重启集群实例时，若其中一台RabbitMQ进程重启失败，则重启后实例状态依然为“运行中”，并提示“部分节点故障”。在每台虚拟机上都有RabbitMQ的守护进程，定时检查RabbitMQ进程是否存在，当进程不存在时会自动拉起RabbitMQ进程。 如果RabbitMQ实例异常持续超过1分钟，会上报告警。 RabbitMQ集群实例如何均衡分发请求到每个虚拟机？ 集群内部使用LVS做负载均衡，由LVS将请求均衡分发到每个虚拟机节点。 RabbitMQ实例集群内部的队列是否有冗余备份？ 队列是否做镜像（即冗余备份）取决于用户的需要，如果用户设置了镜像，会在集群中多个代理上存储队列的副本，当某个代理故障，集群会从其他正常的代理中选择一个代理，用来同步队列数据。

本章节介绍Kafka Broker分区Leader不可用故障演练。 背景介绍 分布式消息服务 Kafka 集群中，分区 Leader 节点宕机、网络分区或资源过载等因素均可能造成分区 Leader 不可用，进而引发对应分区服务中断、客户端请求失败、Leader 重选举导致延迟升高及副本同步紊乱等问题，本演练可帮助验证集群高可用选举机制的有效性、监控告警的及时性，以及业务系统应对此类故障的容错能力与恢复效率。 基本原理 指定一个或多个分区Leader，通过调用Kafka模拟Leader故障OpenAPI，触发Leader重新选举。 故障注入 1、纳管实例资源 1. 导航至 故障演练 > 目标应用 > 应用资源页面。 2. 在资源类型页签中选择分布式消息服务Kafka，然后单击添加资源。 3. 在弹出的对话框中，勾选目标分布式消息服务Kafka实例，单击确定。 2、编排演练任务 1. 导航至 故障演练 > 目标应用 > 演练管理 页面，单击新建演练。 2. 在基本信息 页面，按提示填写演练名称和描述，然后单击下一步。 3. 在演练对象配置页面： 配置动作组 ：为动作组 命名，资源类型选择分布式消息服务Kafka。 添加实例 ：单击添加实例 ，勾选上一步中添加的分布式消息服务Kafka实例。 添加故障动作 ：单击立即添加 ，在列表中选择分区Leader不可用动作。 4. 在弹出的参数配置框中，配置所需参数，然后单击确定。 持续时间：故障动作持续时间。 主题分区：指定需要模拟分区Leader不可用的目标对象。

本文介绍如何配置MQTT开源Java SDK的客户端自动连接 要配置MQTT开源Java SDK的客户端自动连接，你需要设置以下参数和逻辑： 1、 设置自动重连：首先，你需要设置客户端的自动重连功能。在MQTT开源Java SDK中，你可以使用setAutomaticReconnect(true)方法来启用自动重连功能。这将使客户端在连接丢失后自动尝试重新连接。 MqttClient client new MqttClient(brokerUrl, clientId); client.setAutomaticReconnect(true); 2、 设置重连间隔：接下来，你可以设置自动重连的时间间隔。这将确定客户端在连接丢失后尝试重新连接的时间间隔。在MQTT开源Java SDK中，你可以使用setReconnectDelay方法来设置重连间隔，单位为毫秒。 client.setReconnectDelay(5000); // 设置重连间隔为5秒 3、 处理连接丢失事件：当连接丢失时，你可以通过实现MqttCallback接口来处理连接丢失事件。在connectionLost方法中，你可以添加自定义的处理逻辑，例如记录日志或触发重新连接。 client.setCallback(new MqttCallback() { @Override public void connectionLost(Throwable cause) { // 处理连接丢失事件 // 添加自定义的处理逻辑，例如记录日志或触发重新连接 } // 其他回调方法... }); 4、 处理重新连接事件：当客户端尝试重新连接时，你可以在reconnectComplete方法中添加自定义的处理逻辑。这个方法会在客户端成功重新连接到MQTT Broker后被调用。 client.setCallback(new MqttCallback() { // 其他回调方法... @Override public void reconnectComplete(boolean reconnect) { // 处理重新连接完成事件 // 添加自定义的处理逻辑 } }); 通过以上配置和逻辑，你可以实现MQTT开源Java SDK的客户端自动连接功能。当连接丢失时，客户端将自动尝试重新连接，并在重新连接完成后触发相应的事件回调方法。你可以根据需要添加自定义的处理逻辑来处理连接丢失和重新连接事件。

如果消息重复消费会影响您的业务处理，要对消息做幂等处理。本文介绍消息幂等的概念、适用场景以及处理方法。 概念 在消息领域，幂等是指Consumer重复消费某条消息时，重复消费的结果与消费一次的结果是相同的，并且多次消费并未对业务系统产生任何负面影响。 例如，在支付场景下，Consumer消费扣款消息，对一笔订单执行扣款操作，扣款金额为500元。如果因网络不稳定等原因导致扣款消息重复投递，Consumer重复消费了该扣款消息，但最终的业务结果是只扣款一次，扣费500元，且用户的扣款记录中对应的订单只有一条扣款流水，不会多次扣除费用。那么这次扣款操作是符合要求的，整个消费过程实现了消息幂等。 适用场景 在互联网应用中，尤其在网络不稳定的情况下，分布式消息服务RabbitMQ的消息有可能会出现重复。如果消息重复消费会影响您的业务处理，请对消息做幂等处理。消息重复的可能原因如下： 发送时消息重复 当一条消息已被成功发送到服务端并完成持久化，此时出现了网络闪断或者客户端宕机，导致服务端对客户端应答失败。 如果此时Producer意识到消息发送失败并尝试再次发送消息，Consumer后续会收到两条内容相同并且Message ID也相同的消息。 投递时消息重复 消息消费的场景下，消息已投递到Consumer并完成业务处理，当客户端给服务端反馈应答的时候网络闪断。为了保证消息至少被消费一次，分布式消息服务RabbitMQ的服务端将在网络恢复后再次尝试投递之前已被处理过的消息，Consumer后续会收到两条内容相同并且Message ID也相同的消息。 负载均衡时消息重复（包括但不限于网络抖动、服务端重启以及Consumer应用重启） 当分布式消息服务RabbitMQ的服务端或客户端重启、扩容或缩容时，会触发Rebalance，此时Consumer可能会收到重复消息。

本节介绍了分布式消息服务RabbitMQ续费、到期与欠费内容。 到期前续费 手动续订：对于包年/包月订购的分布式缓存服务，用户在资源到期前进行续费操作，可以延长原有资源到期时间，避免资源到期后冻结或超过保留期后被系统回收。详细操作请参考费用中心续订管理手动续订。 自动续订：自动续订仅针对采用包月、包年计费模式的资源，详细操作请参考费用中心续订管理自动续订。 到期处理 到期后，分布式消息服务RabbitMQ进入保留期，您将不能正常访问及使用天翼云分布式消息服务RabbitMQ服务，已开通的实例资源将予以保留。 若您在到期后15天内续费，自资源续订解冻开始，计算新的服务有效期，按照新的服务有效期计算费用； 若到期15天后您仍未续费，RabbitMQ实例资源将被释放 欠费原因 在按需计费的模式下帐号的余额不足。 按需欠费资源冻结规则 欠费后，资源进入保留期，您将不能正常访问及使用分布式消息服务RabbitMQ，已开通的实例资源将予以保留。 若您在保留期内充值，充值后系统会自动扣减欠费金额。 若保留期到期您仍未充值，RabbitMQ实例资源将被释放。

介绍分布式消息服务Kafka业务迁移内容。 应用场景 Kafka的业务迁移可以应用在多个领域和场景中，包括但不限于以下几个方面： 数据集成和数据仓库：Kafka可以用作数据集成的中间件，将分散的数据源集中到一个统一的平台上。通过使用Kafka的生产者和消费者，可以实现数据的可靠传输和消费，支持实时的数据集成和数据仓库的建设。 实时数据处理和流处理：Kafka提供了流处理功能，可以对实时数据流进行处理和分析。通过使用Kafka Streams、Apache Flink等流处理框架，可以对数据流进行实时的计算、转换、聚合等操作，实现实时数据处理和实时决策。 异步通信和消息队列：Kafka作为消息队列，可以用于异步通信和解耦系统之间的依赖关系。通过将同步的请求和响应转换为异步的消息，可以提高系统的可伸缩性和响应性能，实现松耦合的系统架构。 日志收集和分析：Kafka可以用作日志收集和分析的中间件，用于接收和传递系统和应用程序的日志数据。通过将日志数据发送到Kafka中，可以实现日志的集中存储和分发。同时，可以使用Kafka的消费者来实时消费和分析日志数据，帮助进行故障排查、性能监控和安全审计等工作。 数据同步和复制：在多个数据中心或分布式系统之间进行数据同步和复制时，可以使用Kafka作为数据的中间传输通道。通过使用Kafka的生产者和消费者，可以实现数据的可靠传输和复制，保证数据的一致性和可用性。 这些只是Kafka业务迁移的一部分应用场景，实际应用中还有很多其他的需求和场景。Kafka的高性能、可靠性和可扩展性使其成为业务迁移的理想选择。

操作步骤 1、 天翼云官网点击控制中心，选择产品分布式消息服务RocketMQ。 2、 登录分布式消息服务RocketMQ控制台，点击右上角地域选择对应资源池。 进入实例列表，点击【管理】按钮进入管理菜单。 3、 进入实例列表，点击【管理】按钮进入管理菜单。 4、 进入Topic管理菜单，点击【重置消费】按钮，弹出窗口选择消费组名称。 5、 选择重置时间点，点击确定即可重置消费位置。 注意：当将消费时间重置到历史时间T1,则T1之前生产的消息，都变成已消费，T1之后生产的消息，都变成未消费；将消费时间重置到未来时间T2，则存量的所有消息都变成已消费。 备注： 重置时间：是指消息生产的时间 不选时间：所有消息都变成未消费 未来时间：所有消息都变成已消费 某时间点：该时间之前变成已消费，改时间之后未消费； v1消费模式必须关闭消费客户端