参数 参数说明 区域 选择区域。 不同区域的云服务产品之间内网互不相通。请就近选择靠近您业务的区域，可减少网络时延，提高访问速度。 集群名称 集群名称不允许重复。 只能由字母、数字、中划线和下划线组成，并且长度为1～64个字符。 默认名称为mrsxxxx，xxxx为字母和数字的四位随机组合数，系统自动组合。 集群版本 当前版本为MRS 3.1.0。 集群类型 提供几种集群类型： 分析集群：用来做离线数据分析，提供的是Hadoop体系的组件。 流式集群：用来做流处理任务，提供的是流式处理组件。 混合集群：既可以用来做离线数据分析，也可以用来做流处理任务，提供的是Hadoop体系的组件和流式处理组件。建议同时需要做离线数据分析和流处理任务时使用混合集群。 自定义：用户可按照业务需求调整集群服务的部署方式，具体请参见 说明 MRS流式集群不支持“作业管理”和“文件管理”功能。如需在集群中安装全部组件，请选择“自定义”类型集群。 组件选择 MRS配套的组件如下： 分析集群组件 Presto：开源、分布式SQL查询引擎。 Hadoop：分布式系统基础架构。 Spark：内存分布式系统框架。（MRS 3.x版本不支持） Spark2x：Spark2x是一个对大规模数据处理的快速和通用引擎,基于开源Spark2.x版本开发。（仅MRS 3.x版本支持） Hive：建立在Hadoop上的数据仓库框架。 HBase：分布式列数据库。 Tez：提供有向无环图的分布式计算框架。 Hue：提供Hadoop UI能力，让用户通过浏览器分析处理Hadoop集群数据。 Loader：基于开源sqoop 1.99.7开发，专为Apache Hadoop和结构化数据库（如关系型数据库）设计的高效传输大量数据的工具。（MRS 3.x版本不支持） Hadoop为必选组件，且Spark与Hive组件需要配套使用。请根据业务选择搭配组件。 Flink：分布式大数据处理引擎，可对有限数据流和无限数据流进行有状态计算。 Oozie：Hadoop作业调度系统。（仅MRS 3.x版本支持） HetuEngine：HetuEngine是一个同异构大数据集的分布式SQL查询引擎。（仅MRS 3.1.x LTS版本支持） Ranger：一个基于Hadoop平台监控和管理数据安全的框架。 Impala：一种处理大量数据的SQL查询引擎。 ClickHouse：ClickHouse是一个用于联机分析(OLAP)的列式数据库管理系统(DBMS)。CPU架构为鲲鹏计算的ClickHouse集群表引擎不支持使用HDFS和Kafka。 Kudu：一种列存储管理器。 流式集群组件 Kafka：提供分布式消息订阅的系统。 Flume：提供分布式、高可用、高可靠的海量日志采集、聚合和传输系统。 ZooKeeper：一个集中的服务，它用于维护配置信息、命名、提供分布式的同步和提供分组服务。（仅MRS 3.x版本支持） Ranger：一个基于Hadoop平台监控和管理数据安全的框架。（仅MRS 3.x版本支持） 元数据 是否使用外部数据源存储元数据。 本地元数据： 元数据存储在集群本地。 数据连接：使用外部数据源元数据，若集群异常或删除时将不影响元数据，适用于存储计算分离的场景。 支持Hive或Ranger组件的集群支持该功能。 组件名 当“元数据”选择“数据连接”时该参数有效。用于表示可以设置外部数据源的组件类型。 Hive Ranger 数据连接类型 当“元数据”选择“数据连接”时该参数有效。用于表示外部数据源的类型。 Hive组件支持的数据连接类型： − RDS服务MySQL数据库− 本地数据库 Ranger组件支持的数据连接类型： − RDS服务MySQL数据库− 本地数据库 数据连接实例 当“数据连接类型”选择“RDS服务MySQL数据库”时，该参数有效。用于表示MRS集群与RDS服务数据库连接的名称，该实例必选先创建才能在此处引用。可单击“创建数据连接”进行创建，具体请参考

