本节主要介绍分布式消息服务Kafka的创建Topic说明 背景信息 Kafka的主题（Topic）是对消息进行分类和组织的概念。主题是Kafka中最基本的逻辑单元，用于区分不同类型的消息。 每个主题可以有多个分区（Partition），每个分区都是一个有序的消息日志。分区允许消息在集群中进行并行处理，提高了吞吐量和并发性能。 主题的特点包括： 逻辑分类：主题可以根据业务需求或数据类型进行逻辑分类，例如订单主题、日志主题、用户行为主题等。 多分区：主题可以被分成多个分区，每个分区都有自己的消息顺序。分区可以在不同的节点上进行存储和处理，实现了水平扩展和负载均衡。 副本复制：每个分区可以有多个副本（Replica），副本用于提供高可用性和容错性。副本可以分布在不同的节点上，当节点故障时可以自动进行故障转移。 持久化存储：主题中的消息被持久化存储在磁盘上，确保消息的可靠性和持久性。消费者可以随时消费主题中的历史消息，而不仅仅是最新的消息。 通过使用主题，Kafka能够有效地组织和管理大规模的消息流。主题的分区和副本机制提供了高可用性和容错性，使得Kafka成为处理大规模实时数据流的理想选择。 前提条件 已购买并部署分布式消息服务Kafka 操作步骤 创建Topic （1）登录管理控制台。 （2）进入Kafka管理控制台。 （3）在实例列表页在操作列，目标实例行点击“管理”。 （4）点击“Topic管理”后、点击“创建Topic”。 （5）点击“创建Topic”后，输入Topic名称、分区数等参数。 （6）创建好Topic后，可以点击“更多”按钮中的“生产消息”来测试其是否正常。

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

1、 天翼云官网点击控制中心，选择产品分布式消息服务RocketMQ。 2、 登录分布式消息服务RocketMQ控制台，点击左上角地域选择对应资源池。 3、 进入实例列表，点击【管理】按钮进入管理菜单。 4、 进入用户管理菜单，点击【创建用户】按钮 5、 在弹出的创建用户页面，填写如下字段信息，点击确认保存。 1、用户ID：必填，764位字符,只能输入大小写字母，下划线，数字。 2、密钥：必填，8～26个字符，必须同时包含大写字母、小写字母、数字和英文格式特殊符号(@%^+!$.)中的至少三种类型且不能包含空格。 3、IP白名单：非必填。 4、默认Topic权限 DENY：拒绝所有操作。 PUB：仅允许发布消息到Topic。 PUBSUB：允许发布+订阅。 SUB：仅允许订阅Topic消息。 5、默认消费组权限 DENY：拒绝所有操作。 SUB：允许订阅Topic并关联该消费组。 6、备注：按需备注该用户的说明。 7、管理员：有Topic和消费组所有权限。

本节介绍如何管理Smart Connect任务。 在创建Smart Connect任务后，可以在控制台对任务进行管理操作，包括查看任务、删除任务、启动/暂停任务、重启任务等。 前提条件 已创建Smart Connect任务。 查看Smart Connect任务 1、登录管理控制台。 2、在管理控制台左上角单击，选择Kafka实例所在的区域。 3、在管理控制台左上角单击，选择“应用服务 > 分布式消息服务 Kafka”，进入分布式消息服务Kafka专享版页面。 4、单击Kafka实例名称，进入实例详情页面。 5、在左侧导航栏单击“Smart Connect”，进入Smart Connect任务列表页面。 6、单击Smart Connect任务名称，进入Smart Connect任务详情页面。 7、查看Smart Connect任务的基本信息、源端信息和目标端信息。 说明 如果Smart Connect任务没有配置源端信息/目标端信息，在Smart Connect任务详情页面不显示源端信息/目标端信息。 结束 删除Smart Connect任务 1、登录管理控制台。 2、在管理控制台左上角单击，选择Kafka实例所在的区域。 3、在管理控制台左上角单击，选择“应用服务 > 分布式消息服务 Kafka”，进入分布式消息服务Kafka专享版页面。 4、单击Kafka实例名称，进入实例详情页面。 5、在左侧导航栏单击“Smart Connect”，进入Smart Connect任务列表页面。 6、在待删除的Smart Connect任务所在行，单击“删除”，弹出“删除任务”对话框。 7、单击“确定”，完成Smart Connect任务的删除。 结束

