分类 组件 版本 微服务 org.springframework.cloud:springclouddependencies 2021.0.8 微服务 org.springframework.boot:springbootstarterweb 2.7.12 微服务 org.springframework.cloud:springcloudstarteropenfeign 3.1.8 微服务 org.apache.dubbo:dubbo 2.7.13 负载均衡 org.springframework.cloud:springcloudloadbalancer 3.1.7 微服务注册中心 com.alibaba.cloud:springcloudstarteralibabanacosdiscovery 2021.1 消息队列 org.apache.rocketmq:rocketmqclient 4.8.0 数据库 mysql:mysqlconnectorjava 5.1.40及以上版本 8.0.15及以上版本 数据库 org.postgresql:postgresql 42.5.0及以上版本

指标ID 指标名称 指标含义 取值范围 测量对象 监控周期（原始指标） topicproducemsg 消息生产数 Topic一分钟收到的消息数单位：Count >0 RocketMQ实例队列 1分钟 topicconsumemsg 消息消费数 Topic一分钟被消费的消息数单位：Count >0 RocketMQ实例队列 1分钟 topicproducerate 消息生产速率 Topic每秒收到的消息数单位：Count/s >0 RocketMQ实例队列 1分钟 topicconsumerate 消息消费速率 Topic每秒被消费的消息数单位：Count/s >0 RocketMQ实例队列 1分钟 topicbytesinrate 生产流量 当前主题的生产流量单位：Byte/s说明：2022年5月16号及以后购买的实例，支持此监控项。 >0 RocketMQ实例队列 1分钟 topicbytesoutrate 消费流量 当前主题的消费流量单位：Byte/s说明：2022年5月16号及以后购买的实例，支持此监控项。 >0 RocketMQ实例队列 1分钟

分布式消息服务RocketMQ版的事务消息支持在业务逻辑与发送消息之间提供事务保证，通过两阶段的方式提供对事务消息的支持，事务消息交互流程如图1所示。 图1 事务消息交互流程 事务消息生产者首先发送半消息，然后执行本地事务。如果执行成功，则发送事务提交，否则发送事务回滚。服务端在一段时间后如果一直收不到提交或回滚，则发起回查，生产者在收到回查后重新发送事务提交或回滚。消息只有在提交之后才投递给消费者，消费者对回滚的消息不可见。 收发事务消息前，请参考收集连接信息收集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作为一款热门的消息队列中间件，具备高效可靠的消息异步传递机制，主要用于不同系统间的数据交流和传递，在企业解决方案、金融支付、电信、电子商务、社交、即时通信、视频、物联网、车联网等众多领域都有广泛应用。 异步通信 将业务中属于非核心或不重要的流程部分，使用消息异步通知的方式发给目标系统，这样主业务流程无需同步等待其他系统的处理结果，从而达到系统快速响应的目的。 如网站的用户注册场景，在用户注册成功后，还需要发送注册邮件与注册短信，这两个流程使用Kafka消息服务通知邮件发送系统与短信发送系统，从而提升注册流程的响应速度。 图 串行发送注册邮件与短信流程 图 借助消息队列异步发送注册邮件与短信流程 错峰流控与流量削峰 在电子商务系统或大型网站中，上下游系统处理能力存在差异，处理能力高的上游系统的突发流量可能会对处理能力低的某些下游系统造成冲击，需要提高系统的可用性的同时降低系统实现的复杂性。电商大促销等流量洪流突然来袭时，可以通过队列服务堆积缓存订单等信息，在下游系统有能力处理消息的时候再处理，避免下游订阅系统因突发流量崩溃。消息队列提供亿级消息堆积能力，3天的默认保留时长，消息消费系统可以错峰进行消息处理。 另外，在商品秒杀、抢购等流量短时间内暴增场景中，为了防止后端应用被压垮，可在前后端系统间使用Kafka消息队列传递请求。 图 消息队列应对秒杀大流量场景

本节介绍分布式消息服务Kafka操作类常见问题 消息在kafka保留多长时间？ 消息保存72小时，超过72小时的消息将会被删除。 Kafka可以创建多少个主题？ Kafka基础版可以创建50个主题、Kafka高级版可以创建100个主题。 如果想消费已经被消费过的数据？ consumer是底层采用的是一个阻塞队列，只要一有producer生产数据，那consumer就会将数据消费。当然这里会产生一个很严重的问题，如果你重启一消费者程序，那你连一条数据都抓不到，但是log文件中明明可以看到所有数据都好好的存在。换句话说，一旦你消费过这些数据，那你就无法再次用同一个groupid消费同一组数据了。针对这种情况，你可在控制台重置消费组消费点（3天内）。 是否需要预先创建消费组 消费组和消费组订阅主题关系虽然业务应用客户端接入时可自动创建，但建议都先预先创建做好管理。 出现“Not authorized to access group”的错误信息 没有创建消费组时会遇到此报错信息，创建消费组可解决此问题。 为什么PHP发送延时比较长？ PHP发送延时比较长是PHP的语言特性导致的。PHP每次发送时，都会重新初始化一个KafkaProducer对象，这个初始化会进行各种操作，包括连接各个Broker、更新元数据等，在VPC内耗时100ms以上，在公网可能耗时500ms以上。相比之下，Java会复用KafkaProducer，发送延迟较低。 哪里可以找到生产消费消息的示例 最佳实践 生产者实践、消费者实践。

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