介绍分布式消息服务Kafka按时间查询消息功能操作内容。 场景描述 Kafka按时间查询是指通过指定时间范围来查询Kafka主题中的消息。以下是一些常见的按时间查询的场景描述： 数据分析和报告：在数据分析和报告生成的场景中，经常需要按时间查询Kafka主题中的消息。通过指定起始时间和结束时间，可以获取在特定时间范围内产生的消息，用于进一步的数据分析和报告生成。 故障排查：当出现故障或问题时，按时间查询可以帮助定位问题的发生时间和相关的消息。通过指定故障发生的时间范围，可以获取相关的消息，用于故障排查和问题分析。 监控和警报：按时间查询还可以用于监控和警报系统。通过定期按时间查询Kafka主题中的消息，可以检查是否有异常或异常事件发生，并触发相应的警报机制。 数据回溯和重播：按时间查询功能还可以用于数据回溯和重播。通过指定特定的时间范围，可以获取过去某个时间段内的消息，并进行数据回溯或重播操作。 数据同步和复制：在数据同步和复制的场景中，按时间查询可以帮助确保数据的一致性。通过按时间查询源和目标主题中的消息，可以比较不同时间段内的消息，并进行数据同步和复制操作。 操作步骤 （1）登录管理控制台。 （2）进入Kafka管理控制台。 （3）在实例列表页在操作列，目标实例行点击“管理”。 （4）点击“消息查询”后默认就是按点位查询。 （5）选择按时间查询，选择需要查询的Topic，输入分区号以及时间段，点击”查询“按钮，会返回时间段内的所有消息。 （6）点击消息列表的“消息详情”可查看消息详情信息。

本章节介绍如何删除SASLSSL用户。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例名称，进入实例详情页面。 步骤 5 通过以下任意一种方法，删除SASLSSL用户。 在“用户管理”页签，在待删除的SASLSSL用户所在行，单击“删除”。 在“用户管理”页签，勾选SASLSSL用户名左侧的方框，可选一个或多个，单击信息栏左上侧的“删除”。 说明 创建Kafka实例时设置的SASLSSL用户无法删除。 步骤 6 在弹出的“删除用户”对话框中，单击“是”，完成SASLSSL用户的删除。

介绍分布式消息服务Kafka使用MirrorMaker跨集群数据同步。 应用场景 使用MirrorMaker进行跨集群数据同步是Kafka中常见的场景之一。MirrorMaker是Kafka提供的一个工具，用于将一个Kafka集群中的消息复制到另一个Kafka集群，实现跨数据中心、跨地域或跨集群的数据同步。 以下是一些常见的使用MirrorMaker进行跨集群数据同步的场景： 1. 备份和灾备： MirrorMaker可用于将生产环境中的消息复制到备份集群，以实现数据的备份和灾备。当主集群发生故障或不可用时，可以切换到备份集群，确保数据的可用性和连续性。 2. 数据复制和分发： 如果有多个数据中心或地理位置，可以使用MirrorMaker将消息从一个集群复制到另一个集群，实现数据的复制和分发。这样可以将数据从一个地区或数据中心传输到另一个地区或数据中心，以满足不同地区或数据中心的业务需求。 3. 跨云厂商数据同步： 当使用多个云厂商的Kafka服务时，MirrorMaker可以用于将消息从一个云厂商的Kafka集群复制到另一个云厂商的Kafka集群，实现跨云厂商的数据同步和迁移。 4. 数据聚合和分析： 当需要将多个Kafka集群中的消息聚合到一个集群中进行分析时，可以使用MirrorMaker将消息从多个集群复制到目标集群，以实现数据的聚合和分析。 需要注意的是，MirrorMaker在进行数据同步时，会引入一定的延迟，并且在网络传输和资源消耗方面会有一定的开销。因此，在使用MirrorMaker进行跨集群数据同步时，需要根据实际情况进行配置和调优，以满足业务需求和性能要求。

修改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终端节点需要放通此网段。

本章节介绍如何在控制台创建消费组。 如果在“配置参数”中开启了“auto.create.groups.enable”，则该操作为可选。“auto.create.groups.enable”表示在消费者进入未创建的消费组时，会自动创建此消费组。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例的名称，进入实例详情页面。 步骤 5 在左侧导航栏选择“消费组管理”，进入消费组列表页面。 步骤 6 单击“创建消费组”，弹出“创建消费组”对话框。 步骤 7 输入消费组名称，单击“确定”。 在消费组列表页面，查看新创建的消费组。

后续操作 开启Smart Connect后，您可以配置Kafka间的数据复制、配置Kafka数据转储至OBS，实现分布式消息服务Kafka和其他云服务之间的数据同步。