本章节将为您介绍分布式消息服务Kafka入门的基本流程，主要包括环境准备、购买实例、创建Topic以及编译运行Demo Java工程，帮助您快速上手Kafka。 操作流程 操作流程如下： 1. 环境准备 创建Kafka实例先要准备好虚拟私有云、子网和安全组，可选弹性公网IP。 2. 创建实例 在订购分布式消息Kafka填写和确认实例名称、计费模式等信息，确认费用后点击下一步，等待开通流程结果通知成功后完成创建实例。 3. 创建Topic 开通实例后，在控制台相关页面按照指引创建主题，用于发送消息。点击“生产消息”来测试Topic是否正常。 4. 编译运行Demo Java工程 以上工作完成后，在客户端应用编译工程进行生产消费，包括引入依赖，生产消息和消费消息。

本文主要介绍如何在控制台删除消息。 操作场景 说明 消息删除后无法恢复，请谨慎操作。 前提条件 删除消息前，请先在客户端中设置“auto.offset.reset”参数。“auto.offset.reset”用来指定当Kafka中没有初始偏移量或者当前偏移量不存在（例如当前偏移量已被删除）时，消费者的消费策略。取值如下： latest：偏移量自动被重置到最晚偏移量。 earliest：偏移量自动被重置到最早偏移量。 none：向消费者抛出异常。 操作步骤 1. 登录管理控制台。 2. 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 3. 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 4. 单击Kafka实例名称，进入实例详情页面。 5. 在左侧导航栏选择“Topic管理”，进入Topic列表页面。 6. 在待删除消息的Topic所在行，单击“消息删除”，弹出“消息删除”对话框。 7. 设置消息删除参数，如下表所示。 图消息删除 表 消息删除参数说明 参数 说明 分区 选择消息所在的分区编号。 偏移量 输入偏移量，此偏移量之前的数据将被删除。 说明 如果“偏移量”设置为“1”，表示删除分区中所有的消息。 如果您输入的偏移量不在指定分区的最早偏移量和最晚偏移量之间，消息将不会被删除。 如果需要删除多个分区的消息，单击“添加分区”，设置需要删除消息的分区和偏移量。每次最多可选择10个分区。 8. 单击“确定”，弹出“清理结果”对话框，单击“确定”，完成消息的删除。 图 清理结果

产品服务协议请查看： 天翼云分布式消息服务协议

本节主要介绍分布式消息服务Kafka产品的API中资源池ID列表。 分布式消息服务Kafka的资源池列表。 资源池 资源池ID(Region ID) 华东1 bb9fdb42056f11eda1610242ac110002 华北2 200000001852 西南1 200000002368 华南2 200000002530 上海36 200000001790 青岛20 200000001703 长沙42 200000002401 南昌5 200000002527 武汉41 200000001781 杭州7 200000003329 太原4 200000002689 郑州5 200000002586 西安7 200000001851 芜湖4 200000003327 呼和浩特3 200000003573 乌鲁木齐7 200000004098 庆阳2 200000003664 西南2贵州 200000002927

本文介绍Kafka消费端从服务端拉取不到消息或拉取消息缓慢原因及解决方案 问题现象 Topic中有消息并且Consumer未消费到最新的位置，出现消费端从服务端拉取不到消息或拉取消息缓慢的情况（特别是公网消费时）。 可能原因 消费流量达到网络带宽。 单个消息大小超过网络带宽。 Consumer每次拉取的消息量超过网络带宽。 说明 Consumer每次消息的拉取量受以下参数影响： max.poll.records：每次拉取的最多消息数。 fetch.max.bytes：每次拉取的最大总byte数。 max.partition.fetch.bytes：每个Partition每次拉取的最大总byte数。 解决方案 （1）登录分布式消息服务Kafka控制台查询消息。 如果能查询到消息，请继续尝试以下步骤。 （2）在实例详情页面，单击左侧导航栏的监控信息，查看消费流量是否已达到网络带宽。 如果消费流量已经达到网络带宽，您需要扩充网络带宽。 （3）检查Topic中是否存在单个消息的大小超过网络带宽。 如果存在单个消息的大小超过网络带宽，请提高网络带宽，或者减小单个消息的大小。 （4）检查Consumer每次拉取的消息量是否超过网络带宽。 说明 如果每次拉取的消息量超过网络带宽，您需要调整以下参数。 网络带宽>fetch.max.bytes 网络带宽>max.partition.fetch.bytes总订阅Partition数

