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

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

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

本章节介绍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不可用的目标对象。

本章节介绍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不可用的目标对象。

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

介绍分布式消息服务MQTT实例列表内容。 操作说明 1、 进入控制台查看已购买的实例列表，若列表为空，可点击右上角【订购实例】进入购买页面，创建实例详情见具体操作步骤。 2、 支持按照实例名称查询，输入查询内容，点击【查询】按钮即可展示需要的实例数据。 3、 查看实例基本信息，包括实例ID、规格、计费模式、创建时间、到期时间、状态。其中状态说明见下文。 运行状态 1、 登录管理控制台。 2、 进入MQTT管理控制台。 3、 当前页面会列出所购买的MQTT实例，并查看状态，状态说明如下 状态 说明 运行中 MQTT实例正常运行状态。在这个状态的实例可以运行您的业务。 已关闭 MQTT实例处于故障的状态。 变更中 MQTT实例正在进行规格变更操作。 变更失败 MQTT实例处于规格变更失败状态。 暂停 MQTT实例处于已冻结状态，用户可以在“更多”中续费开启冻结的MQTT实例。 注销 MQTT实例已经过期并关闭，需要重新购买实例。 回收站 1、实例退订后，在回收站可查看已退订实例。 2、回收站实例销毁。 点击回收站实例列表“销毁”，进入实例销毁页面，提交订单。

RocketMQ无法避免消息重复，原因主要有以下几点： 签收的偏移量是定时（每5秒/次）同步到服务端的。 为保证消息不丢失，SDK每次提交的总是队列未签收的最小偏移量（比如无序消费，offset为1、2、3、4、5 的消息，1、3、4、5消费并已签收，2未签收，签收时最后提交的偏移量将会是2，如果此时客户端重启，会从2这个位置开始消费）。 有网络交互就不能确保每一次的交互数据都是送达的，为保证数据不丢失就要进行重试，有重试就存在重复的可能。 如果业务对消费重复非常敏感，务必要注意，建议可以采用以下两种方式处理： 业务层面可以根据msgId做去重处理，如果key字段为业务唯一字段，也可采用key去重。 业务逻辑实现消费幂等，即多次处理同一消息，对业务的影响是幂等的。

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

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

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

本章节介绍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 压力。通常保持默认值即可。

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

本章节介绍了天翼云分布式消息服务协议。 天翼云分布式消息服务协议

本文为您介绍分布式消息服务MQTT的操作指南连接查询。 分布式消息服务MQTT提供按 Client ID 查询客户端的连接信息和订阅关系，进行剔除（下线）和主题退订操作。 操作步骤 1、 天翼云官网点击控制中心，选择产品分布式消息服务MQTT。 2、 登录分布式消息服务MQTT控制台，点击右上角地域选择对应资源池。 3、 进入实例列表，点击【管理】按钮进入管理菜单。 4、 进入连接查询菜单页面，输入Client ID，点击查询按钮。 5、 展示客户端的连接信息和订阅关系，支持剔除（下线）和主题退订操作。 查询结果 ClientID：一个在客户端连接到MQTT代理（MQTT broker）时分配的唯一标识符，用于标识特定的MQTT客户端。 客户端IP：客户端的IP地址。 连接协议：MQTT协议。 连接时间：指客户端与MQTT代理（MQTT broker）之间建立和维护连接的时间。 cleanSession： MQTT 协议中一个重要的连接选项，用于定义客户端与 MQTT 代理（broker）之间的会话保持行为。这个选项决定了当客户端重新连接到代理时，是否应该恢复之前的会话状态。 认证用户：认证授权模块创建的用户。 当前订阅数：指当前连接到MQTT代理的客户端数量，这些客户端已经发送了订阅请求并成功订阅了一个或多个主题。每个订阅都占用一个资源，因此订阅数可以反映出代理的负载和资源使用情况。 发送消息数：在MQTT代理（broker）上发布或传输的消息数量。这个指标反映了在特定时间段内从一个或多个发布者发送到MQTT代理并通过代理传递到一个或多个订阅者的消息数量。发送消息数通常用于衡量MQTT通信的活动水平和负载。 接收消息数：MQTT代理（broker）上接收并传递给订阅者（Subscriber）的消息数量。这个指标表示了在特定时间段内，MQTT代理成功接收到的消息数量，然后将这些消息传递给相应的订阅者。 是否在线：客户端是否在线，支持剔除下线操作。 连接断开时间：客户端与服务器之间的连接保持（Keep Alive）时间。在MQTT中，客户端和服务器之间的连接是通过心跳机制来保持的，以确保连接的活跃性和可靠性。