本文主要介绍查看磁盘使用量。 本章节指导您在控制台查看每个代理上磁盘的使用量。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例的名称，进入实例的“基本信息”页签。 步骤 5 在“磁盘存储统计”页签，查看磁盘的使用量。 图 查看磁盘的使用量 页面支持从以下三个方面对代理中的Topic磁盘使用量进行查询：使用量排名前xx 个、使用量达到xx GB和使用占比达 xx %。 在页面右上角，单击“查看监控数据”，跳转到云监控服务的监控指标页面。在监控指标页面，您可以查看Kafka实例的各项监控指标。

介绍Logstash对接Kafka具体内容。 应用场景 通过Logstash对接Kafka，可以实现以下功能： 1. 数据收集：Logstash可以从Kafka主题中消费数据，将数据从Kafka集群中获取到Logstash中进行处理和转发。这样可以方便地将分布式系统、应用程序、传感器数据等各种数据源的数据集中收集起来。 2. 数据处理和转换：Logstash提供了丰富的过滤器插件，可以对从Kafka中消费的数据进行各种处理和转换操作。例如，可以进行数据清洗、解析、分割、合并、字段映射等操作，以满足不同数据源和目标的数据格式要求。 3. 数据传输和转发：Logstash可以将处理后的数据发送到不同的目标位置，如Elasticsearch、MySQL、文件系统、消息队列等。通过配置适当的输出插件，可以将数据传输到目标系统，以便后续的数据分析、存储、可视化等操作。 4. 实时数据处理：Logstash与Kafka结合使用，可以实现实时的数据处理和传输。Kafka作为高吞吐量的消息队列，可以确保数据的高效传输和缓冲。而Logstash作为数据处理引擎，可以对从Kafka中消费的数据进行实时处理，满足实时数据分析和监控的需求。 5. 分布式部署和负载均衡：Logstash支持分布式部署，可以通过配置多个Logstash节点来实现高可用性和负载均衡。多个Logstash节点可以同时从Kafka主题中消费数据，并进行并行处理和转发，以提高整体系统的性能和吞吐量。 总之，通过Logstash对接Kafka，可以实现灵活、可扩展和高效的数据处理和传输。Logstash提供了丰富的插件和配置选项，可以根据实际需求进行定制化的数据处理流程。同时，Logstash还具有良好的可扩展性和可靠性，适用于各种规模和类型的数据处理场景。

本章节介绍了如何创建Topic。 操作场景 Topic，即消息主题，消息发送与接收的基本单元。购买RocketMQ实例成功后，需要手动创建Topic，然后才能进行生产消息和消费消息。 前提条件 已购买RocketMQ实例。 操作步骤 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，单击“Topic管理”，进入“Topic管理”页面。 4. 单击“创建Topic”，弹出“创建Topic”页面。 5. 参考表1，填写Topic名称和配置信息。 表 Topic参数说明 参数 说明 :: Topic名称 您可以自定义Topic名称，用于区分不同的Topic。 Topic名称不能设置为相同，否则会创建失败。 创建Topic后不能修改名称。 说明： 如果Topic名称中包含“%”或“ 权限 Topic的权限，包括发布+订阅、发布和订阅。 关联代理 选择在某个代理上创建Topic，并设置Topic的队列个数。 如果实例部署在多个代理上，单击“添加关联代理”，可以在其他代理上创建Topic，并设置Topic的队列个数。 图 创建Topic 6. 配置完成后，单击“确定”，完成Topic的创建。