本文为您介绍分布式消息服务MQTT各功能项使用限制。 功能信息项 限制值 说明 实例名称 长度4~40个字符，大小写字母开头，只能包含大小写字母、数字及分隔符() 实例名称不可重复。 Topic 长度 64 个字符 Topic 长度不得超过该限制，否则会导致无法发送或者订阅。 Topic 和 Client ID 可用字符 仅限数字 09，字母 az 或 AZ，和符号“”、“” Client ID 和MQTT 的 Topic 不允许使用“/”、“:”、“,”、“%”等其他非常规字符，否则可能导致无法连接和收发消息。 Client ID 长度 64 个字符 使用微消息队列 MQTT 收发消息时，Client ID 长度不得超过该限制，否则会导致连接被断开。 消息大小 64 KB 字节 消息负载不得超过该限制，否则消息会被丢弃。 单个设备订阅Topic 数量 30 个 每个客户端最多允许同时订阅 30 个 Topic，超过该限制会导致无法新增新的订阅关系。 每个持久化会话离线消息存储数量 1,000,条 服务端会限制每个持久化会话的离线消息数量。超过该限制后，服务端会从最早的消息开始清理。因此，请合理使用持久化订阅模式，以免产生很多无用的离线消息。

参数 说明 示例 规则名称 订阅规则名称 输入：aomkafkatest。 订阅内容 支持“指标”和“告警”。 选择：指标。 订阅目标类型 选择“自定义Kafka”或“分布式消息服务DMS”。 分布式消息服务DMS 实例 选择DMS实例，如没有DMS实例，请单击“创建DMS实例”，创建DMS实例。 kafkaaom7160

本章节介绍了如何创建分布式消息服务RocketMQ实例的消费组。 操作场景 消费组，即一组具有相同行为的消费者集合，用于管理和维护消息的消费状态。 前提条件 已购买RocketMQ实例。 操作步骤 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，单击“消费组管理”，进入“消费组管理”页面。 4. 单击“创建消费组”，弹出“创建消费组”页面。 图1 创建消费组 5. 参考表1，填写消费组名称和配置信息。 表1 消费组参数说明 参数 说明 消费组名称 您可以自定义消费组名称，用于区别不同的消费行为。 创建消费组后不能修改名称。 关联代理 选择在某个或者某几个代理上创建消费组。 创建消费组后不能修改关联代理。 最大重试次数 设置普通消费的最大重试次数。如果是顺序消费，可通过消费者的setMaxReconsumeTimes函数配置。 广播消费 是否设置为广播消费。 若开启广播消费，每条消息都会被消费组内的所有消费者消费一次。否则，每条消息只会被消费组内的某一消费者消费。 6. 配置完成后，单击“确定”，完成消费组的创建。

本文主要介绍如何转储日志数据至分布式消息服务Kafka。 云日志服务采集到数据后，支持将数据转储至对天翼云分布式消息服务kafka，用于实时流计算等场景。 前提条件 已创建日志项目和日志单元 已开通天翼云分布式消息服务kafka 已采集到日志 创建日志转储任务 1. 登录云日志服务控制台。 2. 左侧点击【日志转储】菜单，进入日志转储页面。 3. 在页面上方选择“kafka”。 4. 点击【创建转储任务】。 5. 在创建转储任务功能面板，配置如下参数，然后单击确定，完成转储任务创建。 参数 说明 日志项目名称 选择源日志所在的日志项目。 日志单元名称 选择源日志所在的日志单元。 转储任务名称 转储任务的名称。 kafka实例 选择转出目标的kafka实例。 Topic 选择当前kafka的topic。（当前仅支持转储至公有 Topic） 转储时间范围 1、某时间开始：指定转储任务的开始时间，从该时间点开始进行数据转储，直到您手动停止转储任务。(开始时间不得早于当前时间，早于当前时间则以当前时间为准) 2、特定时间范围：指定转储任务的起止时间，转储任务将执行到您指定的结束时间后自动停止。(开始时间不得早于当前时间，早于当前时间则以当前时间为准)