本文主要介绍查询消息。 操作场景 您可以查看指定Topic不同分区的偏移量、消息大小、创建时间以及消息正文。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例名称，进入实例详情页面。 步骤 5 选择“消息查询”页签，在消息页签页面，设置查询的Topic名称、分区以及查询方式。 如果未设置具体分区，查询结果显示Topic所有分区的消息。 查询方式支持以下两种方式： 按创建时间查询：即按生产该消息的时间。 按偏移量查询：即记录消息的位置。 说明 当Topic中的数据量比较大时，单副本Topic查询消息可能会报“内部服务错误”，建议根据数据量适当减小查询时间范围。 步骤 6 单击“搜索”，查询消息。 查询结果如下： 图 查询Topic消息 消息的参数说明如下： Topic名称：消息所在的Topic名称。 分区：消息所在的分区。 偏移量：消息在分区中的位置。 消息大小：消息存入磁盘的大小，单位为B。 创建时间：消息的创建时间。创建时间由生产客户端在生产消息时通过CreateTime指定的，如果生产消息时没有设置此参数，创建时间会默认为1970。 步骤 7 单击“查看消息正文”，弹出“查看消息正文”对话框，查看消息的内容，包括Topic名称、分区、偏移量、创建时间和消息正文。 步骤 8 （可选）如果需要恢复默认设置，单击“重置”。

介绍APM监控详情里消息监控中的RabbitMqProducer相关指标的名称、含义等信息。 RabbitMqProducer监控指标说明表 指标类别 指标 指标说明 数据类型 异常 exceptionstacktrace 异常产生的堆栈信息 String 异常 exceptioncount 错误数 Int Exchange监控 messagedestination Exchange名 Enum Exchange监控 messageerrorcount 推送消息的错误次数 Int Exchange监控 messagecount 推送消息数 Int Exchange监控 messagetotalpayload 推送字节数 Int Exchange监控 Max(messagetotalpayload) 单次推送最大字节数 Int Exchange监控 messagemaxcost 推送消息的最大响应时间 Int Exchange监控 sum(messagecost) 推送消息的总响应时间 Int Total监控 messagetotalpayload 推送字节数 Int Total监控 messagecount 推送消息数 Int Total监控 messageerrorcount 推送消息的错误次数 Int Total监控 messagemaxcost 推送消息的最大响应时间 Int Total监控 sum(messagecost) 推送消息的总响应时间 Int Total监控 ms0To10Count 010ms次数 Int Total监控 ms10To100Count 10100ms次数 Int Total监控 ms100To500Count 100500ms次数 Int Total监控 ms500To1000Count 5001000ms次数 Int Total监控 ms1000To10000Count 110s次数 Int Total监控 msMorethan10000Count 10s以上次数 Int

本文主要介绍日志处理工具Logstash接入Kafka。 最佳实践概述 场景描述 Kafka与Logstash互联，可实现将Kafka作为inputs把消息传递给Logstash进行处理、或者是作为outputs将Kafka用作消息目的地。 技术架构图 暂无。 方案优势 可以异步处理数据：防止突发流量。 解耦：当发生异常的时候不会影响上游工作。 前提条件 需已购买Kafka实例、创建Topic，并且已成功消费消息。 资源规划 本实践方案内容仅涉及Kafka专享版实例和Logstash。 分布式消息服务 Figure 分布式消息服务 资源类型 配置项 配置明细 说明 :::: 企业中间件 DMS Kafka专享实例 需已购买kafka专享实例，创建好Topic，并成功消费消息。 Logstash下载与安装 下载并安装Logstash，验证运行成功。 方案正文 配置Kafka 1、登录分布式消息服务Kafka版控制台，按需要单击“实例名称”，进入实例基本信息页面。 图 Kafka控制台实例列表 2、在“连接地址”模块，获取Kafka连接IP地址。 图 连接IP地址与端口 3、创建Logstashtest的Topic。 图 创建Topic