本文主要介绍产品优势。 分布式消息服务Kafka完全兼容开源社区版本，旨在为用户提供便捷高效的消息队列。业务无需改动即可快速迁移上云，为您节省维护和使用成本。 一键式部署，免去集群搭建烦恼 您只需要在实例管理界面选好规格配置，提交订单。后台将自动创建部署完成一整套Kafka实例。 兼容开源，业务零改动迁移上云 兼容社区版Kafka的API，具备原生Kafka的所有消息处理特性。 业务系统基于开源的Kafka进行开发，只需加入少量认证安全配置，即可使用分布式消息服务Kafka，做到无缝迁移。 说明 Kafka实例兼容开源社区Kafka 1.1.0、2.3.0和2.7版本。在客户端使用上，推荐使用和服务端版本一致的版本。 安全保证 独有的安全加固体系，提供业务操作云端审计，消息存储加密等有效安全措施。 在网络通信方面，除了提供SASL（Simple Authentication and Security Layer）认证，还借助虚拟私有云（VPC）和安全组等加强网络访问控制。 数据高可靠 Kafka实例支持消息持久化，多副本存储机制。副本间消息同步、异步复制，数据同步或异步落盘多种方式供您自由选择。 集群架构与跨AZ部署，服务高可用 Kafka后台为多集群部署，支持故障自动迁移和容错，保证业务的可靠运行。 Kafka实例支持跨AZ部署，代理部署在不同的AZ，进一步保障服务高可用。不同AZ之间基于Kafka ISR（insync replica）进行数据同步，Topic需要选择数据多副本并且将不同副本分布到不同的ISR上，在ISR正常同步状态下，故障RPO（Recovery Point Objective）趋近于0。 无忧运维 云服务平台提供一整套完整的监控告警等运维服务，故障自动发现和告警，避免724小时人工值守。Kafka实例自动上报相关监控指标，如分区数、主题数、堆积消息数等，并支持配置监控数据发送规则，您可以在第一时间通过短信、邮件等获得业务消息队列的运行使用和负载状态。 海量消息堆积与弹性扩容 内建的分布式集群技术，使得服务具有高度扩展性。分区数可配置多达100个，存储空间弹性扩展，保证在高并发、高性能和大规模场景下的访问能力，轻松实现百亿级消息的堆积和访问能力。 多规格灵活选择 Kafka实例的带宽与存储资源可灵活配置，并且自定义Topic的分区数、副本数。

接口功能介绍 查看一分钟发布消息量 接口约束 无 URI GET /v3/api/getPubTpm 路径参数 无 Query参数 参数 是否必填 参数类型 说明 示例 下级对象 prodInstId 是 String 实例ID 445d30ce3d5d40a2a79dc444a8c83859 请求参数 请求头header参数 参数 是否必填 参数类型 说明 示例 下级对象 regionId 是 String 资源池编码 bb9fdb42056f11eda1610242ac110002 华东1 请求体body参数 无 响应参数 参数 参数类型 说明 示例 下级对象 message String 错误信息 statusCode String 800成功 其他失败 returnObj Object 返回数据 returnObj error String 错误码 表 returnObj 参数 参数类型 说明 示例 下级对象 prodInstId String 实例ID tpm Integer 一分钟生产消息量 枚举参数 无 请求示例 请求url 请求头header { "regionId":"bb9fdb42056f11eda1610242ac110002" } 请求体body 无

本章节介绍了分布式消息服务RocketMQ的相关术语解释。 主题（Topic） 消息关联的基础逻辑单元。消息生产与消费时的基础单位。 队列（Queue） 一个主题由多个队列组成。队列数越大消费的并发度越大。 生产者（Producer） 消息写入的触发者，负责将消息推送到服务端。 生产者组（Producer Group） 同一类生产者的集合，这类生产者发送同一类消息且发送逻辑一致。 消费者（Consumer） 接收消息的对象，负责从服务端获取消息。 消费组（Consumer Group） 多个消费者组成同一个消费组，同一消费组内的消费者具有相同的消费属性。 代理（Broker） 一组节点构成的一个业务集群。 NameServer 存储元数据信息的轻量级注册中心。生产者/消费者在生产/消费消息前，需要先从NameServer获取元数据。

本文主要介绍分布式消息服务RabbitMQ的实例规格。 RabbitMQ实例规格 RabbitMQ实例兼容开源RabbitMQ 3.8.35、3.7.17，实例类型包括单机和集群，实例规格请参考下表。 说明 为了保证稳定性，服务端限制了单条消息的最大长度为50MB，请勿发送大于此长度的消息。 下表中TPS，是指以2KB大小的消息为例的每秒处理消息条数，测试场景为不开启持久化的非镜像队列，实时生产实时消费，队列无积压。此数据仅供参考，生产使用需要以实际压测性能为准。 服务端的性能主要跟以下因素相关：队列数、消息堆积、连接数、channel、消费者数、镜像队列、优先级队列、消息持久化和exchange类型等，在选择实例规格时，请根据业务模型压测结果选择。 一条连接最多可以开启2047个channel。 表1 RabbitMQ实例规格 型号 代理数 存储空间范围 TPS参考值 单个代理最大消费者数 单个代理建议队列数 单个代理最大连接数 rabbitmq.2u4g.single 1 100GB~30000GB 10000 20000 200 3000 rabbitmq.4u8g.single 1 100GB~30000GB 20000 30000 400 4500 rabbitmq.8u16g.single 1 100GB~30000GB 35000 50000 800 7500 rabbitmq.16u32g.single 1 100GB~30000GB 45000 80000 1600 12000 rabbitmq.24u48g.single 1 100GB~30000GB 50000 100000 2400 15000 rabbitmq.2u4g.cluster 3/5/7 3/5/7100GB~30000GB 30000~70000 20000 200 3000 rabbitmq.4u8g.cluster 3/5/7 3/5/7100GB~30000GB 45000~80000 30000 400 4500 rabbitmq.8u16g.cluster 3/5/7 3/5/7100GB~30000GB 85000~120000 50000 800 7500 rabbitmq.12u24g.cluster 3/5/7 3/5/7100GB~30000GB 100000~150000 60000 1200 10000 rabbitmq.16u32g.cluster 3/5/7 3/5/7100GB~30000GB 130000~180000 80000 1600 12000 rabbitmq.24u48g.cluster 3/5/7 3/5/7100GB~30000GB 150000~200000 100000 2400 15000

