创建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. 规则创建成功后可在告警规则页面查看其状态。

方案一：先迁生产，再迁消费 指先将生产消息的业务迁移到新的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倍。

云服务分类 云服务名称 产品控制台创建资源时是否支持绑定标签 产品控制台列表是否支持绑定和解绑标签 标签字符长度限制 标签字符内容限制 单实例默认可绑定标签个数 计算 弹性云主机 是 是 128位 首尾不包含空格 50 计算 GPU云主机 是 是 128位 首尾不包含空格 50 计算 物理机 是 是 128位 首尾不包含空格 50 计算 镜像服务 是 是 128位 首尾不包含空格 50 计算 弹性伸缩服务 是 是 128位 首尾不包含空格 50 计算 云主机快照 是 是 128位 首尾不包含空格 50 计算 SSH秘钥对 是 是 128位 首尾不包含空格 50 存储 云硬盘 是 是 128位 开头不包含空格 50 存储 弹性文件服务 否 是 128位 首尾不包含空格 50 存储 对象存储 是 是 128位 首尾不包含空格 50 存储 并行文件服务HPFS 是 是 128位 首尾不包含空格 50 存储 海量文件服务OceanFS 是 是 128位 首尾不包含空格 50 存储 云硬盘备份 是 是 128位 开头不包含空格 50 存储 云主机备份 是 是 128位 开头不包含空格 50 网络 弹性负载均衡 是 是 128位 首尾不包含空格 50 网络 共享流量包 否 是 128位 首尾不包含空格 50 网络 VPC终端节点 是 是 128位 首尾不包含空格 50 网络 NAT网关 是 是 128位 首尾不包含空格 50 网络 网关负载均衡 是 是 128位 首尾不包含空格 50 网络 内网DNS 否 是 128位 首尾不包含空格 50 网络 弹性IP 否 是 128位 首尾不包含空格 50 网络 共享带宽 否 是 128位 首尾不包含空格 50 网络 虚拟私有云 否 是 128位 首尾不包含空格 50 网络 流量镜像 否 是 128位 首尾不包含空格 50 网络 对等连接 是 是 128位 首尾不包含空格 50 网络 云间高速（标准版） 否 是 128位 首尾不包含空格 50 网络 VPN连接 否 是 128位 首尾不包含空格 50 网络 云专线CDA 是 是 128位 首尾不包含空格 50 专属云 专属云（计算独享型） 是 是 128位 首尾不包括空格 50 云原生 云容器引擎 是 是 128位 首尾不包含空格 50 云原生 容器镜像服务 否 是 128位 首尾不包含空格 50 云原生 云日志服务 否 是 128位 首尾不包含空格 50 云原生 应用性能监控 否 是 128位 首尾不包含空格 20 云原生 微服务云应用平台MSAP 否 是 128位 首尾不包含空格 50 云原生 微服务引擎API网关 否 是 128位 首尾不包含空格 20 云原生 微服务引擎微服务治理 否 是 128位 首尾不包含空格 20 云原生 微服务引擎注册配置中心 否 是 128位 首尾不包含空格 20 云原生 服务网格 否 是 128位 首尾不包含空格 50 云原生 分布式缓存服务Redis版 是 是 128位 首尾不包含空格 50 云原生 分布式消息队列RocketMQ 是 是 128位 首尾不包含空格 50 云原生 分布式消息队列RabbitMQ 是 是 128位 首尾不包含空格 50 云原生 分布式消息队列Kafka 是 是 128位 首尾不包含空格 50 云原生 分布式消息队列MQTT 是 是 128位 首尾不包含空格 50 云原生 弹性容器实例ECI 否 是 128位 首尾不包含空格 50 云原生 分布式容器云平台CCE ONE 是 是 128位 首尾不包含空格 20 云原生 Serverless容器引擎 是 是 128位 首尾不包含空格 50 云原生 函数计算 否 是 128位 首尾不包含空格 50 安全及管理 Web应用防火墙（原生版） 否 是 无限制 首尾不包含空格 50 安全及管理 服务器安全卫士（原生版） 否 是 128位 首尾不包含空格 10 安全及管理 云等保专区 否 是 128位 首尾不包含空格 50 安全及管理 数据库审计 否 是 128位 首尾不包含空格 10 安全及管理 云堡垒机（原生版） 否 是 128位 首尾不包含空格 10 安全及管理 日志审计（原生版） 否 是 128位 首尾不包含空格 10 安全及管理 秘钥管理 否 是 128位 首尾不包含空格 10 安全及管理 云密评专区 否 是 128位 首尾不包含空格 10 数据库 文档数据库服务 是 是 128位 首尾不包含空格 50 数据库 分布式关系型数据库 是 是 128位 首尾不包含空格 10 数据库 关系数据库PostgreSQL版 是 是 128位 首尾不包含空格 50 数据库 关系数据库MySQL版 是 是 128位 首尾不包含空格 50 数据库 云数据库ClickHouse版 是 是 128位 首尾不包含空格 50 数据库 关系数据库SQL Server版 是 是 128位 首尾不包含空格 50 数据库 数据传输服务DTS 否 是 128位 首尾不包含空格 50

说明 分布式消息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