本文主要介绍云日志服务的投递概述。 天翼云云日志服务可以将日志单元中的数据投递到天翼云对象存储ZOS与分布式消息服务kafka，以满足归档或者计算的需求。 转储至对象存储ZOS 您将云主机和云服务的日志数据上报至云日志服务后，可对日志存储时长进行设置，默认存储时间为365天，超出存储时间后，日志数据将会被自动清理。若您需要长期存储日志数据进行日志持久化，可使用日志转储功能，将日志数据转储至对象存储中进行长期保存。具体操作请查看转储至对象存储ZOS。 说明 日志转储的实时性较弱，通常为5分钟~30分钟。数据延迟依赖于存储系统。 关于对象存储更多内容，请参考对象存储产品文档。 转储至分布式消息服务kafka 云日志服务采集到数据后，支持将数据转储至对天翼云分布式消息服务kafka，用于实时流计算等场景。具体操作请查看转储至分布式消息服务kafka。

测试场景 本章节主要测试RocketMQ不同产品规格在发送1KB大小的消息，实例的网络入流量、网络出流量、消息生产速率、消息消费速率、CPU核均负载和内存使用率。 测试环境 购买实例 名称 规格 存储空间 rocketmq01 c7.xlarge.2 4C8G 超高I/O 500GB rocketmq02 c7.2xlarge.2 8C16G 超高I/O 500GB rocketmq03 c7.4xlarge.2 16C32G 超高I/O 500GB 控制台创建主题 名称 权限 关联代理 队列个数 topic01 发布+订阅 broker1 8 控制台创建消费组 名称 关联代理 最大重试次数 是否允许以广播模式消费 group01 broker1 16 否 测试工具准备 购买1台ECS服务器（区域、可用区、虚拟私有云、子网、安全组与RocketMQ实例保持一致，Linux系统），安装jdk，并配置环境变量 tar zxvf jdk8u131linuxx64.tar.gz vi /etc/profile 追加以下内容 export JAVAHOME/root/jdk1.8.0131 export PATH$JAVAHOME/bin:$PATH 生效配置 source /etc/profile 下载测试工具 wget 解压测试工具 unzip rocketmqall5.1.4binrelease.zip 测试命令

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

本章节介绍了如何使用实例诊断功能快速定位问题。 操作场景 使用分布式消息服务RocketMQ消费消息时，如果遇到消息消费缓慢或者消费不到消息等消费异常的场景，借助实例诊断功能可以帮助您快速定位问题。 前提条件 已创建消费组，且消费组中已存在消费者。 已创建Topic。 实例诊断 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，选择“实例诊断”，进入实例诊断页面。 4. 选择待诊断的消费组名称，单击“开始诊断”。 诊断完成后，在“诊断记录”区域，新增一条诊断记录。 5. 单击“查看报告”，进入“实例诊断”页面，查看诊断结果。 图1 诊断结果 说明： 消费组中没有消费者时，会导致诊断失败。 消费组不在线时，可以诊断成功，但是会显示1项异常。 删除诊断记录 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，选择“实例诊断”，进入实例诊断页面。 4. 选择以下任意一种方法删除诊断记录。 在待删除的诊断记录所在行，单击“删除”。 如需批量删除诊断记录，勾选待删除的诊断记录，单击“批量删除”。 5. 弹出确认删除对话框，单击“是”，完成诊断记录的删除。