Kafka作为一款热门的消息队列中间件，具备高效可靠的消息异步传递机制，主要用于不同系统间的数据交流和传递，在企业解决方案、金融支付、电信、电子商务、社交、即时通信、视频、物联网、车联网等众多领域都有广泛应用。 异步通信 将业务中属于非核心或不重要的流程部分，使用消息异步通知的方式发给目标系统，这样主业务流程无需同步等待其他系统的处理结果，从而达到系统快速响应的目的。 如网站的用户注册场景，在用户注册成功后，还需要发送注册邮件与注册短信，这两个流程使用Kafka消息服务通知邮件发送系统与短信发送系统，从而提升注册流程的响应速度。 图 串行发送注册邮件与短信流程 图 借助消息队列异步发送注册邮件与短信流程 错峰流控与流量削峰 在电子商务系统或大型网站中，上下游系统处理能力存在差异，处理能力高的上游系统的突发流量可能会对处理能力低的某些下游系统造成冲击，需要提高系统的可用性的同时降低系统实现的复杂性。电商大促销等流量洪流突然来袭时，可以通过队列服务堆积缓存订单等信息，在下游系统有能力处理消息的时候再处理，避免下游订阅系统因突发流量崩溃。消息队列提供亿级消息堆积能力，3天的默认保留时长，消息消费系统可以错峰进行消息处理。 另外，在商品秒杀、抢购等流量短时间内暴增场景中，为了防止后端应用被压垮，可在前后端系统间使用Kafka消息队列传递请求。 图 消息队列应对秒杀大流量场景 日志同步 在大型业务系统设计中，为了快速定位问题，全链路追踪日志，以及故障及时预警监控，通常需要将各系统应用的日志集中分析处理。 Kafka设计初衷就是为了应对大量日志传输场景，应用通过可靠异步方式将日志消息同步到消息服务，再通过其他组件对日志做实时或离线分析，也可用于关键日志信息收集进行应用监控。 日志同步主要有三个关键部分：日志采集客户端，Kafka消息队列以及后端的日志处理应用。 1. 日志采集客户端，负责用户各类应用服务的日志数据采集，以消息方式将日志“批量”“异步”发送Kafka客户端。 Kafka客户端批量提交和压缩消息，对应用服务的性能影响非常小。 2. Kafka将日志存储在消息文件中，提供持久化。 3. 日志处理应用，如Logstash，订阅并消费Kafka中的日志消息，最终供文件搜索服务检索日志，或者由Kafka将消息传递给Hadoop等其他大数据应用系统化存储与分析。 图 日志同步示意图 上图中Logstash、ElasticSearch分别为日志分析和检索的开源工具，Hadoop表示大数据分析系统。

接口功能介绍 查看一分钟消费消息量 接口约束 无 URI GET /v3/api/getSubTpm 路径参数 无 Query参数 参数 是否必填 参数类型 说明 示例 下级对象 prodInstId 是 String 实例ID 445d30ce3d5d40a2a79dc444a8c83859 请求参数 请求头header参数 参数 是否必填 参数类型 说明 示例 下级对象 regionId 是 String 资源池编码 bb9fdb42056f11eda1610242ac110002 华东1 请求体body参数 无 响应参数 参数 参数类型 说明 示例 下级对象 message String 错误信息 statusCode String 800成功 其他失败 returnObj Object 返回数据 returnObj error String 错误码 表 returnObj 参数 参数类型 说明 示例 下级对象 prodInstId String 实例ID tpm Integer 一分钟消费消息量 枚举参数 无 请求示例 请求url 请求头header { "regionId":"bb9fdb42056f11eda1610242ac110002" } 请求体body 无