容器与中间件 分类 云产品 支持审计的关键操作列表 容器与中间件 云容器引擎 云容器引擎支持的关键操作列表 容器与中间件 容器镜像服务 容器镜像服务支持的关键操作列表 容器与中间件 分布式消息服务Kafka 分布式消息服务Kafka支持的关键操作列表 容器与中间件 弹性容器实例 弹性容器实例服务支持的关键操作列表 数据库 分类 云产品 支持审计的关键操作列表 数据库 关系型数据库MySQL版 关系型数据库MySQL版服务支持的关键操作列表 数据库 关系型数据库PostgreSQL版服务 关系型数据库PostgreSQL版服务支持的关键操作列表 数据库 关系型数据库SQL Server版 关系型数据库SQL Server版服务支持的关键操作列表 数据库 分布式关系型数据库 分布式关系型数据库支持的关键操作列表 数据库 文档数据库服务 文档数据库服务支持的关键操作列表 数据库 分布式缓存服务Redis版 分布式缓存服务Redis版支持的关键操作列表 数据库 数据传输服务DTS 数据传输服务DTS支持的关键操作列表

配置项 操作说明 示例 触发器类型 选择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如下资源池支持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

本章节主要介绍配置Kafka/DMS Kafka源端参数。 作业中源连接为配置Kafka连接或配置DMS Kafka连接时，源端作业参数如下表所示。 表 Kafka作为源端时的作业参数 参数 说明 取值样例 Topics 支持单个或多个topic。 est1,est2 偏移量参数 从Kafka拉取数据时的初始偏移量： 最新：最大偏移量，即拉取最新的数据。 最早：最小偏移量，即拉取最早的数据。 已提交：拉取已提交的数据。 时间范围：拉取时间范围内的数据。 最新 是否持久运行 用户自定义是否永久运行。 是 消费组ID 用户指定消费组ID。 如果是从DMS Kafka导出数据，专享版请任意输入，标准版请输入有效的消费组ID。 sumergroup 数据格式 解析数据时使用的格式： 二进制格式：适用于文件迁移场景，不解析数据内容原样传输。 CSV格式：以CSV格式解析源数据。 l JSON：以JSON格式解析源数据。 CDC（DRSJSON）：以DRSJSON格式解析源数据。 二进制格式 字段分隔符 默认为空格，使用Tab键作为分隔符请输入“t”。 , 最大消息数/poll 可选参数，每次向Kafka请求数据限制最大请求记录数。 100 最大时间间隔/poll 可选参数，向Kafka请求数据的最大时间间隔。 100

本节介绍如何开启Kafka的Smart Connect功能。 Smart Connect用于Kafka实例和其他云服务（如OBS）之间的数据同步，或者两个Kafka实例之间的数据同步，实现数据的备份或迁移。 Smart Connect的使用流程为： 1. 开启Smart Connect。 2. 创建Smart Connect任务。 本章节主要介绍如何开启Smart Connect。 约束与限制 开启Smart Connect后，实例需要另外收取用于Smart Connect的代理费用。 例如：规格为kafka.4u8g.cluster的实例，会另外创建至少两个规格为kafka.4u8g的代理，用于Smart Connect，所以需要收取相应代理费用。 前提条件 已创建Kafka实例，且实例状态为“运行中”。 “auto.create.groups.enable”已设置为“true”。如果需要修改“auto.create.groups.enable”的取值，请参考修改Kafka实例配置参数。 开启Smart Connect 1、登录管理控制台。 2、在管理控制台左上角单击，选择Kafka实例所在的区域。 3、在管理控制台左上角单击，选择“应用服务 > 分布式消息服务 Kafka”，进入分布式消息服务Kafka专享版页面。 4、通过以下任意一种方法，开启Smart Connect。 在待开启Smart Connect的Kafka实例所在行，单击“更多 > 开启Smart Connect”。 单击Kafka实例名称，进入实例详情页面。单击右上角的“更多 > 开启Smart Connect”。 单击Kafka实例名称，进入实例详情页面。在“Smart Connect”后，单击。 单击Kafka实例名称，进入实例详情页面。在左侧导航栏单击“Smart Connect”，进入Smart Connect页面。单击“开启Smart Connect”。 5、单击，将Smart Connect设置为开启，并设置用于Smart Connect的代理数量，代理数取值范围为2~16个，单击“下一步”。 说明 代理数默认为2个，如果您预估2个Kafka实例间的同步流量比较大，例如大于50MB/s，请设置2个以上的代理数。 6、确认“Smart Connect”为开启状态，单击“提交”。 结束

本节介绍了分布式消息服务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、目前所有的可选规格如下：

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

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

分区平衡前的准备工作 在不影响业务的前提下，适当调小Topic老化时间并等待消息老化，减少迁移数据，加快迁移速度，分区平衡任务结束后可重新调整为初始值。 确保分区平衡的目标Broker磁盘容量充足。如果目标Broker磁盘剩余容量接近分区平衡迁移到该Broker上的数据量，为了给目标Broker上的消息生产预留存储空间，应先进行磁盘扩容，再进行分区平衡。 自动平衡 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例的名称，进入实例详情页面。 步骤 5 在左侧导航栏单击“Topic管理”，进入Topic列表页面。 步骤 6 通过以下任意一种方法，修改分区平衡。 勾选Topic名称左侧的方框，可选一个或多个，单击信息栏左上侧的“一键平衡 > 自动平衡”。 在待修改分区平衡的Topic所在行，单击“更多 > 一键平衡 > 自动平衡”。 步骤 7 设置自动平衡参数。 在“Brokers列表”区域，勾选目标Broker的名称，Topic分区的副本即将迁移至目标Broker上。 在“Topic列表”区域，输入Topic需要自动平衡的副本数，副本数须小于等于Broker的数量。 在“带宽限制”中，输入带宽大小，默认值为“1”，表示不限制带宽。如果实例负载较低，建议不设置带宽限制。如果需要设置带宽限制，建议该参数值不小于待分区平衡Topic的总生产带宽 待分区平衡Topic的最大副本数。带宽限制值的估算方法，请参考带宽限制值估算方法。 图设置自动平衡参数 步骤 7 单击“确定”，自动跳转到Topic列表页面。 在Topic列表页面左上方单击“查看变更任务”，进入“后台任务管理”页面，查看分区平衡任务的状态。当“状态”为“成功”时，表示分区平衡完成。 说明 分区平衡任务启动后，不能删除正在进行分区平衡的Topic，否则会导致分区平衡任务无法结束。 分区平衡任务启动后，无法修改Topic的分区数。 分区平衡任务启动后，无法手动停止任务，需要等到任务完成。