分布式消息服务Kafka提供全托管、免运维的云实例间迁移服务，用于同云实例与分布式消息服务Kafka实例之间的数据同步。本文介绍云实例间迁移的源实例类型、使用限制以及使用流程。 背景信息 云实例间迁移可以把天翼云分布式消息服务Kafka集群的元数据（Topic、Group和SASL User信息）、消息数据和点位信息同步到目标集群，迁移完成后目标集群的元数据与原集群的元数据保持一致，并且支持持续更新和完成后自动停止任务。 使用限制 1.此功能不收费，会占用当前实例的服务器资源，请结合业务流量和服务器资源占用情况，在合理的情况下进行迁移 2.单机版不支持迁移 计费说明 分布式消息服务Kafka的云实例间迁移组件处于公测阶段，不会在分布式消息服务Kafka侧产生费用。同时，天翼云不承诺迁移的SLA。 创建迁移任务 1.登录分布式消息服务Kafka控制台。 2.在左侧导航栏，单击集群迁移 ，然后单击云实例间迁移。 3.在云实例间迁移页签，单击创建任务。 4.在创建 云实例间迁移任务面板，填写任务名称，选择目标实例，然后单击下一步。 5.填写接入点，选择源实例，任务数等，具体请看参数说明。 参数说明： 参数 说明 示例 源实例 下拉框选择源实例 192.168.XX.XX:9092 任务数 选择同步数据的任务数。取值说明如下：1、6、12 1 迁移完成后自动停止任务 迁移任务会自动检测是否完成迁移 是：在检测到消息数据完成同步后停止任务 否：不停止任务，您可以手动停止任务 否 Topic 要迁移的topic信息，选填，多个用半角逗号分开，不填则迁移所有topic topic1,topic2 Groups 要迁移的group信息，选填，多个用半角逗号分开，不填则迁移所有group group1,group2 副本数是否和源集群保持一致 迁移到目标集群的topic的副本数是否和源集群保持一致，默认否 是：副本数是否和源集群保持一致，如原集群topic副本数为1，迁移后的topic副本数也为1 否：迁移后topic的副本数为3 否 6.填写完参数后，单击创建 ，完成任务的创建，在任务列表能看到一条任务，状态为等待迁移，后台调度到此任务时，会将任务状态改为迁移中。 7.其他操作 查看任务详情：单击详情。在任务详情页面，查看云实例间迁移任务的基础信息、源服务、目标服务和运行环境配置信息。 查看同步进度：单击同步进度，选择查看的Topic，可以看到对应Topic的分区id，最早点位，最新点位和当前点位信息。 启停任务：单击停止，然后在提示对话框，单击确认可停止任务。

本文主要介绍产品优势。 分布式消息服务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的分区数、副本数。

本文介绍节点重启后消费者如何重连，以Java中使用的RabbitMQ客户端amqpclient为例。 amqpclient自带重连机制，但是自带的重连机制只会重试一次，一次连不上后就不会再执行了，这时如果消费者没有做额外的重试机制，那么这个消费者就彻底丧失的消费能力。 amqpclient在节点断连后，根据与通道建立的节点不同，产生不同的错误。 如果通道连接的是队列所在的节点，消费者就会收到一个shutdown信号，这时amqpclient的重连机制就会生效，尝试重新连接服务端。如果连上了，这个通道就会继续连接消费。如果连不上，就会执行channel.close方法，关闭这个通道。 如果通道连接的不是队列所在的节点，消费者不会触发关闭动作，而是由服务端发送的一个取消动作，这个动作对amqpclient来说并不是异常行为，所以日志上不会有明显的报错，但是连接最终还是会关闭。 amqpclient出现上面两种错误时，会分别回调handleShutdownSignal以及handleCancel方法，您可以通过重写这两种方法，在回调时执行重写的重连逻辑，就能在通道关闭后重新创建消费者的新通道继续消费。 以下提供一个简单的代码示例，能够解决上面的两种错误，实现消费者的持续消费。 import com.rabbitmq.client.; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.concurrent.TimeoutException; public class MyRabbitConsumer { public static void main(String... args) throws IOException, TimeoutException { ConnectionFactory factory new ConnectionFactory(); factory.setHost("192.168.x.x"); factory.setPort(5672); factory.setUsername("name"); factory.setPassword("password"); Connection connection factory.newConnection(); createNewConnection(connection); } public static void createNewConnection(Connection connection) { try { Channel channel connection.createChannel(); channel.basicQos(64); channel.basicConsume("queue1", false, new CustomConsumer(channel, connection)); } catch (Exception e) { createNewConnection(connection); } } static class CustomConsumer implements Consumer { private final Channel channel; private final Connection connection; public CustomConsumer(Channel channel, Connection connection) { channel channel; connection connection; } @Override public void handleConsumeOk(String consumerTag) {} @Override public void handleCancelOk(String consumerTag) {} @Override public void handleCancel(String consumerTag) throws IOException { createNewConnection(connection); } @Override public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) { createNewConnection(connection); } @Override public void handleRecoverOk(String consumerTag) {} @Override public void handleDelivery(String consumerTag, Envelope env, AMQP.BasicProperties prop, byte[] body) throws IOException { String message new String(body, StandardCharsets.UTF8); System.out.println("收到消息: " + message); channel.basicAck(env.getDeliveryTag(), false); } } }