本文介绍如何应用事件总线EventBridge的事件流功能实现分布式消息服务RocketMQ的消息路由。 前提条件 开通事件总线EventBridge并授权。 开通分布式消息服务RocketMQ并创建最少两个主题。 背景信息 事件流作为更轻量、实时端到端的流式事件通道，提供轻量级的流式数据的过滤和转换的能力，在不同的数据仓库之间、数据处理程序之间、数据分析和处理系统之间进行数据同步。源端分布式消息服务RocketMQ生产的消息可以通过事件流这个通道被路由到目标端的分布式消息服务RocketMQ。 步骤一：创建事件流 1. 登录事件总线EventBridge控制台。 2. 在左侧导航栏，单击事件流。 3. 在事件流页面，单击创建事件流。 4. 在创建事件流面板，设置任务名称和描述，配置以下参数，然后单击保存。 a.在Source（源）配置向导，选择数据提供方为分布式消息服务RocketMQ，设置以下参数，然后单击下一步。 参数 说明 示例 实例名称 前提条件中已创建的分布式消息服务RocketMQ版实例。 xxx Topic 当前实例中的Topic。 topic1 Group 消费组名。 快速创建：自动创建以GIDEVENTBRIDGExxx命名的Group ID。 使用已有：选择当前实例中已创建的Group，请不要与已有业务的Group混用，以免影响已有的消息收发。 group1 消费位点 开始消费的位置。 最新位点：从最新位点开始消费。 最新位点 Tag 用于过滤消息的Tag值，非必填。 tag1 b.在Filtering（过滤）配置向导，设置事件过滤规则，单击下一步。 c.在Sink（目标）配置向导，选择服务类型为分布式消息服务RocketMQ，配置以下参数，单击保存，如图1所示。 参数 说明 示例 实例 选择分布式消息服务RocketMQ实例。 instancexxx Topic 选择RocketMQ实例的Topic。 topic1 消息体 选择消息体（Body）的内容，更多参考“事件内容转换”。 完整事件 自定义属性 选择自定义属性（Properties）的内容，更多参考“事件内容转换”。 空 索引 选择索引（Keys）的内容，更多参考“事件内容转换”。 空 标签 选择标签（Tags）的内容，更多参考“事件内容转换”。 空 图1 创建事件流时选择服务类型为分布式消息服务RocketMQ的事件目标 5. 创建事件流后，会有30秒~60秒的延迟时间，您可以在事件流页面的状态栏查看启动进度。

有序消息 消费消息的顺序要同发送消息的顺序一致，在RocketMQ中，主要有两种有序消息。 普通有序消息 有序消息的一种，在正常情况下可以保证完全的顺序消息，但是一旦发生通信异常，Broker重启，由于队列总数发生变化，哈希取模后定位的队列会变化，产生短暂的消息顺序不一致。如果业务能容忍在集群异常情况（如某个Broker宕机或者重启）下，消息短暂的乱序，使用普通顺序方式比较合适。 严格有序消息 有序消息的一种。无论正常异常情况都能保证顺序，但是牺牲了分布式Failover特性，即Broker集群中只要有一台机器不可用，则整个集群都不可用（或者影响hash值对应队列的使用），服务可用性大大降低。如果服务器部署为同步双写模式，此缺陷可通过备机自动切换为主避免，不过仍然会存在几分钟的服务不可用。若业务能容忍短暂乱序，推荐普通有序消费。

在RocketMQ中，队列数直接影响到消费者实例数的上限，同一消费组消费者实例数的上限等于队列数，对于集群消费的情况，需考虑队列数的设置。 在RocketMQ中，队列能分布到不同的Broker上，是RocketMQ分布式的基础。Queue分布在Broker中，则能使用Broker的资源，包括存储、IO等，一般情况下，分布在某个Broker上的Queue比例越大，则占用此Broker的资源越多，Topic中的Queue分布到的Broker数量越多，则性能越好、存储越大。若Broker的所在机器性能不同，可以通过调整Queue数量，达到资源调优的目的，在应用设计时，需要充分利用上述特性。 在Push消费模式中，API会默认为每个队列预拉取消息1000条，若队列数过大、或者单条消息包体过大，则需要考虑设置减少预拉数量，防止预拉消息过大导致内存溢出。

本章节主要介绍分布式消息服务RabbitMQ的单一活跃消费者特性。 使用场景 单一活跃消费者（Single Active Consumer）表示队列中可以注册多个消费者，但是只允许一个消费者消费消息，只有在此消费者出现异常时，才会自动转移到另一个消费者进行消费。单一活跃消费者适用于需要保证消息消费顺序性，同时提供高可靠能力的场景。 说明 分布式消息服务RabbitMQ3.8.35版本才提供单一活跃消费者特性。 图1 单一活跃消费者消费流程 如图1所示，Producer生产9条消息，由于队列设置了单一活跃消费者特性，只有Consumer 1在消费消息。 更多关于单一活跃消费者的说明，请参考Single Active Consumer。 配置方法 在声明队列时，可以配置单一活跃消费者，只需要将队列的“xsingleactiveconsumer”参数设置为“true”。 以下示例演示在Java客户端设置单一活跃消费者。 Channel ch ...; Map arguments newHashMap (); arguments.put("xsingleactiveconsumer", true); ch.queueDeclare("myqueue", false, false, false, arguments); 以下示例演示在 RabbitMQ WebUI页面设置单一活跃消费者。 图2 设置单一活跃消费者 设置完成后，在“Queues”页面查看队列特性是否包含单一活跃消费者。如图3所示，“SAC”即单一活跃消费者。 图3 查看队列特性

尊敬的天翼云客户，分布式消息服务MQTT自2025年12月27日起，订购和续订2年、3年选项默认不开放，调整为白名单特性。 调整时间 2025年12月27日 影响范围 所有区域 调整影响 新订购和续订的实例默认不开放2年、3年选项，您可以选择1年包年选项，如仍需要23年包周期选项，请联系技术支持开通后使用。 已购买2年、3年且还在服务期间的实例仍可继续正常使用不受影响。