本文主要介绍重启实例。 操作场景 分布式消息服务Kafka管理控制台支持重启运行中的Kafka实例，且可实现批量重启Kafka实例。 说明 在Kafka实例重启过程中，客户端的生产与消费消息等请求会被拒绝。 前提条件 只有当Kafka实例处于“运行中”或“故障”状态，才能执行重启操作。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 通过以下任意一种方法，重启Kafka实例。 勾选Kafka实例名称左侧的方框，可选一个或多个，单击信息栏左上侧的“重启”。 在待重启Kafka实例所在行，单击“重启”。 单击Kafka实例名称，进入实例详情页面。单击右上角的“重启”。 步骤 5 在“重启实例”对话框中，单击“是”，完成重启Kafka实例。 重启Kafka实例大约需要3到15分钟。Kafka实例重启成功后，实例状态切换为“运行中”。 说明 重启Kafka实例只会重启实例进程，不会重启实例所在虚拟机。

本章节介绍了如何删除分布式消息服务RocketMQ实例的用户。 操作场景 用户无需继续使用，需要清理时，参考本章节删除用户。 须知 ： 删除用户后会删除授权关系，原来使用该用户认证的请求将无法访问。 操作步骤 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，单击“用户管理”，进入“用户管理”页面。 4. 在待删除的用户所在行，单击“删除”，弹出“删除用户”对话框。 5. 单击“是”，完成用户的删除。

本文介绍节点重启后消费者如何重连，以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); } } }

数据源类型 简介 数据仓库服务（DWS） DWS是基于Sharednothing分布式架构，具备MPP大规模并行处理引擎，兼容标准ANSI SQL 99和SQL 2003，同时兼容PostgreSQL/Oracle数据库生态，为各行业PB级海量大数据分析提供有竞争力的解决方案。 数据湖探索（DLI） DLI是完全兼容Apache Spark和Apache Flink生态，实现批流一体的Serverless大数据计算分析服务。DLI支持多模引擎，企业仅需使用SQL或程序就可轻松完成异构数据源的批处理、流处理、内存计算、机器学习等，挖掘和探索数据价值。 MapReduce服务（MRS HBase） HBase是一个开源的、面向列（ColumnOriented）、适合存储海量非结构化数据或半结构化数据的、具备高可靠性、高性能、可灵活扩展伸缩的、支持实时数据读写的分布式存储系统。 使用MRS HBase可实现海量数据存储，并实现毫秒级数据查询。选择MRS HBase可以实现物流数据毫秒级实时入库更新，并支持百万级时序数据查询分析。 MapReduce服务（MRS Hive） Hive是一种可以存储、查询和分析存储在Hadoop中的大规模数据的机制。Hive定义了简单的类SQL查询语言，称为HiveQL，它允许熟悉SQL的用户查询数据。 使用MRS Hive可实现TB/PB级的数据分析，快速将线下Hadoop大数据平台（CDH、HDP等）迁移上云，业务迁移 “0”中断，业务代码 “0”改动。 MapReduce服务（MRS Kafka） MapReduce服务可提供专属MRS Kafka集群。Kafka是一个分布式的、分区的、多副本的消息发布订阅系统，它提供了类似于JMS的特性，但在设计上完全不同，它具有消息持久化、高吞吐、分布式、多客户端支持、实时等特性，适用于离线和在线的消息消费，如常规的消息收集、网站活性跟踪、聚合统计系统运营数据（监控数据）、日志收集等大量数据的互联网服务的数据收集场景。 MapReduce服务（MRS Ranger） Ranger提供一个集中式安全管理框架，提供统一授权和统一审计能力。它可以对整个Hadoop生态中如HDFS、Hive、HBase、Kafka、Storm等进行细粒度的数据访问控制。用户可以利用Ranger提供的前端WebUI控制台通过配置相关策略来控制用户对这些组件的访问权限。 MySQL MySQL是目前最受欢迎的开源数据库之一，其性能卓越，架构成熟稳定，支持流行应用程序，适用于多领域多行业，支持各种WEB应用，成本低，中小企业首选。 MapReduce服务（MRS Spark） Spark是一个开源的，并行数据处理框架，能够帮助用户简单的开发快速、统一的大数据应用，对数据进行协处理、流式处理、交互式分析等等。 Spark提供了一个快速的计算、写入以及交互式查询的框架。相比于Hadoop，Spark拥有明显的性能优势。Spark提供类似SQL的Spark SQL语言操作结构化数据。 云数据库RDS RDS是一种基于云计算平台的即开即用、稳定可靠、弹性伸缩、便捷管理的在线关系型数据库服务。 注意，DataArts Studio平台目前仅支持RDS中的MySQL和PostgreSQL数据库。 主机连接 通过主机连接，用户可以在DataArts Studio数据开发中连接到指定的主机，通过脚本开发和作业开发在主机上执行Shell或Python脚本。主机连接保存连接某个主机的连接信息，当主机的连接信息有变化时，只需在主机连接管理中编辑修改，而不需要到具体的脚本或作业中逐一修改。 MapReduce服务（MRS Presto） Presto是一个开源的用户交互式分析查询的SQL查询引擎，用于针对各种大小的数据源进行交互式分析查询。其主要应用于海量结构化数据/半结构化数据分析、海量多维数据聚合/报表、ETL、AdHoc查询等场景。 Presto允许查询的数据源包括Hadoop分布式文件系统（HDFS），Hive，HBase，Cassandra，关系数据库甚至专有数据存储。一个Presto查询可以组合不同数据源，执行跨数据源的数据分析。

