本文主要介绍 查看流控监控。 本章节指导您在控制台查看每个代理上用户/客户端/Topic的流控使用量。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例的名称，进入实例详情页面。 步骤 5 在左侧导航栏单击“流控管理 > 流控监控”，进入流控监控页面。 步骤 6 设置查询流控的参数。 表查询流控的参数说明 参数 说明 :: 统计方式 设置流控的统计方式。 前n个：统计带宽使用量排名前x 个的用户/客户端/Topic，x由您自行输入 带宽速率：统计带宽速率大于x MB/s的用户/客户端/Topic，x由您自行输入 带宽使用量：统计带宽使用量超过x %的用户/客户端/Topic，x由您自行输入 统计类型 设置流控的统计类型。 生产：统计生产消息的流控 消费：统计消费消息的流控 统计维度 设置流控的统计维度。 User/Client：统计用户/客户端的流控 Topic：统计Topic的流控 图 查询流控的参数 步骤 7 单击“查询”，查看每个代理上用户/客户端/Topic的流控使用量。

./kafkaconsoleproducer.sh brokerlist 10.3.196.45:9095,10.78.42.127:9095,10.4.49.103:9095 topic topicdemo producer.config ../config/producer.properties >Hello >DMS >Kafka! >^C[root@ecskafka bin] 如需停止生产使用Ctrl+C命令退出。 步骤 6 执行如下命令消费消息。 ./kafkaconsoleconsumer.sh bootstrapserver ${连接地址} topic ${Topic名称} group ${消费组名称} frombeginning consumer.config ../config/consumer.properties 参数说明如下： 连接地址：从前提条件获取的连接地址，如果是公网访问，请使用“公网连接地址”，如果是VPC内访问，请使用“内网连接地址”，请根据实际情况选择。 Topic名称：Kafka实例下创建的Topic名称。 消费组名称：根据您的业务需求，设定消费组名称。 如果已经在配置文件中指定了消费组名称，请确保命令行中的消费组名称与配置文件中的相同，否则可能消费失败。 消费组名称开头包含特殊字符，例如下划线“”、 号“ ”时，监控数据无法展示。 示例如下： [root@ecskafka bin] ./kafkaconsoleconsumer.sh bootstrapserver 10.3.196.45:9095,10.78.42.127:9095,10.4.49.103:9095 topic topicdemo group ordertest frombeginning consumer.config ../config/consumer.properties Hello DMS Kafka! ^CProcessed a total of 3 messages [root@ecskafka bin] 如需停止消费使用Ctrl+C命令退出。

本文主要介绍查看消费者连接地址。 分布式消息服务Kafka支持通过以下两种方式查看消费者连接地址，您可以根据实际情况选择任意一种方式。 方法一：在管理控制台查看消费者连接地址 方法二：在Kafka Manager中查看消费者连接地址 说明 消费者处于连接Kafka实例时，才可以查看消费者连接地址。 Kafka Manager由于缓存原因，显示的消费者连接地址可能不是当前消费者连接地址，建议重启Kafka Manager解决。 方法一：在管理控制台查看消费者连接地址 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例的名称，进入实例详情页面。 步骤 5 在左侧导航栏选择“消费组管理”，进入消费组列表页面。 步骤 6 单击待查看消费者连接地址的消费组名称，进入消费组详情页。 步骤 7 在“消费者列表”页签，查看消费者连接地址。 图 消费者列表 方法二：在Kafka Manager中查看消费者连接地址 步骤 1 登录Kafka Manager。 步骤 2 单击“kafkacluster”，进入集群详情页。 步骤 3 在顶部导航栏单击“Consumers”，进入消费组列表页面。 图 导航栏 步骤 4 单击待查看消费者连接地址的消费组名称，进入消费组订阅的Topic列表页面。 图 消费组列表页面 步骤 5 单击待查看消费者连接地址的Topic名称，进入Topic详情页。 图 消费组订阅的Topic列表页面 步骤 6 在“Consumer Instance Owner”中，查看消费者连接地址。 图Topic详情页

本文主要介绍编辑流控。 操作场景 本章节指导您在流控创建成功后，修改生产上限速率或者消费上限速率。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例的名称，进入实例详情页面。 步骤 5 在左侧导航栏单击“流控管理 > 流控列表”，进入流控列表页面。 步骤 6 在待修改的流控所在行，单击“编辑”，弹出“编辑流控”对话框。 步骤 7 修改生产上限速率或者消费上限速率，单击“确定”，跳转到“后台任务管理”页面，当流控任务的“状态”为“成功”时，表示流控修改成功。 进入“流控管理 > 流控列表”页面，查看修改后流控的生产上限速率或者消费上限速率。 说明 “生产上限速率”和“消费上限速率”不可同时为空。

号“ ”，会导致监控数据无法展示。 是否支持延迟队列 不支持 消费组 表消费组约束与限制 限制项 约束与限制 是否需要创建消费组、消费者、生产者 不需要单独创建消费组、生产者和消费者，在使用时自动生成，实例创建后，直接使用即可。 重置消费进度 重置消费进度可能会导致重复消费。 消费组名称 消费组名称开头包含特殊字符，例如下划线“”、 号“ ”，会导致监控数据无法展示。 消息 表 消息约束与限制 限制项 约束与限制 消息大小 生产消息的最大长度为10M，超过10M会导致生产失败。 用户 表用户约束与限制 限制项 约束与限制 创建用户的数量 一个Kafka实例最多创建20个SASLSSL用户。