消息重试 对于有序消息：有序消息不能跳跃签收，当消费者消费消息失败后，消息队列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语法的过滤表达式过滤多个属性。

本文主要介绍分布式消息服务RabbitMQ的产品优势。 天翼云分布式消息服务RabbitMQ完全兼容开源社区版本，旨在为用户提供便捷高效的消息队列。业务无需改动即可快速迁移上云，为您节省维护和使用成本。 一键式部署，免去集群搭建烦恼 专享实例只需要在实例管理界面选好规格配置，提交订单，后台将自动创建部署完成一整套RabbitMQ实例。 兼容开源，业务零改动迁移上云 兼容社区版RabbitMQ的API，具备原生RabbitMQ的所有消息处理特性。 业务系统基于开源的RabbitMQ进行开发，只需加入少量认证安全配置，即可使用天翼云分布式消息服务RabbitMQ，做到无缝迁移。 说明 RabbitMQ实例兼容开源社区RabbitMQ 3.8.35版本。 独占式体验 RabbitMQ实例采用物理隔离的方式部署，租户独占RabbitMQ实例，每个RabbitMQ之间互不影响。 高性能 单队列性能最高可达10万TPS（默认配置），增加队列可获得更高性能。 数据安全 独有的安全加固体系，提供业务操作云端审计，消息存储加密等有效安全措施。 在网络通信方面，除了提供SASL认证，还借助虚拟私有云（VPC）和安全组等加强网络访问控制。 无忧运维 天翼云提供一整套完整的监控告警等运维服务，故障自动发现和告警，避免724小时人工值守。RabbitMQ专享实例自动上报相关监控指标，如分区数、主题数、堆积消息数等，并支持配置监控数据发送规则，您可以在第一时间通过短信、邮件等获得业务消息队列的运行使用和负载状态。

产品名称 可用性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%

本节介绍如何配置Kafka数据转储至OBS。 本章节介绍如何创建转储的Smart Connect任务，通过Smart Connect任务可以将Kafka实例中的数据转储至OBS中，实现消息数据的备份。 源Kafka实例中的数据会实时同步到转储文件中。 约束与限制 一个实例最多创建18个Smart Connect任务。 Smart Connect任务创建成功后，不支持修改任务参数。 前提条件 已开启Smart Connect。 已创建Kafka实例，且实例状态为“运行中”。 已创建的OBS桶和Kafka实例需要在同一个区域。 配置Kafka数据转储至OBS 1、登录管理控制台。 2、在管理控制台左上角单击，选择Kafka实例所在的区域。 3、在管理控制台左上角单击，选择“应用服务 > 分布式消息服务 Kafka”，进入分布式消息服务Kafka专享版页面。 4、单击Kafka实例名称，进入实例详情页面。 5、在左侧导航栏单击“Smart Connect”，进入Smart Connect任务列表页面。 6、单击“创建Smart Connect任务”，进入“创建smart connect”页面。 7、在“connect任务名称”中，输入Smart Connect任务的名称，用于区分不同的Smart Connect任务。任务名称需要符合命名规则：长度为4~64个字符，只能由英文字母、数字、中划线、下划线组成。 8、在“预置类型”中，选择“转储”。 9、在“源端配置”中，保持默认设置。 10、在“Topic配置”中，设置以下参数。 表11 Topic配置参数说明 参数 参数说明 正则表达式 表示以正则表达式的方式订阅Topic，并进行转储。 输入/选择 输入/选择待转储的Topic名称，支持选择多个Topic，以逗号“,”分隔。最多输入/选择20个Topic。 11、在“目标端配置”中，设置以下参数。 表12 目标端参数说明 参数 参数说明 转储启动偏移量 支持以下两种转储启动偏移量： 最早：最小偏移量，即读取最早的数据。 最新：最大偏移量，即获取最新的数据。 数据转储周期 两次转储的时间间隔，单位：秒，默认配置为300秒。 若某个时间段内无数据，则此时间段不会生成打包文件 AK 访问密钥ID。 SK 与访问密钥ID结合使用的密钥。 转储地址 设置存储Topic数据的OBS桶的名称。 选择：在下拉列表中选择已创建的OBS桶名称，或者单击“创建转储地址”，新创建一个OBS桶。 输入：输入已创建的OBS桶名称，或者单击“创建转储地址”，新创建一个OBS桶。输入的OBS桶必须和Kafka实例在同一个区域。 转储目录 OBS中存储Topic的目录，多级目录可以用“/”进行分隔。 时间目录格式 数据将存储在转储目录下的时间目录中，时间目录是按时间格式作为层级的目录。例如，当选择的时间目录格式精确到日时，存储目录为：“桶名称/转储目录/年/月/日”。 记录分隔符 在下拉列表中选择一种分隔符，分隔不同的转储记录 存储Key 是否转储Key，开启表示转储Key，关闭表示不转储Key。 说明 消息进行转储时，不支持以Key作为文件名。 12、单击“立即创建”，跳转到Smart Connect任务列表页面，页面右上角显示“创建xxx任务成功”。 结束