本文主要介绍如何设置消息堆积数超过阈值时，发送告警短信/邮件。 操作场景 如果您想要在消费组的消息堆积数超过阈值时，通过短信/邮件及时收到通知信息，可以参考本章节设置告警通知。 您还可以参考分布式消息服务Kafka的支持的监控指标，设置告警通知。 前提条件 已购买Kafka实例、创建Topic，并且已成功消费消息。 操作步骤 1、 登录分布式消息服务Kafka控制台，单击待创建告警通知的实例名称，进入实例详情页。 2、 在左侧导航栏，选择“监控”，进入监控页面。 3、 在“消费组”页签，设置需要创建告警通知的消费组。 图 选择需要创建告警通知的消费组 消费组：选择需要创建告警通知的消费组。 主题：选择“全部Topic”。 4、 选中“消息堆积数（消费组可消费消息数）”图表，单击，创建告警规则。 图 消息堆积数图表 5、 在“创建告警规则”界面，设置告警名称。 图 设置告警名称 名称：您自定义的告警名称，用于识别不同的告警。 描述：告警规则描述，可以不填。 6、 在“创建告警规则”界面，设置告警策略。 图 设置告警策略 触发规则：选择“自定义创建”。 告警策略：触发告警规则的告警策略，是否触发告警取决于连续周期的数据是否达到阈值。 告警级别：根据实际情况选择告警等级。 7、 在“创建告警规则”界面，设置告警通知对象。 图 设置告警通知对象 发送通知：选择开启。 通知方式：选择“主题订阅”。 通知对象：选择云帐号联系人或已创建的告警通知主题，告警通知主题的订阅信息中包含需要接收告警信息的手机号/邮箱地址。 图 创建告警通知主题 图 添加订阅 生效时间：该告警规则仅在生效时间内发送通知消息。 触发条件：触发告警通知的条件。 8、 在“创建告警规则”界面，设置企业项目和标签。 图 设置企业项目和标签 归属企业项目：告警规则所属的企业项目。只有拥有该企业项目权限的用户才可以查看和管理该告警规则。 标签：标签用于标识云资源，当您拥有相同类型的许多云资源时，可以使用标签按各种维度（例如用途、所有者或环境）对云资源进行分类。 9、 单击“立即创建”，完成告警规则的设置。 告警规则创建完成后，在云监控服务的“告警 > 告警规则”界面，查看新创建的告警规则。 图 查看新创建的告警规则

介绍分布式消息服务Kafka按位点查询的功能操作内容。 场景描述 Kafka按位点查询是指通过指定偏移量或时间戳来查询Kafka主题中的消息。以下是一些常见的按位点查询的场景描述： 数据回溯：当需要回溯到过去的某个时间点或特定的消息偏移量时，可以使用按位点查询功能。这在故障排查、数据分析和报告生成等场景中非常有用，可以准确地获取特定时间点或消息的数据。 数据恢复：在某些情况下，如果由于意外情况导致数据丢失或损坏，可以使用按位点查询来恢复丢失的数据。通过指定偏移量或时间戳，可以定位到丢失数据之前的位置，然后将数据重新消费或导出。 数据验证：按位点查询还可以用于数据验证和一致性检查。通过指定特定的偏移量或时间戳，可以比较不同主题或分区之间的消息，以确保数据的一致性和正确性。 数据分析：按位点查询对于数据分析和统计也非常有用。通过指定特定的时间戳或偏移量范围，可以获取所需的数据子集，并对其进行进一步的分析和处理。 操作步骤 （1）登录管理控制台。 （2）进入Kafka管理控制台。 （3）在实例列表页在操作列，目标实例行点击“管理”。 （4）点击“消息查询”后默认就是按位点查询。 （5）下拉选择topic可以切换要查询的topic，填写offset消息位置，partition分区编号，可查询该分区offset位置开始的10条消息。 （6）点击消息列表的“消息详情”可查看消息详情信息

服务后缀 服务说明 zos dns msgc crs apm lts faas gts 全局事务：该服务由天翼云全局事务服务提供，全局事务服务是一款专为实现分布式环境下高性能事务一致性而设计的服务工具。它可以与MySQL、PostgreSQL等数据源，以及Spring Cloud、Dubbo等RPC框架和消息队列等中间件产品配合混合使用，以支持用户各种分布式数据库事务、多库事务、消息事务和服务链路级事务的组合需要 prome 应用性能监控：Prometheus监控为云上托管Prometheus服务，可为您的容器集群提供多种开箱即用的预置监控大盘与监控规则，实现免搭建的高效运维场景。 ags Agent沙箱：Agent 沙箱提供安全、弹性、可扩展的云端沙箱运行环境。在使用Agent沙箱产品时，通过连接该终端节点服务，可实现在内网调用沙箱服务。