本文主要介绍概述 概述 APM Agent会周期性采集一些性能指标数据，用来衡量应用的总体健康状况。可以采集JVM、GC、服务调用、异常、外部调用、数据库访问以及其他中间件的指标调用等数据，帮助用户全面掌握应用的运行情况。 APM对指标数据的采集有严格的定义，每一种采集的数据类型对应一个采集器，比如采集java应用的JVM数据，那么对应有JVM采集器，一个采集器会采集多个指标集的数据。 采集器被部署到环境后形成监控项，在数据采集的时候监控项决定了采集的数据结构和采集行为。 采集周期：监控项具有数据采集器的周期属性。当前数据采集周期为一分钟，不支持用户调整。 监控项状态：默认为enable状态，用户可以将监控项设置为disable状态，这样Agent就不会拦截该指标数据，也不会上报数据。 采集状态：采集实例和监控项会有一个采集状态信息。如果出现采集错误，可以通过采集状态查看。常见错误是主键太多，导致客户端数据汇聚异常。 监控项类型 Agent会自动发现系统采集的插件类型，并且将采集器实例化，形成监控项。监控项是实例化在一个环境上的。 由于采集器种类较多，会导致用户区分困难。系统后台会定义一些类型，每种采集器都会归到一种类型下，这样方便用户查看数据。 根据采集器的作用可以将监控项分为以下几种类型： 接口调用：是指外部服务调用当前应用的监控类型。 基础监控：是用来监控系统性能的基础监控指标的监控类型。 异常：用来监控应用的异常信息。 外部调用：是指当前应用调用外部服务的监控类型。 数据库：是对数据库的访问进行监控。 缓存：是对Redis等缓存系统的监控，会采集指令级别的细粒度的指标数据。 web容器：是对tomcat等web容器的监控，一般会采集系统总的处理线程数，busy线程数，连接数等；用于衡量系统总的容量。 消息队列：是对kafka、RabbitMq等消息系统的监控，包含发送端和接收端的监控。在接收端的处理函数，可以产生调用链信息。 通信协议：是对websocket等通信协议的监控。

事件源是事件的来源，负责将生产的事件发布到事件总线。 事件总线EventBridge支持以下事件源： 天翼云官方事件源 作为事件源接入时，只要开通相应的天翼云服务，就可以自动接入事件总线EventBridge。通过配置预定义的事件源与事件规则，实现从事件源发布事件到云服务专用总线，经过事件模式过滤后把事件路由到事件目标。 自定义事件源 作为事件源接入时，您可以配置自定义应用事件源，使用SDK接入事件总线EventBridge。通过创建自定义总线、配置事件规则，把自定义应用产生的事件发布到自定义总线，经过事件模式过滤后把事件路由到事件目标。 作为事件源接入时，您可以配置消息类事件提供方，把事件主动拉取到事件总线EventBridge。例如，当您配置事件提供方为分布式消息服务Kafka时，事件总线EventBridge会把分布式消息服务Kafka对应资源产生的消息主动拉取到自定义总线中，经过事件模式过滤后把事件路由到事件目标。