天翼云分布式消息服务Kafka 用户使用指南.pdf

本文主要介绍查看实例后台任务。 . 对实例的一些操作，如修改公网访问开关、修改容量阈值策略等，会启动一个后台任务，您可以在后台任务页，查看该操作的状态等信息，同时可通过删除操作，清理任务信息。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例的名称，进入实例的“基本信息”页签。 步骤 5 单击“后台任务管理”页签，进入后台任务管理页面。 界面显示任务列表。 步骤 6 在右上角单击系统默认时间，选择开始时间和结束时间，单击“确定”，界面显示相应时间段内启动的任务。 在后台任务管理页面，您还可以执行以下操作： 单击，刷新任务状态。 单击“删除”，弹出“删除任务记录”对话框，单击“是”，清理任务信息。 说明 您只能在任务已经执行完成，即任务状态为成功或者失败时，才能执行删除操作。

修改Kafka连接地址 1.在管理控制台左上角单击，选择“企业间件 > 分布式消息服务Kafka版”，进入分布式消息服务Kafka控制台页面。 2.单击Kafka集群名称，进入实例详情页面。 3.在“实例详情”页面的“高级配置”区域，单击“添加路由策略”，选择接入类型为跨VPC访问，填入advertised.listeners IP和Port端口，advertised.listeners IP为7和8中记录的节点IP，8为创建终端节点服务时填入的端口。修改完后，单击“保存”。其中dvertised.listeners IP中可填入域名，需客户端支持域名解析。 4.20240810前开通的实例，需要重启实例才能生效 图9 修改advertised.listeners IP 验证接口连通性 参考安全接入点接入，测试是否可以生产和消费消息。 测试接口连通性时，注意以下几点： 连接Kafka实例的地址为“advertiesd.listeners IP:port”，以图9为例，连接Kafka实例的地址为“10.254.0.74:9011,10.254.0.75:9011,10.254.0.88:9011”。 在Kafka实例安全组的入方向规则中放通对应端口，以及 XX.XX.XX.0/16网段的地址。 说明 XX.XX.XX.0/16是为VPC终端节点分配的网段，使用VPC终端节点需要放通此网段。

本文主要介绍设置实例公网访问。 当业务不再使用公网访问功能时，也可以关闭实例的公网访问功能。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例的名称，进入实例的“基本信息”页签。 步骤 5 设置公网访问。 说明 只有处于“运行中”状态的Kafka实例支持修改公网访问配置。 Kafka实例只支持绑定IPv4弹性IP地址。 开启公网访问。 在“公网访问”后，单击，开启公网访问。在“弹性IP地址”中，为每个代理设置对应的弹性IP地址，单击。 您可以在实例的“后台任务管理”页面，查看当前任务的操作进度。任务状态为“成功”，表示操作成功。 图 公网访问 开启公网访问后，需要设置对应的安全组规则，才能成功连接Kafka。连接Kafka的具体操作请参考连接Kafka。 表安全组规则 方向 协议 端口 源地址 说明 入方向 TCP 9094 0.0.0.0/0 通过公网访问Kafka（关闭SSL加密）。 入方向 TCP 9095 0.0.0.0/0 通过公网访问Kafka（开启SSL加密）。 关闭公网访问。 在“公网访问”后，单击，完成公网访问的关闭。 您可以在实例的“后台任务管理”页面，查看当前任务的操作进度。任务状态为“成功”，表示操作成功。

关闭IPv4公网访问（不支持修改SASL开关） 1. 登录管理控制台。 2. 在管理控制台左上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 3. 在管理控制台左上角单击，选择“应用中间件 > 分布式消息服务Kafka”，进入分布式消息服务Kafka专享版页面。 4. 单击Kafka实例的名称，进入实例的“基本信息”页面。 5. 在“公网访问”后，单击，完成公网访问的关闭。 您可以在实例的“后台任务管理”页面，查看当前任务的操作进度。任务状态为“成功”，表示操作成功。 关闭公网访问后，需要设置对应的安全组规则（请见表2），才能通过内网成功连接Kafka。 表2 Kafka实例安全组规则（内网访问） 方向 协议 类型 端口 源地址 说明 入方向 TCP IPv4 9092 0.0.0.0/0 使用内网通过同一个VPC访问Kafka实例（关闭SSL加密）。 入方向 TCP IPv4 9093 0.0.0.0/0 使用内网通过同一个VPC访问Kafka实例（开启SSL加密）。 说明 创建安全组后，系统默认添加入方向“允许安全组内的弹性云主机彼此通信”规则和出方向“放通全部流量”规则，此时使用内网通过同一个VPC访问Kafka实例，无需添加表2的规则。