分布式消息服务MQTT实例在被删除后，会被临时存入回收站中，此时实例中的数据尚未被彻底删除，在保留天数内支持从回收站中恢复此实例。超过保留天数的实例会被彻底删除，无法恢复。 约束与限制 回收站中的实例不会收取实例的费用，但是会收取存储空间的费用。 包年/包月的实例退订后会存入回收站中，此时不会收取实例的费用，但是收取存储空间的费用。 回收站中的按需实例，不支持实例恢复。 回收站管理 1. 登录分布式消息服务MQTT管理控制台。 2. 在左侧导航栏选择“回收站”，进入“回收站”页面。 恢复MQTT实例 1. 登录分布式消息服务MQTT管理控制台。 2. 在左侧导航栏选择“回收站”，进入“回收站”页面。 3. 在待恢复MQTT实例所在行，单击“恢复”。 删除回收站中的实例 1. 登录分布式消息服务MQTT管理控制台。 2. 在左侧导航栏选择“回收站”，进入“回收站”页面。 3. 在待删除MQTT实例所在行，单击“销毁”。 注意 回收站中的MQTT实例删除后，实例中原有的数据将被删除，且没有备份，请谨慎操作。

本文介绍如何在分布式消息服务Kafka控制台创建元数据导入任务，将元数据迁移至分布式消息服务Kafka现有实例。 前提条件 下载JDK8 购买并部署分布式消息服务Kafka实例 背景信息 Kafka集群元数据是指Kafka集群的Topic和Group配置信息。Kafka集群元数据存储于ZooKeeper上，Kafka集群各个节点从ZooKeeper中获取最新的元数据。因此，集群的各个节点的元数据被导出时都是最新且相同的。Kafka集群元数据可以被导出成一份JSON文件，然后被导入到另一个Kafka集群，实现自建Kafka集群元数据迁移上云。 元数据迁移是指将自建Kafka集群的元数据迁移到分布式消息服务Kafka实例。您可以将自建Kafka集群元数据导出，然后导入云分布式消息服务Kafka实例，分布式消息服务Kafka会解析成功导入的元数据，并支持在创建任务时可视化配置Topic和Group，配置完成后在目标实例中创建对应的Topic和Group，实现自建Kafka集群元数据迁移上云。 步骤一：导出元数据 使用元数据导出工具导出自建Kafka集群的元数据文件。 1.下载元数据导出工具，登陆分布式消息服务Kafka控制台>集群迁移>元数据迁移>创建任务>下载工具 下载对应jar包 2.将下载好的工具上传至能连接到Kafka集群的机器上 3.运行以下命令，导出元数据。运行命令后，您将在kafkamigration.jar 同级目录下看到 名为 metadata.json 的元数据文件。 java cp kafkamigration.jar com.ctg.migration.kafka.MigrationMetadata sourceKafkaConnect (resource topic,group username password mechanism configKey ) 参数 说明 示例 sourceKafkaConnect 自建Kafka集群的IP地址和端口号。 192.168.XX.XX:9092 resource 要导出的资源，可以是topic、 group或topic,group 默认topic,group topic,group username 连接kafka用户名，如果sourceKafkaConnect为免密接入点则无需填写 admin user password 连接kafka密码，如果sourceKafkaConnect为免密接入点则无需填写 real password mechanism 对应的协议，支持PLAIN、SCRAMSHA256、SCRAMSHA512，如果sourceKafkaConnect为免密接入点则无需填写 PLAIN configKey 源集群topic配置，支持多个key，多个用半角逗号分开，all：表示所有配置(不推荐)，具体支持哪些配置可以通过(bin/kafkaconfigs.sh help)命令查看 默认值为：retention.ms retention.ms,max.message.bytes

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