本文介绍分布式消息服务MQTT的keepalive设置推荐 在MQTT中，Keep Alive是一个用于保持客户端和服务器之间的活动连接的机制。它是通过客户端定期发送PINGREQ消息来实现的，以确保连接保持活动状态。当服务器接收到PINGREQ消息后，会回复PINGRESP消息，表示连接仍然活动。 Keep Alive的设置取决于你的应用需求和网络环境。以下是一些建议： 1. 根据网络稳定性设置：如果你的网络环境非常稳定，连接很少会断开，你可以将Keep Alive设置为较长的时间，例如60秒或更长。这样可以减少PINGREQ和PINGRESP消息的频繁发送，减少网络流量和资源消耗。 2. 根据应用实时性设置：如果你的应用对实时性要求较高，需要及时获取和响应消息，则可以将Keep Alive设置为较短的时间，例如10秒或更短。这样可以快速检测到连接丢失，并尽快重新连接或采取相应的处理措施。 3. 根据网络延迟设置：如果你的网络延迟较高，连接速度较慢，你可能需要将Keep Alive设置为较长的时间，以允许更多的时间来处理网络延迟和连接恢复。 需要注意的是，Keep Alive的设置不应过于频繁或过于长久。太频繁的Keep Alive会增加网络流量和服务器负载，太长久的Keep Alive可能会导致连接超时和断开。 一般来说，根据网络稳定性和应用实时性的考虑，将Keep Alive设置为30秒到5分钟之间是比较常见的。你可以根据你的具体需求和实际情况进行调整和优化。

本章节介绍了如何修改分布式消息服务RocketMQ实例的Topic。 操作场景 Topic创建成功后，您可以根据业务需要修改以下参数：读队列个数、写队列个数、权限和添加关联代理。 操作步骤 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，单击“Topic管理”，进入“Topic管理”页面。 4. 选择以下任意一种方法修改Topic参数。 在待修改的Topic所在行，单击“编辑”，弹出“编辑Topic”页面。 单击待修改的Topic名称，进入Topic详情页面。在页面右上角，单击“编辑”，弹出“编辑Topic”页面。 5. 修改如表1所示配置信息。 表 Topic参数说明 参数 说明 权限 Topic的权限，包括发布+订阅、发布和订阅。 关联代理 修改读队列个数或写队列个数。 lic读数读队列个数：Top据可用的总队列数。 写队列个数：Topic写数据可用的总队列数。 如果创建Topic时未关联所有代理，单击“添加关联代理”，可以在其他代理上创建Topic，并设置读队列个数和写队列个数。 6. 修改完成后，单击“确定”。

本文主要介绍 重启Kafka Manager。 操作场景 当Kafka Manager无法登录或者无法使用时，例如下图中的报错，可以通过重启Kafka Manager，使Kafka Manager恢复正常。 图 报错信息 说明 重启Kafka Manager不会影响业务。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 通过以下任意一种方法，重启Kafka Manager。 在待重启Manager的Kafka实例所在行，单击“更多 > 重启Manager”。 单击Kafka实例名称，进入实例详情页面。单击右上角的“更多 > 重启Manager”。 步骤 5 单击“确定”。 您可以在实例的“后台任务管理”页面，查看当前任务的操作进度。任务状态为“成功”，表示重启成功。

查看消费者列表（Kafka客户端） 未开启SASL的Kafka实例，在“/{命令行工具所在目录}/kafka{version}/bin/”目录下，通过以下命令查询消费者列表。 ./kafkaconsumergroups.sh bootstrapserver {brokerip}:{port} group {groupname} members describe 已开启SASL的Kafka实例，通过以下步骤查询消费者列表。 （可选）如果已经设置了SSL证书配置，请跳过此步骤。否则请执行以下操作。在Kafka客户端的“/config”目录中创建“ssluserconfig.properties”文件，参考步骤3增加SSL证书配置。 在“/{命令行工具所在目录}/kafka{version}/bin/”目录下，通过以下命令查询消费者列表。 ./kafkaconsumergroups.sh bootstrapserver {brokerip}:{port} group {groupname} members describe commandconfig ./config/ssluserconfig.properties 查看消费进度（控制台） 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例的名称，进入实例详情页面。 步骤 5 在左侧导航栏选择“消费组管理”，进入消费组列表页面。 步骤 6 单击待查看消费进度的消费组名称，进入消费组详情页。 步骤 7 在“消费进度”页签，查看此消费组订阅的Topic列表、Topic中消息的总堆积数、Topic每个分区的堆积数、Topic每个分区的偏移量和最大消息位置。 图 消费进度 步骤 8 （可选）如果需要查询某个Topic的消费进度，在搜索框中，输入Topic名称，单击。