本文主要介绍 重置消费进度。 重置消费进度即修改消费者的消费位置。 说明 重置消费进度可能会导致重复消费，请谨慎操作。 前提条件 Kafka实例不支持在线重置消费进度，请先将待重置消费进度的消费组停止消费，然后重置消费进度。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例的名称，进入实例详情页面。 步骤 5 在左侧导航栏选择“消费组管理”，进入消费组列表页面。 步骤 6 单击待重置消费进度的消费组名称，进入消费组详情页。 步骤 7 在“消费进度”页签，通过以下方法，重置消费进度。 重置单个Topic所有分区的消费进度：在待重置消费进度的Topic后，单击“重置消费进度”。 重置单个Topic中单个分区的消费进度：在待重置消费进度的Topic分区后，单击“重置消费进度”。 步骤 8 在弹出的“重置消费进度”对话框中，参考下表，设置重置消费进度参数。 表 重置消费进度参数说明 参数 说明 重置类型 选择重置类型： 时间：重置消费进度到指定的时间。 偏移量：重置消费进度到指定的偏移量。 一键重置消费进度只支持重置消费进度到指定时间。 时间 当“重置类型”为“时间”时，需要设置此参数。 选择重置消费进度的时间点，重置完成后，将从此时间点开始消费。 最早：最早偏移量 自定义：自定义时间点 最晚：最晚偏移量 偏移量 当“重置类型”为“偏移量”时，需要设置此参数。 选择重置消费进度的偏移量，此偏移量不能小于0，重置完成后，将从此偏移量开始消费。 步骤 9 单击“确定”，弹出确认对话框。 步骤 10 单击“是”，完成消费进度的重置。

配置项 操作 示例 触发器类型 选择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将Topic划分为多个分区，消息被存储在不同的分区中。在同一个消费组内，一个消费者可同时消费多个分区，但一个分区在同一时刻只能被一个消费者消费。 在消息处理过程中，如果客户端的消费速度跟不上服务端的发送速度，未处理的消息会越来越多，这部分消息就被称为堆积消息。消息没有被及时消费就会产生消息堆积，从而会造成消息消费延迟。 导致消息堆积的常见原因如下： 生产者短时间内生产大量消息到Topic，消费者无法及时消费。 消费者的消费能力不足（消费者并发低、消息处理时间长），导致消费效率低于生产效率。 消费者异常（如消费者故障、消费者网络异常等）导致无法消费消息。 Topic分区设置不合理（leader副本都集中在部分Broker中、分区数太少），或新增分区无消费者消费。 Topic频繁重平衡导致性能下降，消费效率降低。 实施步骤 从消息堆积的原因看，消息堆积问题可以从消费者端、生产者端和服务端三个方面进行处理。 消费者端 根据实际业务需求，合理增加消费者个数（消费并发度），建议消费者数量和分区数保持一致。 提高消费者的消费速度，通过优化消费者处理逻辑（减少复杂计算、第三方接口调用和读库操作），减少消费时间。 增加消费者每次拉取消息的数量：拉取数据/处理时间 > 生产速度。 生产者端 生产消息时，给消息Key加随机后缀，使消息均衡分布到不同分区上。（在实际业务场景中，为消息Key加随机后缀，会导致消息全局不保序，需根据实际业务判断是否适合给消息Key加随机后缀。） 服务端 合理设置Topic的分区数，在不影响业务处理效率的情况下，调大Topic的分区数量。 当服务端出现消息堆积时，对生产者进行熔断，或将生产者的消息转发到其他Topic。

步骤二：创建元数据导入任务 1.登陆管理控制台。 2.在管理控制台左上角单击，选择区域（请选择Kafka实例所在的区域）。 3.在管理控制台左上角单击，选择“企业中间件 > 分布式消息服务Kafka版”，进入分布式消息服务Kafka控制台页面。 4.单击集群迁移>元数据迁移>单击创建任务，进入实例详情页面。 5.输入任务名称，选择目标实例，单击上传元数据文件。 6.点击下一步，到Topic编辑页面，这里可以手动增加、删除、修改topic 7.点击下一步，到group编辑页面，这里可以手动增加、删除、修改group 8.点击创建，这样就生成了一条元数据迁移任务 步骤三：查看迁移结果 1.在左侧导航栏，单击实例列表。 2.在实例列表页面，单击目标实例名称。 3.查看资源列表。 4.在左侧导航栏，单击Topic 管理 ，在Topic 管理页面的Topic列表中查看已创建的Topic。 5.在左侧导航栏，单击消费组管理 ，在消费组管理页面的Group列表中查看已创建的Group。