本节介绍了在RabbitMQ实例中如何创建和删除交换器。 背景信息 生产者将消息发送到交换器，由交换器将消息路由到一个或多个队列中（或者丢弃）。交换器根据Routing Key和Binding Key将消息路由到Queue。不同类型的交换器的路由规则不同。 操作步骤 新建交换器 1.登录管理控制台。 2.进入RabbitMQ管理控制台。 3.在实例列表页在操作列，目标实例行点击“管理”。 4.点击“交换器管理”后，点击“新建”按钮。 5.点击“新建”后出现以下窗口，选择虚拟主机，添加交换器名字，选择交换器类型和其他参数，然后点击“确定”即可新建交换器。 各参数说明如下 参数 描述 虚拟主机 选择创建交换器所属的虚拟主机 名称 交换器名称。以amq.开头的为保留字段，因此不能使用。例如：。 类型 Exchange类型。 是否持久化 交换器是否持久化到磁盘 是否自动删除 如果是，交换器将在至少一个队列或交换器绑定到该交换器，然后所有队列或交换器都已解除绑定时删除。 是否内置 如果是，客户端不能直接发布到这个交换器。它只能与其他交换器绑定使用。 其他参数 Alternate exchange：备份交换器是为了实现没有路由到队列的消息，声明交换机的时候添加属性alternateexchange，声明一个备用交换机，一般声明为fanout类型，这样交换机收到路由不到队列的消息就会发送到备用交换机绑定的队列中。 其中交换机类型如下表所示 交换机类型 说明 Direct 完全根据key进行投递的叫做Direct交换机。如果Routing key匹配, 那么Message就会被传递到相应的queue中。其实在queue创建时，它会自动的以queue的名字作为routing key来绑定那个exchange。例如，绑定时设置了Routing key为”abc”，那么客户端提交的消息，只有设置了key为”abc”的才会投递到队列。 Fanout 不需要key的叫做Fanout交换机。它采取广播模式，一个消息进来时，投递到与该交换机绑定的所有队列。 Topic 对key进行模式匹配后进行投递的叫做Topic交换机。比如符号”

KafkaConsumer 总消费次数&总消费字节数 显示总消费次数和总消费字节数的趋势图。 总消费次数：这个指标表示消费者从Kafka中成功读取并处理的消息数量。它是一个计数器，用于跟踪消费者在其生命周期内接收到的消息总数。这对于监控消费者的工作负载和性能非常有用，可以帮助开发者了解消费者是否有效地从生产者那里获取了数据。 总消费字节数：这个指标表示消费者从Kafka中成功读取的数据总量，以字节为单位。它考虑了每条消息的大小，并累加起来给出一个总和。这对于评估网络带宽使用情况、存储需求以及数据传输效率等方面非常重要。例如，如果一个主题有20个分区和5个消费者，每个消费者需要至少4MB的可用内存来接收记录。因此，通过监控总消费字节数，可以更好地理解数据流的规模和处理速度。 这两个指标共同提供了对Kafka消费者行为的全面视图，使得开发者能够优化系统性能，确保数据处理的高效性和稳定性。 通过消息队列接收消息 显示topic列表，表头如下。 topic：是指消息队列中接收到的特定主题（topic）的消息。主题用于将相关的消息进行归类和分组。 调用次数：是指在一段时间内，通过消息队列接收消息的总调用次数。每次调用消息队列的接收操作（例如KafkaConsumer的poll()方法）并成功获取一条消息都会被计算为一次调用。 平均响应时间(ms)：是指从发起接收请求到接收到消息的平均时间。它表示了接收消息的速度和效率。 错误数：表示在接收消息的过程中发生的错误次数。这些错误可能包括接收消息失败、网络连接问题或其他异常情况。 最慢调用(ms)：是指在一段时间内，接收消息过程中最耗时的一次调用的时间。它反映了接收消息中的性能瓶颈或延迟情况。 操作：点击详情，出现弹层显示调用次数、平均响应时间(ms)、错误数在筛选时间段内的趋势图。 这些指标共同构成了对消息队列系统性能和可靠性的全面评估，有助于管理员和开发者优化系统配置，提高系统的稳定性和效率。

本文主要介绍DMS for Kafka 消费者poll的优化。 场景介绍 在DMS提供的原生Kafka SDK中，消费者可以自定义拉取消息的时长，如果需要长时间的拉取消息，只需要把poll(long)方法的参数设置合适的值即可。但是这样的长连接可能会对客户端和服务端造成一定的压力，特别是分区数较多且每个消费者开启多个线程的情况下。 如图所示，Kafka队列含有多个分区，消费组中有多个消费者同时进行消费，每个线程均为长连接。当队列中消息较少或者没有时，连接不断开，所有消费者不间断地拉取消息，这样造成了一定的资源浪费。 图 Kafka消费者多线程消费模式 优化方案 在开了多个线程同时访问的情况下，如果队列里已经没有消息了，其实不需要所有的线程都在poll，只需要有一个线程poll各分区的消息就足够了，当在polling的线程发现队列中有消息，可以唤醒其他线程一起消费消息，以达到快速响应的目的。如图所示。 这种方案适用于对消费消息的实时性要求不高的应用场景。如果要求准实时消费消息，则建议保持所有消费者处于活跃状态。 图 优化后的多线程消费方案 说明 消费者（Consumer）和消息分区（Partition）并不强制数量相等，Kafka的poll(long)方法帮助实现获取消息、分区平衡、消费者与Kafka broker节点间的心跳检测等功能。 因此在对消费消息的实时性要求不高场景下，当消息数量不多的时候，可以选择让一部分消费者处于wait状态。