分布式消息服务Kafka提供全托管、免运维的迁移上云服务，用于自建实例或跨云云实例与分布式消息服务Kafka实例之间的数据同步。本文介绍迁移上云的源实例类型、使用限制、跨VPC数据同步以及使用流程。 背景信息 迁移上云可以把源云分布式消息服务Kafka集群的元数据（Topic和Group配置信息）、消息数据和点位信息同步到目标集群，迁移完成后目标集群的元数据与原集群的元数据保持一致，并且支持持续更新和完成后自动停止任务。 使用限制 1. 此功能不收费，会占用当前实例的服务器资源，请结合业务流量和服务器资源占用情况，在合理的情况下进行迁移。 2. 单机版不支持迁移。 计费说明 分布式消息服务Kafka的迁移上云组件处于公测阶段，不会在分布式消息服务Kafka侧产生费用。同时，天翼云不承诺迁移的SLA，使用迁移上云所依赖的其他产品的SLA和费用说明请以对应产品为准。 环境要求 如果您需要通过迁移上云功能将公网的自建实例的数据同步到分布式消息服务Kafka或者跨地域将某个VPC内的分布式消息服务Kafka的数据同步到另一个VPC的分布式消息服务Kafka，您需要为该实例绑定弹性IP开启公网访问，然后在公网上进行数据同步。 如果您需要将某个VPC内的数据通过迁移上云功能同步到另一个VPC的分布式消息服务Kafka，需要手动打通VPC。具体步骤，请参见：使用VPCEP实现跨VPC访问Kafka

步骤二：替换证书 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 此处请选择与您的应用服务相同的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击待替换证书的实例名称，进入实例详情页面。 步骤 5 在“连接信息 > SSL证书”后，单击“替换”，弹出“替换SSL证书”对话框。 图 连接信息 步骤 6 参考下表，设置替换SSL证书的参数。 图 替换SSL证书 表 替换SSL证书参数说明 参数名称 说明 Key密码 输入制作证书中设置的keystore密码 Keystore密码 输入制作证书中设置的keystore密码 Keystore文件 导入“server.keystore.jks”证书 Truststore密码 输入服务端证书的Truststore密码 Truststore文件 导入“server.truststore.jks”证书 步骤 7 单击“确定”，弹出“替换SSL证书”对话框。 步骤 8 单击“确认”，完成证书的替换。 在“后台任务管理”页签，替换SSL证书任务的“状态”为“成功”时，表示替换证书成功。 说明 证书替换成功后，在实例详情页单击“下载”，下载的证书为分布式消息服务Kafka提供的证书，并非您自己制作的证书。

本文主要介绍修改Topic分区数。 说明 修改分区数，不会影响业务。 修改Topic分区数的方法如下： 方法1：在控制台修改 方法2：在Kafka Manager上修改 方式3：在Kafka客户端上修改 方法1：在控制台修改 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例的名称，进入实例详情页面。 步骤 5 选择“Topic管理”页签，显示已创建的Topic详情。 步骤 6 通过以下任意一种方法，修改Topic分区数。 勾选Topic名称左侧的方框，可选一个或多个，单击信息栏左上侧的“编辑Topic”。 在待修改分区数的Topic所在行，单击“编辑”。 步骤 7 在“编辑Topic”对话框中，输入分区数，单击“确定”。 说明 分区数只支持增加，不支持减少。 出于性能考虑，Kafka控制台限定单个Topic的分区数上限为100。 所有Topic分区数总和不能超过实例允许的分区上限。 方法2：在Kafka Manager上修改 步骤 1 登录Kafka Manager。 步骤 2 在Kafka Manager中，单击“Topic > List”，进入Topic列表界面。 步骤 3 单击Topic名称，进入Topic详情界面。 步骤 4 单击“Add Partitions”，进入增加分区界面。 图 Topic详情界面 步骤 5 输入分区数，单击“Add Partitions”。 图 增加分区数量 显示“Done”，表示分区增加成功。 图分区增加成功 说明 分区数只支持增加，不支持减少。 所有Topic分区数总和不能超过实例允许的分区上限。

云服务名称 是否支持条件键 云通信短信 否 弹性文件服务 否 分布式消息服务Kafka 否 账务 是 分布式消息服务RabbitMQ 否 分布式消息服务RocketMQ 否 云硬盘 是 天翼云电脑（政企版） 是 内网DNS 否 CRM业务中台 是 分布式缓存服务Redis版 否 弹性云主机 是 弹性负载均衡 否 镜像服务 是 云监控 否 弹性伸缩服务 否 虚拟私有云 是 物理机 否 云硬盘备份 否 云主机备份 否 服务器安全卫士（原生版） 否 密钥管理 否 云间高速 否 统一身份认证 是 客服工单 是 容器云服务引擎CCSE 是 微服务应用平台MSAP 否 活动与券 是 消息管理 是 Web应用防火墙（原生版） 否 云审计 否 企业组织 是 VPC终端节点 否 NAT网关 是