介绍分布式消息服务Kafka的新建消费组功能操作介绍。 场景描述 Kafka新建消费组的场景如下： 多个应用程序需要独立消费同一个主题：如果有多个应用程序需要从同一个Kafka主题中独立消费消息，并且每个应用程序需要管理自己的消费进度和分区分配，那么可以为每个应用程序创建一个独立的消费组。 消费者需要实现消息的负载均衡：当有多个消费者需要消费同一个主题的消息时，可以将它们加入同一个消费组，Kafka会自动将主题的分区均匀地分配给消费组中的消费者，实现消息的负载均衡。 消费者需要实现消息的并行处理：如果需要将一个主题的消息并行地处理，可以创建多个消费者实例，并将它们加入同一个消费组。Kafka会将主题的分区分配给消费组中的消费者，每个消费者只消费自己被分配的分区，从而实现消息的并行处理。 消费者需要实现高可用性：为了提高消费者的可用性，可以创建一个消费组，并将多个消费者实例加入该组。当一个消费者实例发生故障时，Kafka会将该消费者的分区重新分配给其他健康的消费者，从而保证消息的连续消费。 操作步骤 （1）登录管理控制台。 （2）进入Kafka管理控制台。 （3）在实例列表页在操作列，目标实例行点击“管理”。 （4）点击“消费组管理”后进入消费组管理页面。 （5）点击“创建消费组”后，输入消费组名称，点击创建。 说明 消费组业务应用接入使用时客户端也可自动创建。

本节主要介绍Kafka Connect接入Kafka。 最佳实践概述 场景描述 Kafka Connect可以将完整的数据库注入到Kafka的Topic中，或者将服务器的系统监控指标注入到Kafka，然后像正常的Kafka流处理机制一样进行数据流处理。而导出工作则是将数据从Kafka Topic中导出到其它数据存储系统、查询系统或者离线分析系统等。 如果您想要在流数据平台场景下，使用Kafka Connect快速实现数据从Kafka和其它系统之间进行导入导出，可以参考本实践方案。 技术架构图 暂无。 方案优势 用户可按业务需要，参考Kafka Connect数据导入导出，实现流数据的消息消费。 前提条件 需已购买Kafka实例、创建Topic，并且已成功消费消息。 资源规划 本实践方案内容仅涉及Kafka专享版实例和Flink的安装配置。 分布式消息服务 Figure 1 分布式消息服务 资源类型 配置项 配置明细 说明 :::: 企业中间件 DMS Kafka专享实例 需已购买kafka专享实例，创建好Topic，并成功消费消息。 Kafka Connect流数据平台 Figure 2 Kafka Connect流数据平台 资源类型 配置项 配置明细 说明 :::: 应用软件 Flink框架 Flink 1.14 使用开源Apache Flink 方案正文 Kafka Connect 目前支持两种执行模式：standalone 和 distributed。 standalone模式 通过以下命令以 standalone 模式启动 connect： 接入Kafka专享版与接入开源 Apache Kafka 没有区别，仅需要修改 bootstrap.servers 为申请实例时分配的 IP。 distributed模式 通过以下命令以 distributed 模式启动 connect： 该模式下，Kafka Connect 会将 offsets、configs 和 task status 信息存储在 Kafka Topic 中，存储的Topic 在 connectdistributed 中的以下字段配置： 这三个 Topic 需要手动创建，才能保证创建的属性符合 connect 的要求。 config.storage.topic 需要保证只有一个 partition，多副本且为 compact 模式。 offset.storage.topic 需要有多个 partition，多副本且为 compact 模式。 status.storage.topic 需要有多个 partition，多副本且为 compact 模式。 配置 bootstrap.servers 为申请实例是分配的 IP。 配置 group.id，用于标识 connect 集群，需要与消费者分组区分开来。

Kafka触发器可以订阅天翼云提供的分布式消息队列Kafka实例，并根据消息触发关联的函数，借此能力，使得函数可以消费指定Topic的消息，执行自定义处理逻辑。 注意事项 Kafka触发器订阅的Kafka实例必须和函数计算的函数实例在相同地域。 前提条件 已创建函数。 已开通分布式消息Kafka实例（KAFKA引擎版），详情请参考创建分布式消息服务Kafka实例。 已创建Topic，创建GroupID（可选）。 操作步骤 1. 登录函数计算控制台，点击目标函数，进入函数详情。 2. 选择详情下顶部的配置选项卡。 3. 在配置选项卡 中，选择左边的触发器选项卡。 4. 点击创建触发器 ，在弹出的右抽屉中选择Kafka触发器，配置参数解释如下。 配置项 操作 示例 触发器类型 选择Kafka触发器。 Kafka触发器 名称 填写自定义的触发器名称。 kafkatrigger 版本或别名 默认值为LATEST，支持选择任意函数版本或函数别名。 LATEST 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