介绍分布式消息服务MQTT认证鉴权内容。 认证鉴权 AK/SK认证：通过AK（Access Key ID）/SK（Secret Access Key）加密调用请求。 AK/SK认证就是使用AK/SK对请求进行签名，在请求时将签名信息添加到消息头，从而通过身份认证。 AK（Access Key ID）：访问密钥ID。与私有访问密钥关联的唯一标识符；访问密钥ID和私有访问密钥一起使用，对请求进行加密签名。 SK（Secret Access Key）：私有访问密钥。与访问密钥ID结合使用，对请求进行加密签名，可标识发送方，并防止请求被修改。 使用AK/SK认证时，您可以基于签名算法使用AK/SK对请求进行签名，也可以使用专门的签名SDK对请求进行签名。详细的签名方法和SDK使用方法请参见API签名指南。 说明： 签名SDK只提供签名功能，与服务提供的SDK不同，使用时请注意。

产品分类 产品名称 存储 混合存储网关、云硬盘、弹性文件服务、媒体存储、云备份、云硬盘备份、对象存储、云主机备份、海量文件服务 OceanFS、云容灾、对象存储（经典版）I型、并行文件 网络与CDN 弹性IP、天翼云SDWAN、NAT网关、VPN连接、云间高速、虚拟私有云、弹性负载均衡、共享流量包、智能边缘云、云专线CDA、应用加速、边缘安全加速平台、CDN加速、智能DNS、VPC终端节点、全站加速、科研助手 计算 弹性云主机、天翼云电脑（政企版）、函数计算、弹性高性能计算、镜像服务、弹性伸缩服务、物理机、云骁智算平台、数据加密 数据库 关系数据库SQL Server版、关系数据库PostgreSQL版、分布式关系型数据库、关系数据库MySQL版、数据传输服务、文档数据库服务、数据管理服务、分布式缓存服务Redis版、时序数据库Influx版、分布式融合数据库HTAP、云数据库ClickHouse 监控与管理 云日志服务、应用性能监控 应用服务 API网关、EasyCoding敏捷开发平台、软件开发生产线CodeArts 数据库 关系型数据库MySQL版（CTRDS）、关系型数据库PostgreSQL版、关系型数据库SQL Server版、云数据库GaussDB、分布式关系型数据库、分布式融合数据库HTAP、文档数据库服务、时序数据库Influx版、分布式缓存服务Redis版 人工智能 AI能力开放平台、慧聚一站式智算服务平台 大数据 云搜索服务、大数据管理平台 DataWings、翼MapReduce 容器与中间件 应用服务网格、弹性容器实例、分布式容器云平台、容器镜像服务、分布式消息服务RabbitMQ、分布式消息服务RocketMQ、微服务应用平台MSAP、云容器引擎、分布式消息服务Kafka、微服务引擎注册配置中心、微服务引擎MSE、应用性能监控apm、分布式容器云平台CCSE ONE 安全及管理 云防火墙（原生版）、统一身份认证、密钥管理、服务器安全卫士（原生版）、云监控、Web应用防火墙（原生版）、证书管理服务、DDoS高防（边缘云版）、云堡垒机、Web应用防火墙（边缘云版）、云审计 企业应用 云通信短信 视频 视频直播、智能视图服务、云点播 专属云 专属云（计算独享型） 价格 账务 其他 数据库审计、云日志服务、轻量型云主机、云原生网关、API网关

本文主要介绍分布式消息服务RabbitMQ的消息持久化最佳实践。 使用场景 默认情况下，RabbitMQ生产者生产的消息存储在内存中，当节点宕机或重启时，如何确保消息不丢失呢？RabbitMQ通过持久化机制实现，持久化包括Exchange持久化、Queue持久化和Message持久化。持久化是将内存中的消息写入到磁盘中，以防异常情况导致内存中的消息丢失。但是磁盘的读写速度远不如内存，开启消息持久化后，RabbitMQ的性能会下降。与惰性队列不同，持久化消息会在磁盘和内存中各存储一份，只有在内存空间不够时，才会将内存中的消息删除，存储到磁盘中。 说明 非持久化队列、Exchange在重启之后会丢失。 非持久化消息在重启之后会丢失（经过持久化队列/Exchange的消息不会自动变为持久化消息）。 持久化消息在尚未完成持久化时，如果服务器重启，消息会丢失。 设置Exchange持久化 在RabbitMQ的Web界面创建Exchange时，设置“durable”为“true”，如图1所示，设置成功后如图2所示。 图1 设置Exchange持久化 图2 持久化的Exchange

Broker信息 展示该集群具体broker信息，包括该broker的详细指标，如下图： 其中关键指标为： commitLogMaxOffset为当前brokercommitLog最大的物理偏移。通过commitLogMaxOffset，RocketMQ可以追踪和管理消息的存储位置。当有新的消息写入时，RocketMQ会将消息追加到Commit Log文件的末尾，并更新commitLogMaxOffset的值。消费者在消费消息时，可以根据commitLogMaxOffset来确定从哪个偏移量开始消费消息。 consumeQueueDiskRatio为消费队列存储的文件占用的磁盘空间比例。通过配置consumeQueueDiskRatio，可以在保证消费队列的性能的同时，控制磁盘空间的占用。较小的consumeQueueDiskRatio值可以提高消费队列的读写性能，但会增加内存的使用。较大的consumeQueueDiskRatio值可以降低内存的使用，但可能会降低消费队列的读写性能。根据实际需求，可以根据系统的内存和磁盘资源情况来调整consumeQueueDiskRatio的值，以获得更好的性能和资源利用率。 putMessageDistributeTime为消息写入commitLog的耗时分布。通过配置putMessageDistributeTime，可以了解消息从发送到最终被消费的整体时间。这对于监控和优化消息传递的性能和延迟非常有用。注意，putMessageDistributeTime是一个估计值，实际的消息传递时间可能会受到网络状况、消费者处理能力等多种因素的影响。因此，在配置putMessageDistributeTime时，需要根据实际情况进行调整，并结合其他指标进行综合分析。