本节介绍Kafka分区平衡功能 场景描述 分区平衡是指将分区的副本重新分配到不同的节点上，解决Kafka节点负载不均衡问题。 需要进行分区平衡的场景如下： 实例扩容节点个数后，新节点没有任何负载，需要将原有Topic分区的副本迁移到新节点上。 将高负载节点上的Leader分区切换为Follower分区。 增加/减少副本数量。 分布式消息服务Kafka版控制台提供两种分区平衡的方法：自动平衡和手动平衡，建议选择自动平衡，确保分区Leader的均匀分布。 操作影响 对数据量大的Topic进行分区平衡，会占用大量的网络和存储带宽，业务可能会出现请求超时或者时延增大，建议在业务低峰期时操作。对Topic进行分区平衡前，根据Kafka实例规格对比当前实例负载情况，评估是否可以进行分区平衡，建议预留足够的带宽进行分区平衡，CPU使用率在90%以上时，不建议进行分区平衡。 带宽限制是指设定Topic进行副本同步的带宽上限，确保不会对该实例上的其他Topic造成流量冲击。但需要注意，带宽限制不会区分是正常的生产消息造成的副本同步还是分区平衡造成的副本同步，如果带宽限制设定过小，可能会影响正常的生产消息，且可能会造成分区平衡一直无法结束。 分区平衡任务启动后，不能删除正在进行分区平衡的Topic，否则会导致分区平衡任务无法结束。 分区平衡任务启动后，无法修改Topic的分区数。 分区平衡任务启动后，无法手动停止任务，需要等到任务完成。 分区平衡后Topic的metadata会改变，如果生产者不支持重试机制，会有少量的请求失败，导致部分消息生产失败。 数据量大的Topic进行分区平衡的时间会比较长，建议根据Topic的消费情况，适当调小Topic老化时间，使得Topic的部分历史数据被及时清理，加快迁移速度。

介绍APM监控详情里消息监控中的KafkaProducer相关指标的名称、含义等信息。 KafkaProducer监控指标说明表 指标类别 指标 指标说明 数据类型 异常 exceptionstacktrace 异常产生的堆栈信息 String 异常 exceptiontype 异常类型 String 异常 exceptioncount 错误数 Int Topic (Topic，Kafka的Topic监控数据) id Clientid和IP信息 Enum Topic (Topic，Kafka的Topic监控数据) topic Kafka的Topic名称 Enum Topic (Topic，Kafka的Topic监控数据) kafkaproducerbyterate 每秒发送字节 Int Topic (Topic，Kafka的Topic监控数据) kafkaproducerrecorderrorrate 每秒错误数 Int Topic (Topic，Kafka的Topic监控数据) kafkaproducerrecordretryrate 每秒重试数 Int Topic (Topic，Kafka的Topic监控数据) kafkaproducerrecordsendrate 每秒发送数 Int Topic (Topic，Kafka的Topic监控数据) kafkaproducerrecordsendtotal 总发送次数 Int Topic (Topic，Kafka的Topic监控数据) kafkaproducerbytetotal 总发送字节数 Int 汇总（Total，KafkaProducer汇总信息统计） kafkaproducerrecordsendtotal 总发送次数 Int 汇总（Total，KafkaProducer汇总信息统计） kafkaproducerbytetotal 总发送字节数 Int 发送方法（DoSendMethod，发送消息方法监控） topic Kafka的Topic名称 Enum 发送方法（DoSendMethod，发送消息方法监控） errorCount 错误数 Int 发送方法（DoSendMethod，发送消息方法监控） invokeCount 调用次数 Int 发送方法（DoSendMethod，发送消息方法监控） maxTime 最慢时延 Int 发送方法（DoSendMethod，发送消息方法监控） ms0To10Count 010ms次数 Int 发送方法（DoSendMethod，发送消息方法监控） ms10To100Count 10100ms次数 Int 发送方法（DoSendMethod，发送消息方法监控） ms100To500Count 100500ms次数 Int 发送方法（DoSendMethod，发送消息方法监控） ms500To1000Count 5001000ms次数 Int 发送方法（DoSendMethod，发送消息方法监控） ms1000To10000Count 110s次数 Int 发送方法（DoSendMethod，发送消息方法监控） msMorethan10000Count 10s以上次数 Int 发送方法（DoSendMethod，发送消息方法监控） totalTime 调用总耗时 Int

消息堆积对业务有什么影响？ 为了保证服务的稳定可靠，分布式消息服务RabbitMQ版采用了默认的40%高水位配置。当内存占用率达到40%高水位后，会触发流控，生产者流程会被阻塞。消息堆积可能造成内存高水位，为了避免高水位的产生，请及时消费积压在队列中的消息。 消费的最长保留时间是多久？ 一般情况下消息如果未被消费会一直保留，只有被消费后，才会被删除。但是如果设置了过期时间（TTL），则以TTL时间为准。

介绍分布式消息服务Kafka与天翼云其他服务关系。 虚拟私有云(CTVPC ，Virtual Private Cloud) 虚拟私有云为分布式消息服务Kafka提供一个逻辑隔离的区域，构建一个安全可靠、 可配置和管理的虚拟网络环境。更多信息请参见虚拟私有云。 弹性云主机（CTECS，Elastic Cloud Server） 分布式消息服务Kafka订购后，默认按照用户选择的实例规格开通弹性云主机，云主机由 CPU、内存、镜像、云硬盘组成，同时结合VPC、安全组、数据多副本保存等能力，打造一个既高效又可靠安全的计算环境，确保分布式消息服务Kafka持久稳定运行。更多信息请参见弹性云主机。 云硬盘（CTEVS，Elastic Volume Service） 分布式消息服务Kafka订购后，默认按照用户选择的存储大小开通云硬盘。云硬盘是一种可弹性扩展的块存储设备，可以为分布式消息服务Kafka提供高性能、高可靠的块存储服务。更多信息请参见云硬盘。 弹性IP（Elastic IP，EIP） 弹性IP是可以独立申请的公网 IP 地址，包括公网IP地址与公网出口带宽服务。可以与分布式消息服务Kafka动态绑定和解绑，实现云资源的互联网访问。针对需要公网访问分布式消息服务Kafka实例的需求，用户可开通弹性IP后，在Kafka实例开通页面进行绑定。更多信息请参见弹性IP。