本文介绍如何应用事件总线EventBridge的事件流功能实现消息路由至函数计算。 前提条件 开通事件总线EventBridge并授权。 开通函数计算并创建对应函数。 开通分布式消息服务Kafka并创建最少两个主题。 背景信息 事件流作为更轻量、实时端到端的流式事件通道，提供轻量级的流式数据的过滤和转换的能力，在不同的数据仓库之间、数据处理程序之间、数据分析和处理系统之间进行数据同步。源端分布式消息服务Kafka生产的消息可以通过事件流这个通道被路由到目标端的函数计算，并触发函数。 步骤一：创建事件流 1. 登录事件总线EventBridge控制台。 2. 在左侧导航栏，单击事件流。 3. 在事件流页面，单击创建事件流。 4. 在创建事件流面板，设置任务名称和描述，配置以下参数，然后单击保存。 a.在Source（源）配置向导，选择数据提供方为分布式消息服务Kafka ，设置以下参数，然后单击下一步。 参数 说明 示例 Kafka实例 选择Kafka实例。 instancexxx Kafka Topic 选择Kafka topic。 topicxxx Group 选择消费组。 快速创建 消费位点 选择消费位点。 最新位点 b.在Filtering（过滤）配置向导，设置事件过滤规则，单击下一步。 c.在Sink（目标）配置向导，选择服务类型为函数计算，配置以下参数，单击保存。 参数 说明 示例 函数 选择函数。 funxxx 函数版本或别名 选择函数版本或别名。 版本/LATEST 执行方式 选择执行函数的方式：同步或异步。 同步 事件 选择调用到函数的事件内容，更多内容请参考事件内容转换。 完整事件 5. 创建事件流后，会有30秒~60秒的延迟时间，您可以在事件流页面的状态栏查看启动进度。

本文主要介绍 创建SASLSSL用户。 分布式消息服务Kafka支持对Topic进行ACL（Access Control List）权限管理，您可以在Topic中为不同的SASLSSL用户设置不同的权限，以达到SASLSSL用户之间的权限隔离。 本章节主要介绍Kafka实例开启SASLSSL后，如何创建SASLSSL用户。创建用户后，对Topic进行SASLSSL用户授权的操作，请参见设置Topic权限。 一个Kafka实例最多创建20个用户。 前提条件 创建Kafka实例时，已开启SASLSSL功能。 操作步骤 步骤 1 登录管理控制台。 步骤 2 在管理控制台右上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“企业中间件”“分布式消息服务”“Kafka专享版”，进入分布式消息服务Kafka专享版页面。 步骤 4 单击Kafka实例名称，进入实例详情页面。 步骤 5 在“用户管理”页签，单击“创建用户”。 步骤 6 在弹出的“创建用户”对话框中，设置用户名和密码，单击“确定”，完成用户的创建。 SASLSSL用户创建成功后，参考设置Topic权限为SASLSSL用户授权。

为了让更多客户享受到分布式消息Kafka产品服务，我们致力于为客户提供更具竞争力的价格，并持续提升服务质量。 分布式消息服务KafkaⅡ类资源池从7月15日开始启用新资费计价，老资费计算增强型同规格对比新费用降幅达19%27%，更多资费信息请参见产品资费。 规格名称 计费单位 实例包月标准价格 降幅 规格名称 计费单位 20240715前价格 新价格 降幅 kafka.4u8g.cluster 元/代理 1726 1260 27% kafka.8u16g.cluster 元/代理 3122 2430 22% kafka.16u32g.cluster 元/代理 5555 4500 19% 注意 通用型规格已经调整为白名单特性，如需了解相关资费请联系技术支持。 本次降价适用于Ⅱ类资源池。

本章节介绍了如何删除分布式消息服务RocketMQ实例的Topic。 操作场景 Topic无需继续使用，需要清理时，参考本章节删除Topic。 前提条件 仅在实例的“状态”为“运行中”时，才可以删除实例中的Topic。 操作步骤 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，单击“Topic管理”，进入“Topic管理”页面。 图1 Topic管理 4. 选择以下任意一种方法删除Topic。 在待删除的Topic所在行，单击“删除”。 单击待删除的Topic名称，进入Topic详情页面。在页面右上角，单击“删除”。 如需批量删除Topic，勾选待删除的Topic，单击“删除Topic”。 须知： 删除Topic会导致消息丢失，请谨慎操作。 5. 单击“是”，确认删除Topic。