本页面主要介绍分布式消息服务MQTT对接的云审计服务使用和查看方法。 操作场景 本服务现已对接天翼云云审计服务，云审计服务提供对各种云资源操作的记录和查询功能，用于支撑合规审计、安全分析、操作追踪和问题定位等场景，同时提供事件跟踪功能，将操作日志转储至对象存储实现永久保存。 云审计可提供的功能服务具体如下： ● 记录审计日志：支持用户通过管理控制台或API接口发起的操作，以及各服务内部自触发的操作。 ● 审计日志查询：支持在管理控制台对7天内操作记录按照事件类型、事件来源、资源类型、筛选类型、操作用户和事件级别等多个维度进行组合查询。 使用限制 ● 云审计服务本身免费，包括时间记录以及7天内时间的存储和检索。 ● 用户通过云审计能查询到多久前的操作事件：7天。 ● 用户操作后多久可以通过云审计查询到数据：5分钟。 ● 其它限制请参考使用限制云审计。 操作步骤 1. 开通云审计服务。 参见开通云审计服务云审计。 2. 查看云审计事件。 参见查看审计事件云审计。 3. 在事件列表中，选择事件来源为“容器与中间件”，资源类型选择“分布式消息服务Kafka”，上方时间选择需要筛选的时间段。点击查询即可。 4. 在审计事件右侧点击详情，可以看到更详细的事件信息。 更多云审计相关使用说明和常见问题请参考用户指南、常见问题。

云原生网关 左侧导航栏，选择资源管理 > 云原生网关。在云原生网关列表展示了当前环境已导入的云原生网关资源。 点击导入按钮，列表会展示与当前环境同VPC的云原生网关列表，勾选云原生网关，点击保存就可将资源导入。 导入失败可点击日志查看具体失败原因，也可点击重试按钮进行重试。 点击移除按钮，可移除选择的云原生网关。 关系数据库 左侧导航栏，选择资源管理 > 关系数据库。在关系数据库列表展示了当前环境已导入的关系数据库资源。 点击导入按钮，列表会展示与当前环境同VPC的关系数据库列表，勾选关系数据库，点击保存就可将资源导入。 导入失败可点击日志查看具体失败原因，也可点击重试按钮进行重试。 点击移除按钮，可移除选择的关系数据库。 分布式缓存 左侧导航栏，选择资源管理 > 分布式缓存。在分布式缓存列表展示了当前环境已导入的分布式缓存资源。 点击导入按钮，列表会展示与当前环境同VPC分布式缓存列表，勾选分布式缓存，点击保存就可将资源导入。 导入失败可点击日志查看具体失败原因，也可点击重试按钮进行重试。 点击移除按钮，可移除选择的分布式缓存。 分布式消息 左侧导航栏，选择资源管理 > 分布式消息。在分布式消息列表展示了当前环境已导入的分布式消息资源。 点击导入按钮，列表会展示与当前环境同VPC分布式消息列表，勾选分布式消息，点击保存就可将资源导入。 导入失败可点击日志查看具体失败原因，也可点击重试按钮进行重试。 点击移除按钮，可移除选择的分布式消息。

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

面向长期稳定客户，我们特别推出了更加优惠的包年订阅选项，旨在通过深度折扣，帮助客户显著降低资源单位成本，优化整体支出。 自2024年11月9日起订购和续订分布式消息服务MQTT产品更长包周期即可享受2年7折，3年5折优惠。

准备资源 要求 创建指导 VPC和子网 不同的RocketMQ实例可以重复使用相同的VPC和子网，也可以使用不同的VPC和子网，请根据实际需要进行配置。在创建VPC和子网时应注意如下要求：·创建的VPC与使用的分布式消息服务RocketMQ应在相同的区域。·创建VPC和子网时，如无特殊需求，配置参数使用默认配置即可。 创建VPC和子网的操作指导请参考创建虚拟私有云和子网，若需要在已有VPC上创建和使用新的子网，请参考为虚拟私有云创建新的子网。 安全组 不同的RocketMQ实例可以重复使用相同的安全组，也可以使用不同的安全组，请根据实际需要进行配置。在创建安全组时应注意如下要求：·创建安全组时，“模板”选择“自定义”。·创建安全组后，请保留系统默认添加的入方向“允许安全组内的弹性云服务器彼此通信”规则和出方向“放通全部流量”规则。·使用分布式消息服务RocketMQ必须添加表2所示安全组规则，其他规则请根据实际需要添加。 创建安全组的操作指导请参考创建安全组，为安全组添加规则的操作指导请参考添加安全组规则。