本文主要介绍监控告警问题。 云监控无法展示Kafka监控数据 Topic监控数据无法展示，可能原因如下： Topic名称开头包含特殊字符，例如下划线“”、号“”。 Kafka实例中没有创建Topic。 解决方法如下： 删除带特殊字符的Topic。 创建Topic。 消费组监控数据无法展示，可能原因如下： 消费组名称开头包含特殊字符，例如下划线“”、号“”。 此消费组从未有消费者连接。 Kafka监控显示消息堆积数跟实例里的消息数不一致？ 问题现象 ：监控显示消息堆积数为8.1亿+，Kafka控制台显示实例中6个Topic的消息数总和为1亿+，两者不一致。 问题结论 ：两者统计方式不同，Kafka控制台显示的消息数为实例中未消费的消息个数，而监控显示的消息堆积数Topic中的消息积压数消费组数。 Kafka的消费组删除了，怎么监控页面还可以看到这个消费组？ 监控数据是每分钟进行采集上报，上报的数据经过整理后才会显示在监控页面上，此过程大约需要几分钟到十几分钟，建议您在删除消费组后，过一段时间再去监控页面查看。

本文主要介绍消费组问题。 Kafka实例是否需要创建消费组、生产者和消费者？ 不需要单独创建消费组、生产者和消费者，在使用时自动生成，实例创建后，直接使用即可。 连接Kafka实例后，生产消息和消费消息，请参考向Kafka实例生产消息和消费消息。 如果消息组中没有在线的消费者（如empty状态），是否14天后会自动被删除？ 消息组中没有在线的消费者（如empty状态），14天后是否会自动被删除与offsets.retention.minutes参数有关： 2020年6月16日前创建的实例，offsets.retention.minutes默认为2147483646分钟，约1491308天，消费组14天后不会被删除。 2020年6月16日以及之后创建的实例，offsets.retention.minutes默认为20160分钟，即14天，消费组14天后会自动被删除。 原因如下： Kafka通过offsets.retention.minutes参数控制消费组中offsets保留时间，在此时间内如果没有提交offset，offsets将会被删除。Kafka判定消息组中没有在线的消费者（如empty状态），且没有offsets时，将会删除此消费组。 客户端删除消费组后，在Kafka Manager中仍可以看到此消费组？ 客户端删除消费组后，此消费组已经被删除了。在Kafka Manager中仍可以看到此消费组，是因为Kafka Manager存在缓存。 通过以下任意一种方法解决此问题： 重启Kafka Manager。 Kafka Manager只显示14天内有消费记录的消费组，如果您不想重启Kafka Manager，可以等待14天后Kafka Manager自动清除此消费组。

本节介绍Kafka消费客户端频繁出现Rebalance原因及解决方案 可能原因 可能是Kafka客户端版本过低或者Consumer没有独立线程维持心跳。 v0.10.2之前版本的客户端：Consumer没有独立线程维持心跳，而是把心跳维持与poll接口耦合在一起。其结果就是，如果用户消费出现卡顿，就会导致Consumer心跳超时，引发Rebalance。 v0.10.2及之后版本的客户端：如果消费时间过慢，超过一定时间（max.poll.interval.ms设置的值，默认5分钟）未进行poll拉取消息，则会导致客户端主动离开队列，而引发Rebalance。 解决方案 首先您需要了解以下几点信息： session.timeout.ms：心跳超时时间（可以由客户端自行设置）。 max.poll.records：每次poll返回的最大消息数量。 v0.10.2之前版本的客户端：心跳是通过poll接口来实现的，没有内置的独立线程。 v0.10.2及之后版本的客户端：为了防止客户端长时间不进行消费，Kafka客户端在v0.10.2及之后的版本中引入了max.poll.interval.ms配置参数。 （1）参考以下说明调整参数值： session.timeout.ms：v0.10.2之前的版本可适当提高该参数值，需要大于消费一批数据的时间，但不要超过30s，建议设置为25s；而v0.10.2及其之后的版本，保持默认值10s即可。 max.poll.records：降低该参数值，建议远远小于 的积。 max.poll.interval.ms：该值要大于 / ( )的值。 （2）尽量提高客户端的消费速度，消费逻辑另起线程进行处理。 （3）减少Group订阅Topic的数量，一个Group订阅的Topic最好不要超过5个，建议一个Group只订阅一个Topic。 （4）将客户端升级至0.10.2以上版本。

实施步骤 （1）购买一台弹性云主机，确保弹性云主机与源集群、目标集群网络互通。 （2）登录弹性云主机，安装Java JDK，并配置JAVAHOME与PATH环境变量。其中“/usr/local/java/jdk1.8.0161”为JDK的安装路径，请根据实际情况修改。 exportJAVAHOME/usr/local/java/jdk1.8.0161 exportPATH$JAVAHOME/bin:$PATH （3）下载安装kafka 下载页面： （4）进入kafka安装目录，修改“config/connectmirrormaker.properties”配置文件，在配置文件中指定源集群和目标集群的IP地址和端口以及其他配置。 （5）在kafka安装目录下，启动MirrorMaker，进行数据同步。 ./bin/connectmirrormaker.sh config/connectmirrormaker.properties 验证数据是否同步 要验证MirrorMaker是否成功同步数据，可以采取以下几种方法： 1. 检查目标集群的主题和分区： 在目标集群上使用Kafka命令行工具或管理工具，查看MirrorMaker复制的主题和分区是否存在。确保目标集群上有与源集群相同的主题和分区。 2. 检查消息偏移量： 使用Kafka消费者API或命令行工具，从目标集群中消费复制的消息。验证消息的偏移量是否与源集群中的消息偏移量一致。如果偏移量相同，表示数据同步成功。 3. 检查消息内容： 从目标集群中消费复制的消息，并与源集群中的消息进行比较。验证消息内容是否一致。可以使用Kafka消费者API或命令行工具来消费消息，并进行比较。 4. 监控MirrorMaker的指标： 在MirrorMaker节点上启用监控，收集关于复制进度、延迟和吞吐量等指标的数据。通过监控指标，可以了解MirrorMaker的工作状态和性能表现，进一步验证数据同步的情况。 5. 进行端到端测试： 在源集群中发送一些测试消息，并在目标集群中验证这些消息是否被成功复制。可以使用Kafka生产者API或命令行工具发送测试消息，并使用Kafka消费者API或命令行工具从目标集群中消费并验证消息。 通过以上方法，可以验证MirrorMaker是否成功同步数据，并确保数据在源集群和目标集群之间的一致性。根据实际需求，可以选择适合的验证方法或组合多种方法来进行验证。

