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

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

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

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

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

本文主要介绍如何实现RabbitMQ的高性能最佳实践。 本章节基于吞吐量和可靠性两个指标，指导您通过设置队列长度、集群负载均衡、优先队列数量等参数，实现RabbitMQ的高性能。 使用较小的队列长度 队列中存在大量消息时，会给内存使用带来沉重的负担，为了释放内存，RabbitMQ会将消息刷新到磁盘。这个过程通常需要时间，由于需要重建索引，重启包含大量消息的集群非常耗时。当刷盘的消息过多时，会阻塞队列处理消息，从而降低队列速度，对RabbitMQ节点的性能产生负面影响。 要获得最佳性能，应尽可能缩短队列。建议始终保持队列消息堆积的数量在0左右。 对于经常受到消息峰值影响的应用程序，和对吞吐量要求较高的应用程序，建议在队列上设置 最大长度 。这样可以通过丢弃队列头部的消息来保持队列长度，队列长度永远不会大于最大长度设置。 最大长度可以通过Policy设置，也可以通过在队列声明时使用对应参数设置。 在Policy中设置 在队列声明时使用对应参数设置 //创建队列 HashMap map new HashMap<>(); //设置队列最大长度 map.put("xmaxlength",10 ); //设置队列溢出方式保留前10 map.put("xoverflow","rejectpublish" ); channel.queueDeclare(queueName,false,false,false,map); 当队列长度超过设置的最大长度时，RabbitMQ的默认做法是将队列头部的信息（队列中最老的消息）丢弃或变成死信。可以通过设置不同的overflow 值来改变这种方式，如果overflow 值设置为 drophead ，表示从队列前面丢弃或deadletter消息，保存后n条消息。如果overflow 值设置为 rejectpublish ，表示最近发布的消息将被丢弃，即保存前n条消息。 说明 如果同时使用以上两种方式设置队列的最大长度，两者中较小的值将被使用。 超过队列最大长度的消息会被丢弃，请谨慎使用。

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

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

本章节介绍了如何通过云审计服务来查看云审计日志。 查看DMS for RocketMQ云审计日志，请参考查看追踪事件。

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

说明 分布式消息RabbitMQ旧资费产品根据实例规格分为基础版和高级版，存储空间按照不同类型收费。 旧资费产品目前在 芜湖2、南京3、上海7、重庆2、北京5 资源池开放订购。 包周期 RabbitMQ当前支持包年/包月的计费方式。且按天翼云的包年订购优惠策略享受相关的一次性包年订购优惠。 RabbitMQ队列实例价格如下： 实例类型 实例规格 节点数量 标准资费(元/月) 基础版（300G） 4C8G 3 3248 高级版（300G） 8C16G 3 6321 其中，RabbitMQ队列额外存储空间价格如下： 存储类型 标准资费（元/100GB/月） 普通IO 30 按需计费 RabbitMQ当前支持按需的计费方式，计费颗粒度为按小时。 RabbitMQ队列实例价格如下： 实例类型 实例规格 节点数量 标准资费(元/小时) 基础版（300G） 4C8G 3 6.73 高级版（300G） 8C16G 3 13.13 其中，RabbitMQ队列额外存储空间价格如下： 存储类型 标准资费（元/100GB/小时） 普通IO 0.05

本文为您介绍分布式消息服务MQTT的计费类常见问题。 如何为购买的消息队列MQTT实例续费？ 为防止资源到期或者浪费，已经购买包月实例的用户，可执行续费操作，延长MQTT实例的有效期，也可以设置到期自动续费的相关操作。 开通的MQTT实例资源何时生效？ 天翼云分布式消息服务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 （可选）如果需要恢复默认设置，单击“重置”。

分类 云产品 支持审计的关键操作列表 容器与中间件 云容器引擎 容器与中间件 容器镜像服务 容器与中间件 分布式消息服务Kafka 容器与中间件 分布式消息服务RabbitMQ 容器与中间件 弹性容器实例 容器与中间件 云日志服务

分布式消息服务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

本文主要介绍日志处理工具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

分布式消息服务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触发器 名称 填写自定义的触发器名称。 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