本节主要介绍分布式消息服务Kafka常用的应用场景 分布式消息服务Kafka主要适用于以下几种场景： 日志收集 构建应用系统和分析系统的桥梁，并将它们之间的关联解耦。 支持实时在线分析系统和类似于 Hadoop 的离线分析系统。 Kafka本身的性能是非常高效的，同时Kafka的特性决定它非常适合作为"日志收集中心"，这是因为Kafka在采集日志的时候业务是无感知的，其能够兼容自己的上游，能够直接地通过配置加密消息。当日志数据发送到Kafka集群里面，其实对于业务而言是完全无侵入的。同时其在下游又能够直接地对接Hadoop/ODPS等离线仓库存储和Strom/Spark等实现实时在线分析。在这样的情况之下，使用Kafka，只需要用户去关注整个流程里面的业务逻辑，而无需做更多的开发就能够实现统计、分析以及报表。 流计算处理 构建应用系统和分析系统的桥梁，并将它们之间的关联解耦。 通过支持流计算引擎，可对接开源 Storm/Samza/Spark 流计算引擎。 Kafka能够做到流计算处理，比如股市走向分析、气象数据测控、网站用户行为分析等领域，由于在这些领域中数据产生快、实时性强、数据量大，所以很难统一采集并入库存储后再做处理，这便导致传统的数据处理架构不能满足需求。而Kafka Stream以及Storm/Samza/Spark等流计算引擎的出现，可以根据业务需求对数据进行计算分析，最终把结果保存或者分发给需要的组件。

介绍分布式消息服务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不同，使用时请注意。

本节介绍产品使用卡券规则。 天翼云分布式消息服务RocketMQ产品卡券使用遵循天翼云统一规则。 ++优惠券使用++ ++代金券使用++ 卡券使用常见问题：卡券管理费用中心常见问题 天翼云。

消息可以批量生产和消费 为提高消息发送和消息消费效率，推荐使用批量消息发送和消费。通常，默认消息消费为批量消费，而消息发送尽可能采用批量发送。同时批量方式可有效减少API调用次数，减少服务使用费用。 如下面两张示意图对比所示，消息批量生产与消费，可以减少API调用次数，节约资源。 图 消息批量生产（发送）与消费 说明 批量发送消息时，单次不能超过10条消息，总大小不能超过512KB。 批量生产（发送）消息可以灵活使用，在消息并发多的时候，批量发送，并发少时，单条发送。这样能够在减少调用次数的同时保证消息发送的实时性。 图 消息逐条生产（发送）与消费 此外，批量消费消息时，消费者应按照接收的顺序对消息进行处理、确认，当对某一条消息处理失败时，不再需要继续处理本批消息中的后续消息，直接对已正确处理的消息进行确认即可。 巧用消费组协助运维 用户使用DMS服务作为消息管理系统，查看队列的消息内容对于定位问题与调试服务是至关重要的。 当消息的生产和消费过程中遇到疑难问题时，通过创建不同消费组可以帮助定位分析问题或调试服务对接。用户可以创建一个新的消费组，对队列中的消息进行消费并分析消费过程，这样不会影响其他服务对消息的处理。

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

本节介绍分布式消息服务Kafka集群模式 是的，Kafka实例通常是以集群模式创建的。Kafka集群由多个Kafka节点组成，每个节点负责存储和处理消息。集群模式可以提供高可用性、容错性和扩展性。 Kafka集群中的每个节点都可以承担多个分区的领导者或副本角色，从而实现数据的冗余和故障恢复。通过复制机制，Kafka可以提供高可用性和容错性，即使在节点故障的情况下也能保证数据的可靠性和可用性。 注意 Kafka集群的规模和配置取决于应用的需求和可用的资源。可以根据实际情况来决定集群中节点的数量、分区的数量以及数据的复制策略，以满足性能、可用性和扩展性的要求。

本节主要介绍分布式消息服务Kafka的退订说明 如果您有退订的需求，可以进行登录天翼云管理中心或产品控制台进行退订操作。天翼云目前支持7天无理由全额退订和非七天无理由退订以及其他退订，详细规则请参考文档费用中心退订。