本节介绍如何配置Kafka之间的数据复制。 本章节介绍如何创建Kafka数据复制的Smart Connect任务，通过Smart Connect任务可以在两个Kafka实例之间，实现数据的单向或双向复制。 源Kafka实例中的数据会实时同步到目标Kafka实例中。 约束与限制 一个实例最多创建18个Smart Connect任务。 使用Kafka数据复制时，两个Kafka实例间只能通过内网连接。如果两个Kafka实例处于不同的VPC中，请先打通网络。 Smart Connect任务创建成功后，不支持修改任务参数。 确保目标Kafka实例Topic的“批处理消息最大值”大于等于524288字节，否则会导致数据无法同步。如果目标Kafka实例没有创建Topic，在数据同步时会自动创建Topic，此Topic的“批处理消息最大值”和源Kafka实例Topic相同，此时需要确保源Kafka实例Topic的“批处理消息最大值”大于等于524288字节。修改“批处理消息最大值”的方法请参考修改Kafka Topic配置。 前提条件 已开启Smart Connect。 已创建Kafka实例，且实例状态为“运行中”。 配置Kafka间的数据复制 1、登录管理控制台。 2、在管理控制台左上角单击，选择Kafka实例所在的区域。 3、在管理控制台左上角单击，选择“应用服务 > 分布式消息服务 Kafka”，进入分布式消息服务Kafka专享版页面。 4、单击Kafka实例名称，进入实例详情页面。 5、在左侧导航栏单击“Smart Connect”，进入Smart Connect任务列表页面。 6、单击“创建Smart Connect任务”，进入“创建smart connect”页面。 7、在“connect任务名称”中，输入Smart Connect任务的名称，用于区分不同的Smart Connect任务。任务名称需要符合命名规则：长度为4~64个字符，只能由英文字母、数字、中划线、下划线组成。 8、在“预置类型”中，选择“Kafka数据复制”。 9、在“当前kafka”区域，设置实例别名。实例别名需要符合命名规则：长度为1~20个字符，只能由英文字母、数字、中划线、下划线组成。 实例别名用于以下两个场景中： 开启“重命名Topic”，且“同步方式”为“推送”/“双向”时，当前Kafka实例的别名作为前缀添加到对端Kafka实例的Topic名称前，形成Topic新的名称。例如当前Kafka实例别名为A，对端Kafka实例的Topic名称为test，重命名后的Topic为A.test。 Kafka数据复制的Smart Connect任务创建成功后，当前Kafka实例会自动创建“mm2offsetsyncs. 对端Kafka实例别名 .internal”的Topic。如果Smart Connect任务开启了“同步消费进度”功能，且“同步方式”为“拉取”/“双向”时，当前Kafka实例还会自动创建“ 对端Kafka实例别名 .checkpoints.internal”的Topic。这两个Topic用于存储内部数据，如果删除，会导致同步数据失败。 10、在“对端kafka”区域，设置以下参数。 表11 对端Kafka实例参数说明 参数 参数说明 实例别名 设置实例别名，实例别名需要符合命名规则：长度为1~20个字符，只能由英文字母、数字、中划线、下划线组成。 实例别名用于以下两个场景中： 开启“重命名Topic”，且“同步方式”为“拉取”/“双向”时，对端Kafka实例的别名作为前缀添加到当前Kafka实例的Topic名称前，形成Topic新的名称。例如对端Kafka实例别名为B，当前Kafka实例的Topic名称为test01，重命名后的Topic为B.test01。 Kafka数据复制的Smart Connect任务创建成功后，如果Smart Connect任务开启了“同步消费进度”功能，且“同步方式”为“推送”/“双向”时，对端Kafka实例会自动创建“当前Kafka实例别名 .checkpoints.internal”的Topic。此Topic用于存储内部数据，如果删除，会导致同步数据失败。 配置类型 支持以下两种配置类型： Kafka地址：输入Kafka实例的连接信息。对端Kafka实例和当前Kafka实例处于不同的VPC下时，请选择此配置类型。 实例名称：选择已创建的Kafka实例。对端Kafka实例和当前Kafka实例处于相同的VPC下时，建议选择此配置类型。 实例名称 “配置类型”为“实例名称”，且对端Kafka实例和当前Kafka实例处于相同的VPC下时，需要设置。 在下拉列表中，选择已创建的Kafka实例。 Kafka地址 “配置类型”为“Kafka地址”时，需要设置。 输入Kafka实例的连接地址和端口号。 使用Kafka数据复制时，两个Kafka实例间只能通过内网连接。如果两个Kafka实例处于不同的VPC中，请先打通网络。 认证方式 支持以下认证方式： SASLSSL：表示实例已开启SASLSSL认证，客户端连接Kafka实例时采用SASL认证，数据通过SSL证书进行加密传输。 SASLPLAINTEXT：表示实例开启SASLPLAINTEXT认证，客户端连接Kafka实例时采用SASL认证，数据通过明文传输。 PLAINTEXT：表示实例未开启认证。 认证机制 “认证方式”为“SASLSSL”/“SASLPLAINTEXT”时，需要设置。 SCRAMSHA512：采用哈希算法对用户名与密码生成凭证，进行身份校验的安全认证机制，比PLAIN机制安全性更高。 PLAIN：一种简单的用户名密码校验机制。 用户名 “认证方式”为“SASLSSL”/“SASLPLAINTEXT”时，需要设置。 首次开启密文接入时设置的用户名，或者创建用户时设置的用户名。 密码 “认证方式”为“SASLSSL”/“SASLPLAINTEXT”时，需要设置。 首次开启密文接入时设置的密码，或者创建用户时设置的密码。 Smart Connect任务创建成功后，如果您修改了对端实例的认证方式、认证机制或者密码，会导致同步任务失败。 您需要删除当前Smart Connect任务，然后重新创建新的Smart Connect任务。 11、在“规则配置”区域，设置以下参数。 表12 复制数据规则参数说明 参数 参数说明 同步方式 支持以下三种同步方式： 拉取：把对端Kafka实例数据复制到当前Kafka实例中。 推送：把当前Kafka实例数据复制到对端Kafka实例中。 双向：两端Kafka实例数据进行双向复制。 Topics 设置需要进行数据复制的Topic。 正则表达式：通过正则表达式匹配Topic。 输入/选择：输入Topic名称，如果需要输入多个Topic名称，先输入一个Topic名称，按“Enter”，然后再输入下一个，按“Enter”，依次输入。或者在下拉列表中，选择Topic。最多输入/选择20个Topic。 Topic名称以internal结尾时（例如：topic.internal），此Topic的数据不会被同步。 任务数 数据复制的任务数。默认值为2，建议保持默认值。 如果“同步方式”为“双向”，实际任务数设置的任务数2。 重命名 Topic 在目标Topic名称前添加源端Kafka实例的别名，形成目标Topic新的名称。例如源端实例别名为A，目标Topic名称为test，重命名后的目标Topic为A.test。 两端实例数据双向复制时，开启“重命名Topic”，防止循环复制。 添加来源header 目标Topic接收复制的消息，此消息header中包含消息来源。 两端实例数据双向复制时，默认开启“添加来源header”，防止循环复制。 同步消费进度 开启“同步消费进度”后，将消费者消费进度同步到目标Kafka实例。 开启“同步消费进度”后，您需要注意以下几点： 源端Kafka实例和目标端Kafka实例不能同时消费，否则会导致同步的消费进度异常。 同步消费进度的频率为每分钟一次，因此会导致目标端的消费进度可能会略小于源端，造成部分消息被重复消费，所以需要消费者客户端业务逻辑兼容重复消费的场景。 从源端同步的offset与目标端的offset并非一致关系，而是映射关系，如果消费进度由消费者客户端自行维护，消费者客户端从消费源端Kafka实例变为消费目标端Kafka实例后，不向目标端Kafka实例获取消费进度，可能会导致offset错误或消费进度重置。 副本数 在对端实例中自动创建Topic时，指定Topic的副本数，此参数值不能超过对端实例的代理数。 如果对端实例中设置了“default.replication.factor”，此参数的优先级高于“default.replication.factor”。 启动偏移量 支持两种偏移量： 最早：最小偏移量，即获取最早的数据。 最新：最大偏移量，即获取最新的数据。 压缩算法 复制消息所使用的压缩算法。 topic映射 通过Topic映射，您可以自定义目标端Topic名称。 最多新增20个Topic映射。不能同时设置“重命名Topic”和“topic映射”。 配置复制数据规则时需要注意以下几点： 创建双向数据复制任务时，为了防止循环复制，控制台限定必须开启“重命名Topic”或者“添加来源header”。如果您在两个实例间，对同一个Topic分别创建拉取和推送的任务，即形成双向数据复制，且两个任务都未开启“重命名Topic”和“添加来源header”，此时会导致数据循环复制。 如果创建两个或以上配置完全相同的任务，即重复创建任务，且任务已开启“同步消费进度”，此时会导致数据重复复制，且目标Topic消费进度异常。 12、（可选）在页面右下角单击“开始检测”，测试两端Kafka实例的连通性。 显示“连通性检测成功”时，表示两端Kafka实例可以正常连接。 13、单击“立即创建”，跳转到Smart Connect任务列表页面，页面右上角显示“创建xxx任务成功”。 Kafka数据复制的Smart Connect任务创建成功后，Kafka会自动创建以下Topic。 当前Kafka实例会自动创建“mm2offsetsyncs. 对端Kafka实例别名 .internal”的Topic。如果Smart Connect任务开启了“同步消费进度”功能，且“同步方式”为“拉取”/“双向”时，当前Kafka实例还会自动创建“ 对端Kafka实例别名 .checkpoints.internal”的Topic。这两个Topic用于存储内部数据，如果删除，会导致同步数据失败。 如果Smart Connect任务开启了“同步消费进度”功能，“同步方式”为“推送”/“双向”时，对端Kafka实例会自动创建“ 当前Kafka实例别名 .checkpoints.internal”的Topic。此Topic用于存储内部数据，如果删除，会导致同步数据失败。 结束

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