应用架构 支持的功能模块 应用双活 管控通道 注册配置中心RCC 应用双活 接入层 云原生网关CGW 应用双活 服务层 SpringCloud 应用双活 服务层 Dubbo 应用双活 数据层 关系型数据库MySQL版 应用双活 数据层 关系型数据库PostgreSQL版 应用双活 消息层 分布式消息服务RocketMQ 数据双活 管控通道 注册配置中心RCC 数据双活 接入层 云原生网关CGW 数据双活 服务层 SpringCloud 数据双活 服务层 Dubbo 数据双活 数据层 关系型数据库MySQL版 数据双活 数据层 关系型数据库PostgreSQL版 数据双活 消息层 分布式消息服务RocketMQ

本文为您介绍分布式消息服务MQTT的计费类常见问题。 如何为购买的消息队列MQTT实例续费？ 为防止资源到期或者浪费，已经购买包月实例的用户，可执行续费操作，延长MQTT实例的有效期，也可以设置到期自动续费的相关操作。 开通的MQTT实例资源何时生效？ 天翼云分布式消息服务MQTT资源在客户支付成功之后，系统经过初始化完成后即可生效使用。

本文为您介绍分布式消息服务MQTT的操作指南认证授权。 创建用户名和密码 终端设备连接MQTT队列需要先创建用户密码。 1、 天翼云官网点击控制中心，选择产品分布式消息服务MQTT。 2、 登录分布式消息服务MQTT控制台，点击右上角地域选择对应资源池。 3、 进入实例列表，点击【管理】按钮进入管理菜单。 4、 进入认证授权菜单，点击【新增】按钮，在弹出框输入认证用户名、用户密码、确认密码等信息。 主题授权 对用户名进行主题授权，客户端方可正常收发。 1、 选择需要授权的用户，点击【授权】按钮。 2、 在弹出框填写已创建的主题名称，主题权限包含3种：pub、sub、pubsub，支持通配符，代表所有主题。

介绍APM监控详情里消息监控中的RabbitMqConsumer相关指标的名称、含义等信息。 RabbitMqConsumer监控指标说明表 指标类别 指标 指标说明 数据类型 异常 exceptionstacktrace 异常产生的堆栈信息 String 异常 exceptiontype 异常类型 String 异常 exceptioncount 错误数 Int 推模式消费维度监控 messageerrorcount 消费消息的错误次数 Int 推模式消费维度监控 ms0To10Count 010ms次数 Int 推模式消费维度监控 ms10To100Count 10100ms次数 Int 推模式消费维度监控 ms100To500Count 100500ms次数 Int 推模式消费维度监控 ms500To1000Count 5001000ms次数 Int 推模式消费维度监控 ms1000To10000Count 110s次数 Int 推模式消费维度监控 msMorethan10000Count 10s以上次数 Int 推模式消费维度监控 messagecount 调用次数 Int Connection监控 notacknowledgedpublished 该连接中未确认的消息数 Int Total监控 messageerrorcount 消费消息的错误次数 Int Total监控 messagecount 消费消息数 Int Total监控 messagetotalpayload 消费字节数 Int Total监控 Max(messagetotalpayload) 单次消费最大字节数 Int Total监控 acknowledged Ack消息数 Int Total监控 rejected Reject消息数 Int Total监控 ms0To10Count 010ms次数 Int Total监控 ms10To100Count 10100ms次数 Int Total监控 ms100To500Count 100500ms次数 Int Total监控 ms500To1000Count 5001000ms次数 Int Total监控 ms1000To10000Count 110s次数 Int Total监控 msMorethan10000Count 10s以上次数 Int Total监控 notacknowledgedpublished 未确认的消息数 Int Total监控 messagemaxcost 最大响应时间 Int Total监控 sum(messagecost) 消费消息的总响应时间 Int