topicconfigs)。 − 图中⑤区域表示队列监控数据统计。 − 图中⑥区域表示队列分区信息，包括分区消息数(Latest Offset)，分区leader(Leader)，副本列表(Replicas)，同步副本列表(In Sync Replicas)。 队列详情页 查看消费组列表 导航栏中单击Consumers页签，即可查看当前集群中的消费组列表。 集群的消费组列表 查看消费组详情页 单击消费组名称可进入消费组详情页面，展示消费组消费的所有队列列表以及每个队列的可消费数(Total Lag)。 消费组详情页面 查看消费组队列详情页 单击队列名称，即可进入详情页面，查看消费组消费在队列中每个分区的消费状态。包括分区编号(Partition)，分区消息数(LogSize)，分区消费进度(Consumer Offset)，分区剩余可消费数(Lag)，最近消费该分区的消费者(Consumer Instance Owner)。 消费组队列详情页面

数据保护技术 RabbitMQ提供了多种数据保护技术，以确保数据在传输和存储过程中的机密性和完整性。 1. 跨AZ容灾：在不同的可用区部署多个RabbitMQ节点，确保节点之间的数据复制和同步。这样，当一个可用区发生故障时，其他可用区上的节点可以继续提供服务。 2. 副本冗余：RabbitMQ副本冗余是一种保证消息队列的高可用性和数据冗余的策略。通过在多个节点上创建副本，可以确保即使一个节点发生故障，其他节点上的副本仍然可以提供服务。 3. 数据持久化：RabbitMQ数据持久化是通过将队列、消息和交换器进行持久化，确保消息队列中的数据在节点重启或故障时不会丢失。 服务韧性 RabbitMQ服务的韧性是指其在面对各种故障和异常情况时能够保持可用性和可靠性的能力。以下是保障RabbitMQ服务韧性的关键方面： 1. AZ内实例容灾：在不同的可用区内部署多个RabbitMQ节点，使它们能够相互复制和同步消息。这样即使一个可用区发生故障，其他可用区的节点仍然可以提供服务。 2. 数据容灾：RabbitMQ数据容灾是通过持久化消息、队列和交换器、备份和复制以及高可用性集群等策略，保护数据免受损失和故障影响的措施。

本文主要介绍分布式消息服务RabbitMQ的入门实践。 当您购买了RabbitMQ实例后，可以根据自身的业务需求使用分布式消息服务RabbitMQ提供的一系列常用实践。 表1 常用最佳实践 实践 描述 RabbitMQ业务迁移 介绍将线下单机或集群实例迁移到天翼云RabbitMQ实例的方案。 RabbitMQ队列迁移 扩容节点、删除队列可能会导致队列在各个节点分布不均衡，本文介绍如何设置队列负载均衡。

专属队列 专属队列是指队列对应的资源为专属资源，空闲时不释放，即无论是否使用均保留资源的队列类型。专属队列可以保证提交作业时资源一定存在。 队列弹性扩缩容 DLI提供了队列弹性扩缩容的功能。用户在创建指定规格队列后，可根据需要进行弹性扩缩容。 根据业务情况，手动更改队列规格。具体操作请参考队列弹性扩缩容。 说明 新创建的队列需要运行作业后才可进行扩缩容。 队列弹性扩缩容定时任务 DLI提供了队列弹性扩缩容定时任务的功能。用户在创建队列后，可根据需要进行弹性扩缩容定时任务。 根据业务情况，设置队列自动扩缩容的时间，由系统定时触发队列扩缩容。具体操作请参考弹性扩缩容定时任务。 说明 新创建的队列需要运行作业后才可进行扩缩容。 队列自动扩缩容 Flink作业使用队列，DLI可根据作业大小自动触发扩缩容，用户无需进行操作。 说明 新创建的队列需要运行作业后才可进行扩缩容。 队列管理页面 队列管理主要包括如下功能： 队列权限管理 创建队列 删除队列 修改队列网段 队列弹性扩缩容 弹性扩缩容定时任务 测试地址连通性 创建消息通知主题 说明 DLI作业执行失败需要通过SMN发送通知消息，因此需要获得访问和使用SMN（消息通知服务）的SMN Administrator权限。 队列管理页面显示用户创建所有的队列和服务预置的default队列。队列列表默认按创建时间排列，创建时间最近的队列显示在最前端。 队列管理参数 参数 参数说明 名称 队列的名称。 类型 队列的类型。 SQL队列 通用队列 Spark队列（兼容老版本） 规格 队列大小，单位：CUs。 CUs是队列的计价单位。1CUs1Core 4GMem。不同规格的队列对应的计算能力不一样，规格越高计算能力越好。 实际CUs 当前队列实际大小值。 弹性扩缩容 定时扩缩容的目标CU值，或当前规格CU值的最大值和最小值。 用户名 队列所有者。 描述 创建队列时，对队列的描述。如果无描述，则显示“”。 操作 删除：删除所选队列。如果队列中有正在提交或者正在运行的作业，将不支持删除操作。 权限管理：查看队列对应的用户权限信息以及对其他用户授权。 更多 − 重启：强制重启队列。 说明 只有SQL队列有“重启”操作。 − 弹性扩缩容：可以根据需要选择“扩容”或“缩容”，目标值大小必须为16CU的整数倍。 − 弹性扩缩容定时任务：可以根据业务周期或使用情况，在不同的时间或周期内设置不同的队列大小，系统将定时自动进行“扩容”或“缩容”。目标值大小必须为16CU的整数倍。 − 修改网段：使用DLI增强型跨源时，DLI队列网段与数据源网段不能重合，可根据需要进行修改。 − 测试地址连通性：测试队列到指定地址是否可达，支持域名和ip，可指定端口。