介绍分布式消息服务Kafka认证与访问控制方式 分布式消息服务Kafka支持CTIAM身份认证和多种访问控制，多维度保障您数据的安全。 身份认证 CTIAM 统一身份认证（Identity and Access Management， 简称：CTIAM）是提供用户进行权限管理的基础服务，可以帮助您安全的控制天翼云服务和资源的访问及操作权限，包括：用户身份认证、权限分配、访问控制等功能。具体介绍请参考统一身份认证产品介绍。 您可以创建IAM用户，并为其设置关联分布式消息服务Kafka实例权限，该用户就可以通过用户名和密码访问授权的实例资源。具体请参见统一身份认证快速入门创建IAM用户。 访问控制 权限控制 购买分布式消息服务Kafka实例之后，您可以使用CTIAM为企业中的员工设置不同的访问权限，以达到不同员工之间的权限隔离，通过CTIAM进行精细的权限管理。 VPC和子网 虚拟私有云（Virtual Private Cloud，VPC）为分布式消息服务Kafka构建隔离、私密的虚拟网络环境，提升数据库的安全性，并简化用户的网络部署。您可以完全掌控自己的专有网络，VPC丰富的功能帮助您灵活管理云上网络，包括创建子网、设置安全组和网络ACL、管理路由表、申请弹性公网IP和带宽等。通过子网与其他网络隔离，独享网络资源，提高网络安全性。具体内容请参见虚拟私有云用户指南创建虚拟私有云和子网 。 安全组 安全组是一个逻辑上的分组，可以为同一个虚拟私有云内具有相同安全保护需求并相互信任的Kafka实例提供相同的访问策略。您可以通过为数据库实例设置安全组，开通需访问Kafka实例的IP地址和端口，来保证保障其运行环境的安全性和稳定性。具体请参见配置安全组

本文主要介绍流数据实时处理框架Flink接入Kafka 。 最佳实践概述 场景描述 如果您想要在流数据实时处理场景下，使用Kafka作为流数据的源或者使用Kafka作为流数据的目的，需要将Kafka与Flink对接，可以参考本实践方案。 技术架构图 暂无。 方案优势 用户可按业务需要，参考设置Kafka与Flink框架对接，实现流数据的消息消费。 前提条件 需已购买Kafka实例、创建Topic，并且已成功消费消息。 资源规划 本实践方案内容仅涉及Kafka专享版实例和Flink的安装配置。 分布式消息服务 Figure 分布式消息服务 资源类型 配置项 配置明细 说明 :::: 企业中间件 DMS Kafka专享实例 需已购买kafka专享实例，创建好Topic，并成功消费消息。 Flink流数据处理框架 Figure Flink框架 资源类型 配置项 配置明细 说明 :::: 应用软件 Flink框架 Flink 1.14 使用开源Apache Flink 方案正文 获取Kafka访问地址 1、登录分布式消息服务Kafka版控制台，按需要单击“实例名称”，进入实例基本信息页面。 图 Kafka控制台实例列表 2、在“连接地址”模块，获取Kafka连接IP地址。 图 连接IP地址与端口

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

从事件源接收的事件 事件模式 模式过滤后的事件 plaintext { "id": "f36b49b668914b59b11cbf689xxxx712", "source": "ctyun:kafka", "type": "ctyun:kafka:Topic:Message", "specversion": "1.0", "datacontentype": "application/json; charsetutf8", "subject": "ctyun:kafka:75dcxx1eb5:topic:source1", "time": "20241203T09:48:00.696Z", "data": { "key": "test2", "value": "test1" } } plaintext { "id": "f36b49b668914b59b11cbf689xxxx713", "source": "ctyun:rocketmq", "type": "ctyun:rocketmq:Topic:Message", "specversion": "1.0", "datacontentype": "application/json; charsetutf8", "subject": "ctyun:rocketmq:75dcxx1eb5:topic:source1", "time": "20241203T09:48:00.696Z", "data": { "key": "test2", "value": "test1" } } plaintext { "type": [ { "contains" : "kafka" }, { "contains" : "rocketmq" } ] } plaintext { "id": "f36b49b668914b59b11cbf689xxxx712", "source": "ctyun:kafka", "type": "ctyun:kafka:Topic:Message", "specversion": "1.0", "datacontentype": "application/json; charsetutf8", "subject": "ctyun:kafka:75dcxx1eb5:topic:source1", "time": "20241203T09:48:00.696Z", "data": { "key": "test2", "value": "test1" } } plaintext { "id": "f36b49b668914b59b11cbf689xxxx713", "source": "ctyun:rocketmq", "type": "ctyun:rocketmq:Topic:Message", "specversion": "1.0", "datacontentype": "application/json; charsetutf8", "subject": "ctyun:rocketmq:75dcxx1eb5:topic:source1", "time": "20241203T09:48:00.696Z", "data": { "key": "test2", "value": "test1" } }

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