顺序消息是分布式消息服务RocketMQ版提供的一种严格按照顺序来发布和消费的消息类型。 顺序消息分为全局顺序消息和分区顺序消息： 全局顺序消息：对于指定的一个Topic，将队列数量设置为1，这个队列内所有消息按照严格的先入先出FIFO（First In First Out）的顺序进行发布和订阅。 分区顺序消息：对于指定的一个Topic，同一个队列内的消息按照严格的FIFO顺序进行发布和订阅。生产者指定分区选择算法，保证需要按顺序消费的消息被分配到同一个队列。 全局顺序消息和分区顺序消息的区别仅为队列数量不同，代码没有区别。 收发消息前，请参考收集连接信息收集RocketMQ所需的连接信息。 准备环境 开源的Java客户端支持连接分布式消息服务RocketMQ版，推荐使用的客户端版本为4.9.7。 通过以下任意一种方式引入依赖： 1. 使用Maven方式引入依赖。 org.apache.rocketmq rocketmqclient 4.9.7 org.apache.rocketmq rocketmqacl 4.9.7 2. 点击下载依赖JAR包：rocketmqall4.9.7binrelease.zip 发送顺序消息 参考如下示例代码 import org.apache.rocketmq.acl.common.AclClientRPCHook; import org.apache.rocketmq.acl.common.SessionCredentials; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.RPCHook; import org.apache.rocketmq.remoting.common.RemotingHelper; public class ProducerFifoExample { private static RPCHook getAclRPCHook() { return new AclClientRPCHook(new SessionCredentials( "accessKey", // 分布式消息服务RocketMQ控制台用户管理菜单中创建的用户ID "accessSecret" // 分布式消息服务RocketMQ控制台用户管理菜单中创建的密钥 )); } public static void main(String[] args) throws Exception { / 创建Producer，如果想开启消息轨迹，可以按照如下方式创建： DefaultMQProducer producer new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook(), true, null); / DefaultMQProducer producer new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook()); // 填入控制台NAMESRV接入点地址 producer.setNamesrvAddr("XXX:xxx"); ; // 如果需要开启SSL，请增加此行代码 producer.start(); for (int i 0; i { // 选择适合自己的分区选择算法，保证同一个参数得到的结果相同。 Integer id (Integer) arg; int index id % mqs.size(); return mqs.get(index); }, orderId); System.out.printf("%s%n", sendResult); } catch (Exception e) { e.printStackTrace(); } } producer.shutdown(); } } 注意 上述代码中，相同id的消息需要保证顺序，不同id的消息不需要保证顺序，所以在分区选择算法中以“id/队列个数的余数”作为消息发送的队列。

本文对分布式消息服务RocketMQ的计费项进行说明。 4.0版本 以下适用于华东1、华北2、西南1、华南2、上海36、青岛20、长沙42、南昌5、武汉41、杭州7、西南2贵州、太原4、郑州5、西安7、呼和浩特3 节点。 分布式消息服务RocketMQ费用计算分为两部分，一部分为实例费用，一部分为存储空间费用。 计费项 含义 适用的计费模式 实例费用 用户选择的选择的实例规格节点数计费，实例规格单价请参考产品资费，节点数为Name Server集群固定3节点+Broker（代理） 2节点。 按需/包周期 存储空间费用 用户选择的实例存储空间计费（每个实例规格您都可以选择高IO、超高IO等多种不同的云硬盘类型以满足您的业务需求）。 按需/包周期 3.0版本 以下适用于南京3、上海7、重庆2、乌鲁木齐27、石家庄20、内蒙6、晋中、北京5 节点。 分布式消息服务RocketMQ按照消息队列的规格版本可分为基础版、中级版和高级版。 计费项：消息队列的规格版本。

规范 规范说明 级别 备注 业务数据分离 防止由于Redis故障或不可用影响所有相关业务收到干扰 建议 业务数据分离，避免多个应用使用一个Redis实例，不相干的业务拆分，公共数据做服务化。 完善redis的容错处理 缓存可能会遇到基础设施的故障或主备切换，应用需要对可能的故障和慢请求做容错处理。 建议 缓存访问失败时，具备降级措施，从DB获取数据；或者具备容灾措施，自动切换到另一个Redis使用。 避免redis内存使用量超过80% 虽然云主机会有一定的内存冗余，但是redis内存过高时，redis全量持久化时引入的COPYONWRITE机制带来额外的内存消耗，再加上大内存带来的内存碎片和云主机相关的管理进程消耗，容易触发操作系统OOM KILL。 强制 建议及时做扩容处理。 防止缓存击穿 防止缓存击穿，推荐搭配本地缓存使用Redis，对于热点数据建立本地缓存。本地缓存数据使用异步方式进行刷新。 建议 就近部署业务 避免网络延迟将极大影响读写性能，如果对于时延较为敏感，请避免使用跨AZ Redis实例。 建议 不当成消息队列 发布订阅场景下，不建议作为消息队列使用。redis的PUB/SUB支持的性能有限。 强制 如果需要消息队列，建议改用kafka或rabbitmq。

本文主要介绍管理实例标签。 标签是Kafka实例的标识，为Kafka实例添加标签，可以方便用户识别和管理拥有的Kafka实例资源。 您可以在创建Kafka实例时添加标签，也可以在Kafka实例创建完成后，在“标签”页面添加标签，您最多可以给实例添加20个标签。另外，您还可以进行修改和删除标签。 标签共由两部分组成：“标签键”和“标签值”，其中，“标签键”和“标签值”的命名规则如下表所示。 表 标签命名规则 参数名称 规则 标签键 不能为空。 对于同一个实例，Key值唯一。 长度不超过36个字符。 不能包含“”,“”,“ ”,“”,“,”,“ 标签值 不能为空。 长度不超过43个字符。 不能包含“”,“”,“ ”,“”,“,”,“ 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 在实例所在行，单击实例名称，进入实例详情页面。 步骤 5 单击“标签”页签，进入标签管理页面。 界面显示该实例的标签列表。 步骤 6 您可以根据实际需要，执行以下操作： 添加标签 单击“创建/删除标签”，弹出“创建/删除标签”对话框。 在“标签键”和“标签值”中，输入标签的键/值，单击“添加”。如果您已经预定义了标签，在“标签键”和“标签值”中选择已经定义的标签键/值，单击“添加”。 单击“确定”，成功为实例添加标签。 删除标签 通过以下任意一种方法，删除标签。 在待删除的标签所在行，单击“删除”，弹出“删除标签”对话框。单击“是”，完成标签的删除。 单击“创建/删除标签”，弹出“创建/删除标签”对话框。在待删除的标签后，单击，然后单击“确定”，完成标签的删除。

对比类 对比项 分布式消息服务Kafka 开源Kafka 简单易用 立等可用 即开即用，可视化操作，自助创建，自动化部署，分钟级创建实例，立即使用，实时查看和管理消息实例。 自行准备服务器资源，安装配置必要的软件并进行配置，等待时间长。 易出错。 简单API 提供简单的实例管理RESTFul API，使用门槛低。 无 成本低廉 按需使用 提供多种规格，按需使用，支持一键式在线进行实例代理个数、磁盘存储空间和代理规格扩容。 搭建消息服务本身需要费用，而且即使没有使用，所占用资源本身依旧要收费。 成本低廉 完全托管 租户不需要单独采购硬件资源，直接使用就绪的服务，无需额外成本。 需要购买硬件资源，自行搭建整个消息服务，使用和维护成本高。 实践验证 成熟度高 经受电商网站大规模访问考验，并且已经在云服务平台许多产品中使用，广泛部署运行在分布于世界各地的电信级客户云业务系统里。满足严苛的电信级故障模式库标准。紧随社区主流版本，修复开源bug，持续上线新功能，进行版本升级。 使用开源软件成熟度低，无法保证关键业务，商业案例少；自研周期长，并需要长时间进行验证。 实践验证 能力强大 100%兼容开源，支持一键扩容，深度优化开源代码提升性能和可靠性，支持消息查询等高级特性。 功能不完善，需额外投入进行开发。 稳定可靠 稳定高可用 支持跨AZ部署，提升可靠性。故障自动发现并上报告警，保证用户关键业务的可靠运行。 需要自己开发或基于开源实现，开发成本高昂，无法保证业务可靠运行。 稳定可靠 无忧运维 后台运维对租户完全透明，整个服务运行具有完备的监控和告警功能。有异常可以及时通知相关人员。避免724小时人工值守。 需要自行开发完善运维功能，尤其是告警及通知功能，否则只能人工值守。 稳定可靠 安全保证 VPC隔离，支持SSL通道加密。 需要自行进行安全加固。

操作场景 分布式消息服务 Kafka 版支持分区动态处理功能，当实例集群出现分区不平衡的情况时，可以手动进行分区均衡，重新分配分区在节点上的分布情况。也可以选择自动分区均衡 ，Kafka 会根据您设定的策略自动检查 Topic 的分区分布，自动分析选择业务低峰期发起分区均衡。 前提条件 自动分区均衡仅支持集群版实例。 自动分区均衡和Topic管理页面的分区平衡任务不能同时运行。 操作影响 分区均衡操作时可能引起流量抖动以及请求超时情况，建议在业务低峰期操作。 分区平衡后Topic的metadata会改变，如果生产者不支持重试机制，会有少量的请求失败，导致部分消息生产失败。 操作步骤 1. 进入Kafka实例管理控制台。 2. 在左侧菜单栏中选择智能运维中的弹性伸缩，进入弹性伸缩功能页面。 3. 在弹性伸缩页面的分区动态处理模块，可以选择开启或关闭自动化分区均衡。 4. 自动分区均衡支持两种策略，通过”配置“按钮可选择自定义时间或自动分析选择业务低峰期。 5. 单击”确定“，完成配置，再点击状态开关进行开启即可。 6. 可通过”查看“按钮或者直接进入智能运维中的后台任务管理功能页面查看自动化分区均衡调整记录详情。

创建Topic流控 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例的名称，进入实例详情页面。 步骤 5 在左侧导航栏单击“流控管理 > 流控列表”，进入流控列表页面。 步骤 6 在页面顶端单击“Topic”，进入“Topic”页签。 步骤 7 在页面左上角单击“创建流控”，弹出“创建流控”对话框。 步骤 8 设置流控参数。 表 流控参数说明 参数名称 说明 Topic名称 输入指定Topic名称，对此Topic进行流控。流控创建完后，无法修改“Topic名称”。 生产上限速率 设置生产上限速率，单位为MB/s。为空时，表示不设置速率。 消费上限速率 设置消费上限速率，单位为MB/s。为空时，表示不设置速率。 说明 “生产上限速率”和“消费上限速率”不可同时为空。 步骤 9 单击“确定”，跳转到“后台任务管理”页面，当流控任务的“状态”为“成功”时，表示流控创建成功。 进入“流控管理 > 流控列表”页面，在“Topic”页签中，在页面右上角的搜索框中输入新创建的流控名称，单击，查看新创建的流控。

本节介绍Kafka发送告警最佳实践 场景描述 如果您想要在消费组的消息堆积数超过阈值时，通过短信/邮件及时收到通知信息，可以参考本章节设置告警通知。 前提条件 已购买Kafka实例、创建Topic，并且已成功消费消息。 操作步骤 1. 登录Kafka实例控制台，单击待创建告警通知的实例名称，进入实例详情页。 2. 在左侧导航栏，选择“告警管理”，进入告警管理页面。 3. 点击“通知组”页签，创建联系人。 4. 点击“通知策略”页签，点击右上角“创建通知策略”按钮。 通知对象：选择前面流程中创建的联系人，在右侧选择联系方式邮件短信。 通知模板：可选择邮件、短信、翼连三种告警方式。 5. 点击“告警规则”页签，点击左上角“创建告警规则”按钮。 说明 告警名称：用于区别不同告警。 告警分组：可选择实例节点、主题和消费组，可根据具体需求选择，本文涉及情况应选择消费组。 告警指标：不同的告警分组包含不同的告警指标，本文涉及情况选择的是消费组中的堆积数告警。 告警条件：根据具体业务需求填写信息。 告警内容：不同的告警指标对应不同的告警内容，不建议改动。 持续时间：可选择马上告警，也可选择持续一段时间后才告警。 告警等级：可选择一般、次要、重要、紧急四个等级。 通知策略：选择前面流程中创建的通知策略。 通知频率：可选择隔多长时间告警一次。 6. 规则创建成功后可在告警规则页面查看其状态。

配置项 操作说明 示例 触发器类型 选择RocketMQ触发器。 RocketMQ触发器 名称 填写自定义的触发器名称。 rocketmqtrigger 版本或别名 默认值为LATEST，支持选择任意函数版本或函数别名。 LATEST RocketMQ 实例 选择已创建的RocketMQ实例。 Topic 选择已创建的RocketMQ实例的Topic。 Group ID 选择已创建的RocketMQ实例的Group ID。 消费位点 选择消息的消费位点，即触发器从RocketMQ实例开始拉取消息的位置。取值说明如下。 最新位点：从最新位点开始消费。 最早位点：从最早位点开始消费。 指定时间戳：从指定时间戳开始消费。 最新位点 调用方式 选择函数调用方式。 同步调用：指触发器消费topic消息后投递到函数是同步调用，会等待函数响应后继续下一个消息投递。 异步调用：指触发器消费topic消息后投递到函数是异步调用，不会等待函数响应，可以快速消费事件。 同步调用 用户ID RocketMQ实例用户ID，需要在RocketMQ控制台创建。 密钥 RocketMQ实例用户密钥，需要在RocketMQ控制台创建。 触发器启用状态 创建触发器后是否立即启用。默认选择开启，即创建触发器后立即启用触发器。 推送配置 批量推送条数：批量推送的最大值，积压值达到后立刻推送，取值范围为 [1, 10000]。 批量推送间隔：批量推送的最大时间间隔，达到后立刻推送，单位秒，取值[0,15]。默认0无需等待，数据直接推送。 推送格式：函数收到的事件格式，详情请查阅触发器事件消息格式。 重试策略 消息推送函数失败后重试的策略，共两种： 指数退避：指数退避重试，重试5次，重试周期为2，4，8，16，32（秒）。 线性退避：线性退避重试，重试5次，重试周期为1，2，3，4，5（秒）。 容错策略 当重试次数耗尽后仍然失败时的处理方式： 允许容错：当异常发生并超过重试策略配置时直接丢弃。 禁止容错：当异常发生并超过重试策略配置时继续阻塞执行。 死信队列 当容错策略为：允许容错时，可以额外开启死信队列。当开启死信队列时且异常发生并超过重试策略配置时，消息会被投递到指定的消息队列里，当前只支持投递到kafka和rocketmq

方案一：先迁生产，再迁消费 指先将生产消息的业务迁移到新的Kafka，原Kafka不会有新的消息生产。待原有Kafka实例的消息全部消费完成后，再将消费消息业务迁移到新的Kafka，开始消费新Kafka实例的消息。 1、将生产客户端的Kafka连接地址修改为新Kafka实例的连接地址。 2、重启生产业务，使得生产者将新的消息发送到新Kafka实例中。 3、观察各消费组在原Kafka的消费进度，直到原Kafka中数据都已经被消费完毕。 4、将消费客户端的Kafka连接地址修改为新Kafka实例的连接地址。 5、重启消费业务，使得消费者从新Kafka实例中消费消息。 6、观察消费者是否能正常从新Kafka实例中获取数据。 7、迁移结束。 本方案为业界通用的迁移方案，操作步骤简单，迁移过程由业务侧自主控制，整个过程中消息不会存在乱序问题， 适用于对消息顺序有要求的场景 。但是该方案中需要等待消费者业务直至消费完毕，存在一个时间差的问题，部分数据可能存在较大的端到端时延。 方案二：同时消费，后迁生产 指消费者业务启用多个消费客户端，分别向原Kafka和新Kafka实例消费消息，然后将生产业务切到新Kafka实例，这样能确保所有消息都被及时消费。 1、启动新的消费客户端，配置Kafka连接地址为新Kafka实例的连接地址，消费新Kafka实例中的数据。 说明 原有消费客户端需继续运行，消费业务同时消费原Kafka与新Kafka实例的消息。 2、修改生产客户端，Kafka连接地址改为新Kafka实例的连接地址。 3、重启生产客户端，将生产业务迁移到新Kafka实例中。 4、生产业务迁移后，观察连接新Kafka实例的消费业务是否正常。 5、等待原Kafka中数据消费完毕，关闭原有消费业务客户端。 6、迁移结束。 迁移过程由业务自主控制。本方案中消费业务会在一段时间内同时消费原Kafka和新Kafka实例。由于在迁移生产业务之前，已经有消费业务运行在新Kafka实例上，因此不会存在端到端时延的问题。但在迁移生产的开始阶段，同时消费原Kafka与新Kafka实例，会导致部分消息之间的生产顺序无法保证，存在消息乱序的问题。此场景 适用于对端到端时延有要求，却对消息顺序不敏感的业务 。 FAQ：如何将持久化数据也一起迁移 如果需要将原Kafka的已消费数据也迁移到Kafka专享实例，可以使用开源工具MirrorMaker，模拟成原Kafka的消费客户端，以及新Kafka实例的生产客户端，将Kafka所有消息数据迁移到新的Kafka实例。 需要注意的是，天翼云Kafka实例为3副本存储，因此建议实例存储空间为原业务的单副本消息存储的3倍。

本节介绍如何关闭Smart Connect。 如果不再使用Smart Connect相关功能，您可以关闭Smart Connect，释放资源。 关闭Smart Connect不会影响业务。 约束与限制 关闭Smart Connect后，实例会停止收取用于Smart Connect的代理费用，并自动删除用于Smart Connect的代理。 关闭Smart Connect后，再重新开启Smart Connect，已删除的Smart Connect任务无法找回，需要重新创建。 前提条件 已创建Kafka实例，且实例状态为“运行中”。 关闭Smart Connect前，请先删除所有的Smart Connect任务，否则无法关闭Smart Connect。此操作是为了防止关闭Smart Connect导致正在运行的Smart Connect任务丢失。 关闭Smart Connect 1、登录管理控制台。 2、在管理控制台左上角单击，选择Kafka实例所在的区域。 3、在管理控制台左上角单击，选择“应用服务 > 分布式消息服务 Kafka”，进入分布式消息服务Kafka专享版页面。 4、通过以下任意一种方法，关闭Smart Connect。 在待关闭Smart Connect的Kafka实例所在行，单击“更多 > 关闭Smart Connect”。 单击Kafka实例名称，进入实例详情页面。单击右上角的“更多 > 关闭Smart Connect”。 5、单击，将Smart Connect设置为关闭，单击“下一步”。 6、确认“Smart Connect”为关闭状态，单击“提交”。 结束

产品名称 可用性SLO 弹性云主机 99.995% 物理机 99% 弹性伸缩服务 99.95% 云硬盘 99.999% 弹性文件服务 99.9% 对象存储（经典I型） 99.99% 对象存储（原生版）I型 99.95% 媒体存储 99.99% 弹性IP 99% 弹性负载均衡 99.95% VPN连接 99.95% 分布式缓存Redis 99.95% 分布式消息RocketMQ 99.95% 分布式消息RabbitMQ 99.95% 分布式消息Kafka 99.95% CDN加速 99.90%

本节介绍了分布式消息服务RabbitMQ产品的实例规格。 （一）以下规格选型适用于14个节点 华东1、华北2、西南1、华南2、上海36、青岛20、长沙42、南昌5、武汉41、杭州7、西南2贵州、太原4、郑州5、西安7、呼和浩特3 注意 通用型规格已调整为白名单特性，如需了解该规格参数请联系技术支持。 云原生引擎已调整为白名单特性，如需了解该引擎请联系技术支持。 单机版实例面向用户体验和业务测试场景，无法保证性能和高可用。如果需要在生产环境使用RabbitMQ实例，建议购买集群版实例。 TPS参考值，是指以2K大小的消息为例的每秒处理消息条数，测试场景为不开启持久化的非镜像队列，实时生产实时消费，队列无积压。此数据仅供参考，在实际业务中，队列数、消息堆积、连接数、channel、消费者数、镜像队列、优先级队列、消息持久化和exchange类型等因素会对TPS造成较大影响，不一定能够达到上述性能。 1、RabbitMQ实例规格 RabbitMQ兼容开源RabbitMQ 3.8版本，提供集群和单机两种部署架构实例，支持X86和ARM计算，主机类型为Intel、海光和鲲鹏计算增强型，节点可选择3节点、5节点、7节点、9节点。 存储类型可选择高IO（SAS）、超高IO（SSD），默认选择超高IO。 2、目前所有的可选规格如下：

顺序消息是分布式消息服务RocketMQ版提供的一种严格按照顺序来发布和消费的消息类型。 顺序消息分为全局顺序消息和分区顺序消息： 全局顺序消息：对于指定的一个Topic，将队列数量设置为1，这个队列内所有消息按照严格的先入先出FIFO（First In First Out）的顺序进行发布和订阅。 分区顺序消息：对于指定的一个Topic，同一个队列内的消息按照严格的FIFO顺序进行发布和订阅。生产者指定分区选择算法，保证需要按顺序消费的消息被分配到同一个队列。 全局顺序消息和分区顺序消息的区别仅为队列数量不同，代码没有区别。 收发消息前，请参考收集连接信息收集RocketMQ所需的连接信息。 准备环境 1. 在命令行输入python，检查是否已安装Python。得到如下回显，说明Python已安装。 PS C:> python Python 3.9.9 (tags/v3.9.9:ccb0e6a, Nov 15 2021, 18:08:50) [MSC v.1929 64 bit (AMD64)] on win32 Type "help", "copyright", "credits" or "license" for more information. 如果未安装Python，请使用以下命令安装： pip install rocketmqclientpython 2. 安装librocketmq库和rocketmqclientpython。 说明 建议下载rocketmqclientcpp2.2.0，获取librocketmq库。 3. 将librocketmq.so添加到系统动态库搜索路径。 1. 查找librocketmq.so的路径。 find / name librocketmq.so 2. 将librocketmq.so添加到系统动态库搜索路径。 ln s /查找到的librocketmq.so路径/librocketmq.so /usr/lib sudo ldconfig 以下示例代码中的参数说明如下，请参考收集连接信息获取参数值。 GROUP：表示消费组名称。 ENDPOINT：表示实例连接地址和端口。 TOPIC：表示Topic名称。

对比项 开源自建 分布式消息服务Kafka 低成本运维 需要专业人员资源规划、部署、运维。 一键开通，全托管。提供多种规格，按需使用，支持一键式节点数、磁盘存储空间和节点规格扩容。 分区规模 千级分区稳定性下降。 支持万级分区稳定写入。 消息查询 命令行可以消费，但无法根据位点或者时间直接定位到具体的消息。 控制台可视化按时间或者位点直接查看消息。 ACL访问控制 命令行，配置复杂。 灵活配置，一键生效。 可视化配置 命令行，配置复杂。 控制台全方位可视化配置管理。 运维监控 缺乏配套的监控运维能力。 提供全引擎指标可视化监控，告警及时发现问题。 集群巡检 命令行，配置复杂。 引擎状态、主题一键健康检测、启停。 稳定可靠 需要自己开发或基于开源实现，开发成本高昂，无法保证业务可靠运行。 支持跨AZ部署，提升可靠性。 安全保证 需要自行进行安全加固。 VPC隔离，支持SSL通道加密。 简单易用 无，需要自己开发。 提供简单的实例管理RESTful API，使用门槛低。

消息重试 对于有序消息：有序消息不能跳跃签收，当消费者消费消息失败后，消息队列RocketMQ会自动不断进行消息重试 （每次间隔时间为1秒），此时应用会出现消息消费被阻塞的情况。因此建议使用有序消息时，务必保证应用能够及时监控并处理消费失败的情况，避免阻塞现象的发生。 对于无序消息：消息队列RocketMQ默认允许每条消息最多重试16次。 每次重试的间隔时间如下： 第几次重试 与上次重试的间隔时间 1 10秒 2 30秒 3 1分钟 4 2分钟 5 3分钟 6 4分钟 7 5分钟 8 6分钟 9 7分钟 10 8分钟 11 9分钟 12 10分钟 13 20分钟 14 30分钟 15 1小时 16 2小时 如果消息重试16次后仍然失败，消息将不再投递。如果严格按照上述重试时间间隔计算，某条消息在一直消费失败的前提下，将会在接下来的4小时46分钟之内进行16次重试，超过这个时间范围消息将不再重试投递。 消息过滤 消费者订阅了某个Topic后，消息队列RocketMQ会将该主题中的所有消息投递给消费者。若消费者只需要关注部分消息，可通过设置过滤条件在消息队列RocketMQ版服务端进行过滤，只获取到需要关注的消息子集，避免接收到大量无效的消息。 消息过滤主要通过以下几个关键流程实现： 生产者：生产者在初始化消息时预先为消息设置一些属性和标签，用于后续消费时指定过滤目标。 消费者：消费者在初始化及后续消费流程中向服务端上报需要订阅指定主题的哪些消息，即过滤条件。 服务端：消息队列RocketMQ服务端根据消费者上报的过滤条件的表达式进行匹配，将符合条件的消息投递给消费者进行消费。 消息队列RocketMQ支持两种过滤方式： 通过Tag进行过滤：生产者在发送消息时，设置消息的Tag标签，消费者通过 Tag标签指定需要消费的信息。 通过SQL属性过滤：通过生产者为消息设置的属性（Key）及属性值（Value）进行匹配。生产者在发送消息时可设置多个属性，消费者订阅时可设置SQL语法的过滤表达式过滤多个属性。

说明 分布式消息服务Kafka旧资费根据实例规格分为基础版和高级版，存储空间按照不同类型收费。 目前在 芜湖2、上海7、重庆2、乌鲁木齐27、石家庄20、内蒙6、北京5 资源池开放订购。 实例价格如下： 规格 标准资费（元/月） 标准资费（元/小时） 基础版（100 MB/s） 1300 2.71 高级版（300 MB/s ） 2100 4.38 存储空间价格如下： 存储类型 标准资费（元/G/月） 标准资费（元/G/小时） 普通 IO 0.45 0.0008 高 IO 0.6 0.0013 超高 IO 1.5 0.0032

说明 分布式消息服务Kafka如下资源池支持Kafka2.8、3.6版本引擎，提供集群和单机两种规格实例，支持X86和ARM计算CPU架构类型的计算增强型主机，可选350代理数量。 目前在 华东1、华北2、西南1、华南2、上海36、青岛20、长沙42、南昌5、武汉41、杭州7、西南2贵州、太原4、郑州5、西安7、呼和浩特3 资源池开放订购。 上述资源池订购和续订可享受1年83折，2年7折，3年5折优惠。 价格计算公式 分布式消息服务Kafka费用由实例费用和存储费用两部分组成，两者单价如下表所示，计费公式为： 实例费用实例规格单价 代理数量，单机版代理数量为1。 存储费用存储类型单价 代理数量 单节点存储空间GB大小，单机版代理数量为1。 实例规格单价 Intel计算增强型 规格名称 实例单价（单个节点） 规格名称 按需标准价格(元/小时) 包月标准价格(元/月) Kafka.2u4g.cluster 0.98 441 Kafka.4u8g.cluster 2.24 1008 Kafka.8u16g.cluster 4.86 2187 Kafka.12u24g.cluster 7.38 3321 Kafka.16u32g.cluster 9 4050 Kafka.24u48g.cluster 15.12 6804 Kafka.32u64g.cluster 20.16 9072 Kafka.48u96g.cluster 30.24 13608 Kafka.64u128g.cluster 40.32 18144