本章节介绍了如何查询分布式消息服务RocketMQ实例的消费组信息。 操作场景 消费组创建成功后，查询消费组相关的配置和状态信息。 操作步骤 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，单击“消费组管理”，进入“消费组管理”页面。 4. 单击需要查询的消费组名称，进入消费组详情页面。 在详情页上方可以查看消费组名称、关联代理、最大重试次数和广播消费的配置。 在详情页下方可以查看到消费组消费的所有Topic以及消费进度等。 图1 消费组详情

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

步骤三：在控制台中发布事件 1. 登录事件总线EventBridge管理控制台，在左侧导航栏，单击事件总线。 2. 在事件总线页面，找到步骤一中选中的总线，在其右侧操作列单击发布事件。 3. 在发送事件面板，事件源下拉列表选择已创建的自定义事件源，在事件体代码框输入事件内容，然后单击确定 。事件包含的参数规范，请参见事件概述。 4. （可选项）亦可选择通过SDK发送事件，详见SDK概述。 注意 仅自定义事件总线支持通过管理控制台发布事件。 步骤四：结果验证 1. 登录分布式消息服务Kafka管理控制台。 2. 在左侧导航栏，单击实例列表，选择目标实例。 3. 在左侧导航栏点击消息查询页面，点击按位点查询，验证主题刚才收到的消息是否与步骤三发送的事件内容一致，详见图2。 图2 在分布式消息Kafka管理控制台消息查询页面查询消息

本章节介绍应用容灾多活的消息层配置。 概述 消息层配置主要包括消息队列 和消息同步，每个单元组可以单独进行配置，不同单元的消息队列之间通过消息路由服务进行同步。 应用容灾多活通过控制面与数据面协同，对指定Topic消息进行打标与过滤。当某个单元发生故障时，可以通过切流任务将故障单元的消息在其他正常单元进行回溯接管，从而保障消息不丢失。 图 消息层配置 前提条件 已开通消息层功能模块。 已完成系统架构配置以及路由规则配置。 已为应用开通分布式消息服务RocketMQ实例。 已为不同站点的实例创建同名的Topic和Group。 配置消息队列 1. 登录应用高可用服务控制台。 2. 单击左侧菜单栏应用容灾多活 ，在应用容灾多活菜单下单击数据双活/应用双活 ，进入数据双活/应用双活管理页面。 3. 在应用系统列表中找到需要配置的应用系统，单击应用系统名称 ，进入应用系统概览页面。 4. 单击左侧导航栏容灾配置 ，在容灾配置菜单下单击消息层配置 ，进入容灾配置消息层配置页面。 5. 单击消息队列 配置列表上方创建 按钮，弹出创建消息队列配置页面。 6. 如下表示例，填写配置信息，单击确定按钮，完成配置。 7. 如需修改或删除消息队列配置，可单击所选配置操作列修改 或删除按钮，根据界面提示进行操作。 表 消息队列配置 配置项 描述 示例 消息队列名称 消息队列自定义标识。 商城北京消息队列 站点 消息队列部署所在站点。 北京站点 产品类型 消息队列技术类型： RocketMQ RocketMQ 实例类型 消息队列产品类型。 云实例 消息队列实例 消息队列对应的云实例ID。 RocketMQ810

产品服务协议请查看： 天翼云分布式消息服务协议