案例背景 某电商平台订单处理系统采用分布式微服务架构，其概要结构图如下，核心流程通过分布式消息服务Kafka集群实现异步解耦： 订单创建：用户在前端提交订单，请求经过负载均衡器转发至订单服务集群。订单服务处理业务逻辑，生成包含订单详情的消息数据，并将该消息可靠地发送至Kafka集群的特定Topic。 下游处理：库存服务（负责扣减库存）、支付服务（负责发起支付请求）和物流服务（负责生成物流配送单）是主要的下游消费者。它们各自订阅Kafka集群中的相应Topic，并行地拉取并处理消息。 Kafka集群作为异步通信的核心，其稳定性和性能对整个订单系统的可用性至关重要。在高并发的场景下，Kafka Broker可能会面临各种异常情况，其中CPU负载过高是常见的一种，可能由突发流量洪峰、JVM垃圾回收或底层基础设施瓶颈等原因引起。Kafka Broker的CPU高负载会直接影响消息的处理速度，导致生产和消费的延迟，甚至引发整个系统的不稳定。 故障演练服务提供了分布式消息服务KafkaBroker节点CPU高负载演练动作，通过模拟这种真实的故障场景，我们可以量化分析Kafka集群在Broker节点CPU受限时的表现，验证订单处理系统及其下游服务对Kafka异常的容忍度和应对能力，并提升团队在真实故障发生时的异常响应与恢复能力。

本文介绍了分布式消息服务Kafka版提供的SDK语言版本。 SDK列表 下表提供了分布式消息服务Kafka版支持的SDK列表。 编程语言 参考文档 :: Java 开发指南Java Python 开发指南Python Go 开发指南Go

本文为您介绍分布式消息服务MQTT的环境准备内容。 在创建天翼云分布式消息服务MQTT实例之前，您需要做一些准备工作。 首先，您需要设置一个虚拟私有云（VPC），这是一个隔离的网络环境，用于托管MQTT实例。 接下来，您需要创建一个子网，它是VPC内部的一个子网络，用于划分不同的部分和区域。 最后，您需要配置一个安全组，用于控制入站和出站的流量规则，以保证MQTT实例的安全性。 每个分布式消息服务MQTT实例都会被部署在特定的VPC中，并与特定的子网和安全组相关联。这种方式可以让您自主配置和管理MQTT实例的网络环境，并提供安全保护策略。如果您已经有了现成的VPC、子网和安全组，可以重复使用它们，无需重复创建。这样可以节省时间和资源，并确保一致性和可靠性。 VPC和子网 VPC和子网可重复使用，您也可以使用不同的VPC和子网来配置MQTT实例，您可根据实际需要进行配置。在创建VPC和子网时应注意如下要求： VPC与使用的天翼云分布式消息服务MQTT服务应在相同的区域。 如无特殊需求，创建VPC和子网的配置参数使用默认配置即可。 创建VPC和子网的操作请参考虚拟私有云创建VPC、子网搭建私有网络。 若需要在已有VPC上创建和使用新的子网，请参考虚拟私有云子网管理创建子网。

介绍分布式消息服务Kafka 如何提高消息处理效率。 要提高Kafka的消息处理效率，可以考虑以下几个方面： 分区和副本设置：合理设置分区和副本数量，可以提高并行处理能力和容错性。较大的分区数可以增加并行处理的能力，而较多的副本数可以提高数据的冗余和可用性。 优化消息生产者：在消息生产者端，可以采取一些措施来提高效率。例如，使用批量发送来减少网络开销，使用异步发送来提高吞吐量，设置适当的缓冲区大小来避免频繁的IO操作等。 优化消息消费者：在消息消费者端，可以通过增加消费者实例来提高并行处理能力。同时，可以使用多线程或多进程方式来并行处理消息，提高处理效率。 合理设置Kafka参数：根据实际情况，可以调整Kafka的一些参数来提高性能。例如，调整消息的最大大小、网络缓冲区大小、批量发送的大小等。 使用分区和消费者组：合理使用分区和消费者组可以提高消息的负载均衡和并行处理能力。分区可以将消息分散到多个消费者实例上，而消费者组可以将消息分发给不同的消费者组成的消费者实例。 监控和调优：定期监控Kafka集群的性能指标，如吞吐量、延迟等，并进行调优。可以通过调整参数、增加资源、优化代码等方式来提高性能。 总之，提高Kafka的消息处理效率需要综合考虑多个因素，包括分区和副本设置、优化生产者和消费者、调整参数、硬件优化等。根据实际情况进行优化，可以提高Kafka的性能和吞吐量。 对使用分布式消息服务kafka的生产者和消费者有如下的使用建议：