配置项 操作说明 示例 触发器类型 选择Kafka触发器。 Kafka触发器 名称 填写自定义的触发器名称。 kafkatrigger Kafka实例 选择已创建的Kafka实例。 Topic 选择已创建的Kafka实例的Topic。 Group ID • 快速创建 ：推荐方案。自动创建以GROUPFCTrigger{triggername}{uuid}命名的Group ID。 • 使用已有 ：选择Kafka实例已有的GroupID，请您注意不要与已有的业务混用GroupID，否则会影响已有的消息收发。 消费任务并发数 消费者的并发数量，有效取值范围为[1,20]，建议不超过Topic的分区数。该值同时影响投递到函数的并发数。 消费位点 选择消息的消费位点，即触发器从kafka消息队列开始拉取消息的位置。 • 最早位点 ：从最早位点开始消费。 • 最新位点 ：从最新位点开始消费。 最新位点 调用方式 选择函数调用方式。 • 同步调用 ：指触发器消费topic消息后投递到函数是同步调用，会等待函数响应后继续下一个消息投递。但消费任务并发数大于1时，多个消费者有可能会并发消费消息并投递，并发的情况视topic队列本身积存的消息而定。 • 异步调用 ：指触发器消费topic消息后投递到函数是异步调用，不会等待函数响应，可以快速消费事件。 同步调用 触发器启用状态 创建触发器后是否立即启用。默认选择开启，即创建触发器后立即启用触发器。 启用 推送配置 • 批量推送条数 ：批量推送的最大值，积压值达到后立刻推送，取值范围为[1,10000]。 • 批量推送间隔 ：批量推送的最大时间间隔，达到后立刻推送，单位秒，取值[0,15]。默认0无需等待，数据直接推送。 • 推送格式 ：函数收到的事件格式，详情请查阅触发器事件消息格式。 重试策略 消息推送函数失败后重试的策略，共两种： • 指数退避 ：指数退避重试，重试5次，重试周期为2,4,8,16,32（秒）。 • 线性退避 ：线性退避重试，重试5次，重试周期为1,2,3,4,5（秒）。 容错策略 当重试次数耗尽后仍然失败时的处理方式： • 允许容错 ：当异常发生并超过重试策略配置时直接丢弃。 • 禁止容错 ：当异常发生并超过重试策略配置时继续阻塞执行。 死信队列 当容错策略为：允许容错时，可以额外开启死信队列。当开启死信队列时且异常发生并超过重试策略配置时，消息会被投递到指定的消息队列里，当前只支持投递到kafka和rocketmq

本章节介绍了如何查询分布式消息服务RocketMQ实例的Topic。 操作场景 Topic创建成功后，查询Topic相关的配置和状态信息。 操作步骤 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，单击“Topic管理”，进入“Topic管理”页面。 4. 单击需要查询的Topic名称，进入Topic详情页面。 在详情页上方可以查看Topic名称、关联代理、读队列个数、写队列个数和权限。 在详情页下方可以查看Topic在每个代理上的队列状态，包括队列ID、最小偏移量、最大偏移量和消息更新时间。还可以查看消费组消费此Topic的情况，包括消费组名称、最大重试次数和广播消费。 图 Topic详情

根据ID查询 根据消息ID查询唯一消息，选择消费组后，可以查询到该消息是否被该消费组消费过。 点击“查看”，可以查询该消息的包体内容。 消费状态标志含义： （1）Toconsume：未消费。 （2）Consumed：已签收。 （3）Consuming：已拉取，未签收。 根据offset查询 根据指定队列指定偏移量查询唯一消息，选择订阅组后，可以查询到该消息是否被该订阅组消费过。 点击“查看”，可以查询该消息的包体内容。 基于Topic查询 死信队列查询

Java客户端必须使用Push Consumer 使用Pull可以实现的所有场景，均可使用Push实现，并且更简单。 Push其实是长轮询的Pull（依然是由客户端发起），在客户端通过配置参数是可以实现流控的，并不会出现服务端的流量压垮客户端的情况。 Push封装了拉取消息，分发给消费线程的线程模型，非流控的情况下，由后台线程主动拉取消息，并缓存在本地，消费线程池有空闲线程时，分发给消费线程，在有堆积量的情况下，可以保证消费线程一直工作，性能更高（备注：Pull只提供了拉取消息的功能，并且何时去拉取，拉取时机，这些都需要应用去控制；分发给消费线程的逻辑需要应用封装，除了增加应用工作量外，还可能有不稳定、性能问题等）。 Push经过多个大型项目的长时间的使用，更成熟稳定。 Push会自动订阅重试队列，不需要再次调用拉取重试队列的API来取得重试队列的消息（备注：Pull需要另外调用API拉取重试队列的消息）。 Pull是一种遗留的消费模式（兼容早期的API），新开发的应用，或者未上线的应用，都要求使用Push消费模式。