Kafka触发器 Kafka触发器可以订阅天翼云提供的分布式消息队列Kafka实例，并根据消息触发关联的工作流，借此能力，使得工作流可以消费指定topic的消息，执行特定工作流处理逻辑。 注意事项 Kafka触发器订阅的Kafka实例必须和工作流在相同地域。 前提条件 已创建工作流。 已开通分布式消息Kafka实例（KAFKA引擎版），详情请参考创建分布式消息服务Kafka实例。 已创建Topic，创建GroupID（可选） 触发消息格式 Kafka触发器有两种消息格式：RawData和CloudEvent格式，可在触发器配置里选择。 CloudEvent格式： plaintext [ { "id": "eca534636baf4d568f86cbdb748208ed", "source": "ctyun.faas.trigger.kafka", "specversion": "1.0", "type": "kafka:topic:sendmessage", "datacontenttype": "application/json", "subject": "kafkatriggermqbjvsezbpdialtest:testforfaas", "time": "20250522T02:04:16Z", "data": { "headers": {}, "timestamp": 1747879456, "topic": "testforfaas", "partition": 0, "offset": 15280, "key": "", "value": "msg[9]: 154b2a0e2c3d4b03ae9ec225b5370c3b, ts20250522 02:04:16" } } ] 参数 类型 示例值 描述 id string eca534636baf4d568f86cbdb748208ed 事件ID。标识事件的唯一值。 source string ctyun.faas.trigger.kafka 事件源。Kafka触发器固定为ctyun.faas.trigger.kafka。 specversion string 1.0 CloudEvents协议版本。 type string kafka:topic:sendmessage 事件类型。 datacontenttype string application/json 参数data的内容形式。 subject string kafkatriggermqbjvsezbpdialtest:testforfaas 事件主体。格式为[SourceName]:[消息topic]。 time string 20250522T02:04:16Z 消息被触发的时间。 data object Kafka触发器独有消息格式，详细参见下文RawData描述。 RawData格式是CloudEvent格式的子集，只包含原始kafka消息的信息，消息结构相当于CloudEvent的data字段 plaintext [ { "offset" : 15280, "partition" : 0, "headers" : {}, "topic" : "testforfaas", "key" : "testkey", "timestamp" : 1747879456, "value" : "msg[9]: 154b2a0e2c3d4b03ae9ec225b5370c3b, ts20250522 02:04:16" } ] 参数 类型 示例值 描述 offset int 15280 消息偏移量。 partition int 0 分区信息。 headers map 消息携带的header。 topic string testforfaas topic的名称。 key string testkey 消息的key。 timestamp int 1747879456 Unix时间戳（秒）。 value string hello，kafka 消息的内容。

支持将分布式消息服务Kafka的重平衡日志接入LTS。重平衡日志记录了Rebalance的详情，包括Rebalance时间、原因和触发Rebalance的客户端等。重平衡日志存储在云日志服务中，由LTS提供查询功能。 接入操作说明 1. 登录分布式消息服务Kafka管理控制台。 2. 在实例列表中，点击目标Kafka实例名称。 3. 点击左侧菜单栏“智能运维”重平衡日志。点击【立即开启】按钮，即可将Kafka日志接入至云日志服务的grouprebalancelog日志项目中。 4. 等待若干分钟后，系统将自动生成可视化仪表大盘。 5. 点击页面上方“日志查询”页签，即可查询详细的重平衡日志信息。 6. 您也可以登录云日志服务控制台，在“grouprebalancelog”日志项目中，可查看所有已接入日志服务的Kafka重平衡日志的日志单元，并进行查询、分析或告警配置等操作。

本章节介绍了如何修改分布式消息服务RocketMQ实例的消费组参数。 操作场景 消费组创建成功后，您可以根据业务需要修改以下参数：最大重试次数、广播消费。 操作步骤 1. 登录分布式消息服务RocketMQ控制台。 2. 单击RocketMQ实例的名称，进入实例详情页面。 3. 在左侧导航栏，单击“消费组管理”，进入“消费组管理”页面。 4. 选择以下任意一种方法修改消费组参数。 在待修改的消费组所在行，单击“编辑”，弹出“编辑消费组”页面。 单击待修改的消费组名称，进入消费组详情页面。在页面右上角，单击“编辑”，弹出“编辑消费组”页面。 图1 编辑消费组 图2 编辑消费组 5. 修改如表1所示配置信息。 表1 消费组参数说明 参数 说明 最大重试次数 设置消费失败的最大重试次数。 广播消费 是否设置为广播消费。 若开启广播消费，每条消息都会被消费组内的所有消费者消费一次。否则，每条消息只会被消费组内的某一消费者消费。 6. 修改完成后，单击“确定”。

删除创建失败的Kafka实例 步骤 1 登录管理控制台。 步骤 2 在管理控制台左上角单击，选择区域。 说明 请选择Kafka实例所在的区域。 步骤 3 在管理控制台左上角单击，选择“应用服务 > 分布式消息服务 Kafka”，进入分布式消息服务Kafka专享版页面。 步骤 4 若当前存在创建失败的Kafka实例，界面信息栏会显示“创建失败任务”及失败数量信息。 说明 创建失败的实例，不会占用其他资源。 步骤 5 单击“创建失败任务”或者此参数后图标/数量，弹出“创建失败任务”对话框。 步骤 6 在“创建失败任务”界面删除创建失败的Kafka实例。 单击“清理失败任务”，一键式删除所有创建失败的Kafka实例。 单击需要删除的Kafka实例右侧的“删除任务”，依次删除创建失败的Kafka实例。