参数 是否必选 说明 节点状态轮询时间（秒） 是 设置轮询时间（1~60秒），每隔x秒查询一次节点是否执行完成。 节点执行的最长时间 是 设置节点执行的超时时间，如果节点配置了重试，在超时时间内未执行完成，该节点将不会再重试，直接置为失败状态。 失败重试 是 节点执行失败后，是否重新执行节点。 是：重新执行节点，请配置以下参数。 − 最大重试次数 − 重试间隔时间（秒） 否：默认值，不重新执行节点。 说明 如果作业节点配置了重试，并且配置了超时时间，该节点执行超时后将不会再重试，直接置为失败状态。 失败策略 是 节点执行失败后的操作： 终止当前作业执行计划：停止当前作业运行，当前作业实例状态显示为“失败”。 继续执行下一节点：忽略当前节点失败，当前作业实例状态显示为“忽略失败成功”。 挂起当前作业执行计划：暂停当前作业运行，当前作业实例状态显示为“等待运行”。 终止后续节点执行计划：停止后续节点的运行，当前作业实例状态显示为“失败”。 空跑 否 如果勾选了空跑，该节点不会实际执行，将直接返回成功。

本章节主要介绍节点参考的Kafka Client。 功能 通过Kafka Client向Kafka的Topic中发送数据。 参数 用户可参考下表配置Kafka Client节点的参数。 参数 是否必选 说明 数据连接 是 选择管理中心中已创建的MRS Kafka连接。 Topic名称 是 选择需要上传数据的Topic，如果有多个partition，默认发送到partition 0。 节点名称 是 节点名称，可以包含中文、英文字母、数字、“”、“”、“/”、“ ”等各类特殊字符，长度为1～128个字符。 发送数据 是 发送到Kafka的文本内容。可以直接输入文本或单击使用EL表达式编辑。 高级参数 参数 是否必选 说明 节点执行的最长时间 是 设置节点执行的超时时间，如果节点配置了重试，在超时时间内未执行完成，该节点将不会再重试，直接置为失败状态。 失败重试 是 节点执行失败后，是否重新执行节点。 是：重新执行节点，请配置以下参数。 − 最大重试次数 − 重试间隔时间（秒） 否：默认值，不重新执行节点。 说明 如果作业节点配置了重试，并且配置了超时时间，该节点执行超时后将不会再重试，直接置为失败状态。 失败策略 是 节点执行失败后的操作： 终止当前作业执行计划：停止当前作业运行，当前作业实例状态显示为“失败”。 继续执行下一节点：忽略当前节点失败，当前作业实例状态显示为“忽略失败成功”。 挂起当前作业执行计划：暂停当前作业运行，当前作业实例状态显示为“等待运行”。 终止后续节点执行计划：停止后续节点的运行，当前作业实例状态显示为“失败”。 空跑 否 如果勾选了空跑，该节点不会实际执行，将直接返回成功。

参数 是否必选 说明 节点执行的最长时间 是 设置节点执行的超时时间，如果节点配置了重试，在超时时间内未执行完成，该节点将不会再重试，直接置为失败状态。 失败重试 是 节点执行失败后，是否重新执行节点。 是：重新执行节点，请配置以下参数。 − 最大重试次数 − 重试间隔时间（秒） 否：默认值，不重新执行节点。 说明 如果作业节点配置了重试，并且配置了超时时间，该节点执行超时后将不会再重试，直接置为失败状态。 失败策略 是 节点执行失败后的操作： 终止当前作业执行计划：停止当前作业运行，当前作业实例状态显示为“失败”。 继续执行下一节点：忽略当前节点失败，当前作业实例状态显示为“忽略失败成功”。 挂起当前作业执行计划：暂停当前作业运行，当前作业实例状态显示为“等待运行”。 终止后续节点执行计划：停止后续节点的运行，当前作业实例状态显示为“失败”。 空跑 否 如果勾选了空跑，该节点不会实际执行，将直接返回成功。

参数 是否必选 说明 节点状态轮询时间（秒） 是 设置轮询时间（1~60秒），每隔x秒查询一次节点是否执行完成。 节点执行的最长时间 是 设置节点执行的超时时间，如果节点配置了重试，在超时时间内未执行完成，该节点将不会再重试，直接置为失败状态。 失败重试 是 节点执行失败后，是否重新执行节点。 是：重新执行节点，请配置以下参数。 − 最大重试次数 − 重试间隔时间（秒） 否：默认值，不重新执行节点。 说明 如果作业节点配置了重试，并且配置了超时时间，该节点执行超时后将不会再重试，直接置为失败状态。 失败策略 是 节点执行失败后的操作： 终止当前作业执行计划：停止当前作业运行，当前作业实例状态显示为“失败”。 继续执行下一节点：忽略当前节点失败，当前作业实例状态显示为“忽略失败成功”。 挂起当前作业执行计划：暂停当前作业运行，当前作业实例状态显示为“等待运行”。 终止后续节点执行计划：停止后续节点的运行，当前作业实例状态显示为“失败”。 空跑 否 如果勾选了空跑，该节点不会实际执行，将直接返回成功。

DRDS支持直接连接DBProxy执行DDL语句,本文为您介绍删除表的DDL语句。 注意 仅V5.1.20.0.13及以后版本的实例，支持使用本文介绍的DDL语句。 语法 plaintext DROP TABLE [IF EXISTS] tblname 参数说明 tblname：待删除的数据表名称。 IF EXISTS：如果指定此选项，则当表不存在时不会报错。 注意 V5.1.9.6020.2544以前版本的DRDS实例，默认开启表回收站功能，当您执行DROP TABLE命令后，可以在表回收站中查看、恢复或者彻底删除该表，具体语法说明，请参见表回收站的DDL语句。若您无需使用表回收站功能，则您可以在执行DROP TABLE命令前使用HINT语法关闭回收站模式，具体语法说明，请参见回收站模式删除表。 V5.1.9.6020.2544及以后版本的DRDS实例，是否开启表回收站功能以控制台上分组参数 模块中的enableRecycleBinMode 参数配置与HINT语法指定的recycleBinMode 参数为准，其中，两种方式的参数优先级为控制台分组参数配置 < HINT语法参数配置 。具体参数配置，请参见管理分组参数和回收站模式删除表。 语法示例 示例1：删除单个表 plaintext DROP TABLE mytable; 示例2：如果表存在，则删除该表 plaintext DROP TABLE IF EXISTS mytable;

在开始 API 接入前，请确保已完成以下准备工作： 1. 开通产品：已在天翼云控制中心成功开通大模型安全护栏服务。 2. 创建应用：在控制台“应用管理”中创建至少一个应用，并从应用的编辑页面获取以下鉴权凭证： 凭证名称 说明 获取位置 accountid 租户 ID 天翼云控制台账号信息页 XAPPCODE（即AK） 应用编码，作为 AK 使用 应用管理 > 操作 > 复制APP鉴权信息 publicKey 公钥 应用管理 > 操作 > 复制APP鉴权信息 privateKey 私钥（SK） 应用管理 > 操作 > 复制APP鉴权信息 （请妥善保管，勿泄露） 获取服务地址 ：请通过工单或者服务热线（4008109889）获取服务地址。

code subCode 说明 处理建议 200 200 请求成功，检测结果在data字段中 正常处理 500 500 请求参数错误（如 reqId 为空、参数格式不正确） 检查请求 Body 参数是否完整、格式是否正确 401 401 鉴权失败 检查所有 Header 签名信息；确认Eopdate未超过 15 分钟有效期；确认 XHMACSIGNATURE 签名计算正确 403 403 无访问权限 检查accountid与XAPPCODE是否匹配；确认应用状态为启用 429 429 请求频率超限 降低调用频率，或联系天翼云技术支持提升配额

本文为您介绍DRDS管理命令中的CHECK命令。 UDAL CHECKDB '?' 语法说明 数据库检查 示例 plaintext mysql> UDAL CHECKDB 'udaladmin'; ++ Datanode Result ++ udaladmin success udaladmin09 success udaladmin06 success udaladmin02 success udaladmin04 success udaladmin07 success udaladmin01 success udaladmin08 success udaladmin05 success udaladmin03 success udaladmin10 success ++ 11 rows in set (0.02 sec) UDAL CHECKTABLE '?' 语法说明 数据库表检查 示例 plaintext mysql> UDAL CHECKTABLE 'student'; +++ Datanode Table Result +++ udaladmin09 student success udaladmin06 student success udaladmin02 student success udaladmin04 student success udaladmin07 student success udaladmin01 student success udaladmin08 student success udaladmin03 student success udaladmin05 student success udaladmin10 student success +++ 10 rows in set (0.08 sec) UDAL CHECKMETA CONSISTENT [ WHERE SCHEMA IN ( '?' , ... ) ] 注意 仅V5.1.9.6020.2533及以后版本的实例，支持执行该命令。 语法说明 用于全局元数据一致性校验（集群内元数据一致性校验），即校验ZooKeeper上的库表与底层标准库、分片间结构信息是否一致，用于数据迁移服务DTS容灾场景下，保障数据可用性。 返回信息说明如下： 参数 描述 Scheme 进行元数据一致性校验的库。 Table 进行元数据一致性校验的表。 Result 校验结果，可能值： success：库或者表元数据校验一致。 failed：库或者表元数据校验不一致，如果库不一致，表示库不存在；如果表信息校验不一致，则会返回详细信息。 示例

本章节介绍注册配置中心Nacos引擎的高可用最佳实践 概述 对于注册配置中心而言，高可用是非常关键的。注册配置中心是微服务体系中的核心依赖，当注册配置中心不可用时，整个微服务体系都将无法对外提供服务。 本文将介绍注册配置中心Nacos引擎的高可用最佳实践。遵照下述设置，将发挥注册配置中心最大的高可用能力。 版本推荐 springcloud：推荐使用2.2.6.RELEASE及以上版本。 dubbo：推荐使用2.7.12及以上版本。 springboot：推荐使用2.3.x及以下版本，2.4.x版本存在兼容问题，不推荐使用。 集群高可用 实例在开通时将提供以下选项，遵照这些选项进行设置，将提升集群的高可用能力。 1. 多节点：提供多节点集群的选项。对于Nacos引擎而言，节点数为奇数（3，5，7，9）。集群节点数越多，可用性越高。 2. 跨可用区：对于多节点的Nacos集群，可提供跨可用区部署。使用跨可用区部署时，节点的故障发生事件将相互独立，集群可用性将进一步提高。 注意 您需要在开通实例时指定可用区模式为“多可用区部署”，才能触发跨可用区高可用能力。 此外，注册配置中心实例将提供额外的手段以提高集群的可用性，这些手段是实例自动附带的，默认为开启状态。这些手段包括： 1. 服务进程保活。Nacos引擎提供了进程保活机制对机器中的Nacos进程进行探测和自动拉起。 2. 典型异常故障自愈。 3. 服务可用性实时探活，极端情况下，不可用时自动告警，人工将介入恢复，保障可用性。

Consumer模块 application.yml配置文件： server: port: 21060 spring: application: name: dubboconsumer dubbo: registry: address: nacos://Nacos访问地址?username用户名&password密码 application: name: dubboconsumer consumer: timeout: 30000 controller，调用公共接口模块创建的接口： @RestController public class InfoController { @DubboReference(check false) private InfoService infoService; @GetMapping("/getInfo") public String getInfo() { return infoService.getInfo(); } } Consumer启动类： @EnableDubbo @SpringBootApplication public class DubboConsumerApplication { public static void main(String[] args) { SpringApplication.run(DubboConsumerApplication.class, args); } } 服务调用测试 启动服务提供者和服务消费者，打MSE 开Nacos控制台>服务管理服务列表页面查看注册中心中的服务，查看服务注册情况。 调用服务消费者的getInfo接口，服务提供者会返回结果。 至此，dubbo融合MSE Nacos作为注册中心成功。

Nacossdkgo Nacossdkgo是Nacos的Go语言客户端，它实现了服务发现和动态配置的功能 使用限制 支持Go>v1.15版本 支持Nacos>2.x版本 安装 安装开源SDK go get github.com/nacosgroup/nacossdkgo/v2 快速使用 ClientConfig constant.ClientConfig{ TimeoutMs uint64 // 请求Nacos服务端的超时时间，默认是10000ms NamespaceId string // ACM的命名空间Id CacheDir string // 缓存service信息的目录，默认是当前运行目录 UpdateThreadNum int // 监听service变化的并发数，默认20 NotLoadCacheAtStart bool // 在启动的时候不读取缓存在CacheDir的service信息 UpdateCacheWhenEmpty bool // 当service返回的实例列表为空时，不更新缓存，用于推空保护 Username string // Nacos服务端的API鉴权Username Password string // Nacos服务端的API鉴权Password LogDir string // 日志存储路径 RotateTime string // 日志轮转周期，比如：30m, 1h, 24h, 默认是24h MaxAge int64 // 日志最大文件数，默认3 LogLevel string // 日志默认级别，值必须是：debug,info,warn,error，默认值是info } ServerConfig constant.ServerConfig{ ContextPath string // Nacos的ContextPath，默认/nacos，在2.0中不需要设置 IpAddr string // Nacos的服务地址 Port uint64 // Nacos的服务端口 Scheme string // Nacos的服务地址前缀，默认http，在2.0中不需要设置 GrpcPort uint64 // Nacos的 grpc 服务端口, 默认为 服务端口+1000, 不是必填 } 我们可以配置多个ServerConfig，客户端会对这些服务端做轮询请求 服务发现 注销实例：DeregisterInstance success, err : namingClient.DeregisterInstance(vo.DeregisterInstanceParam{ Ip: "${ipAddr}", Port: 8848, ServiceName: "demo.go", Ephemeral: true, Cluster: "clustera", // 默认值DEFAULT GroupName: "groupa", // 默认值DEFAULTGROUP }) 获取服务信息：GetService services, err : namingClient.GetService(vo.GetServiceParam{ ServiceName: "demo.go", Clusters: []string{"clustera"}, // 默认值DEFAULT GroupName: "groupa", // 默认值DEFAULTGROUP }) 获取所有的实例列表：SelectAllInstances services, err : namingClient.GetService(vo.GetServiceParam{ ServiceName: "demo.go", Clusters: []string{"DEFAULT"}, // 默认值 GroupName: "groupa", // 默认值DEFAULTGROUP }) 获取实例列表 ：SelectInstances // SelectInstances 只返回满足这些条件的实例列表：healthy${HealthyOnly},enabletrue 和weight>0 instances, err : namingClient.SelectInstances(vo.SelectInstancesParam{ ServiceName: "demo.go", GroupName: "groupa", // 默认值DEFAULTGROUP Clusters: []string{"clustera"}, // 默认值DEFAULT HealthyOnly: true, }) 获取一个健康的实例（加权随机轮询）：SelectOneHealthyInstance // SelectOneHealthyInstance将会按加权随机轮询的负载均衡策略返回一个健康的实例 // 实例必须满足的条件：healthtrue,enabletrue and weight>0 instance, err : namingClient.SelectOneHealthyInstance(vo.SelectOneHealthInstanceParam{ ServiceName: "demo.go", GroupName: "groupa", // 默认值DEFAULTGROUP Clusters: []string{"clustera"}, // 默认值DEFAULT }) 监听服务变化：Subscribe // Subscribe keyserviceName+groupName+cluster // 注意:我们可以在相同的key添加多个SubscribeCallback. err : namingClient.Subscribe(vo.SubscribeParam{ ServiceName: "rccdemo", GroupName: "DEFAULTGROUP", // 默认值DEFAULTGROUP Clusters: []string{"clustera"}, // 默认值DEFAULT SubscribeCallback: func(services []model.Instance, err error) { log.Printf("nn callback return services:%s nn", utils.ToJsonString(services)) }, })

创建客户端CEK数据密钥，用来对数据库用户数据加解密。 plaintext CREACREATE COLUMN ENCRYPTION KEY columnencryptionkeyname WITH( CLIENTMASTERKEY clientmasterkeyname, ALGORITHM algorithmtype, ENCRYPTEDVALUE encryptedvalue ); 参数描述： columnencryptionkeyname：CEK数据密钥对象名，在同一命名空间中满足唯一性约束 CLIENTMASTERKEY：指定用于对当前CEK加密的CMK ALGORITHM：指定该CEK将用于何种加密算法，支持： AEADAES256CBCHMACSHA256：采用AES256 CBC模式对数据加密实现数据机密性，通过消息验证码(SHA256生成数据摘要)确保数据的完整性和身份校验。 AEADAES128CBCHMACSHA256：采用AES128 CBC模式对数据加密实现数据机密性，通过消息验证码(SHA256生成数据摘要)确保数据的完整性和身份校验。 SM4SM3：采用国密SM4 CBC模式对数据加密实现数据机密性，通过消息验证码(国密SM3生成数据摘要)确保数据的完整性和身份校验。 ENCRYPTEDVALUE（可选项）：用户指定的密钥口令，如果不指定，则会自动生成密钥。密钥口令长度范围为[28, 256]个字符。28个字符派生出来的密钥安全强度满足AES128。如果采用用AES256，密钥口令的长度需要39个字符。如果不指定，则会自动生成256比特的密钥。

本页介绍如何根据用户输入的加密算法ID将其从系统中注销。 plaintext SELECT MLSTRANSPARENTCRYPTDROPALGORITHM(algoid) 参数描述： algoid：加密算法唯一性ID

本文为您介绍SELECT语法的特殊应用场景，例如多行变成单行和一列变成多行。 多行变成单行 plaintext teledb select arraytostring(array(select nickname from teledbpg1),','); arraytostring test,pg (1 row) 一列变成多行 plaintext teledb insert into teledbpg1 values(6, 'teledb, teledb1'); INSERT 0 1 teledb select regexpsplittotable((select nickname from teledbpg1 where id6 limit 1), ','); regexpsplittotable teledb teledb1 (2 rows)

本文为您介绍如何管理表达式索引。 plaintext teledb create table tupper(id int,mc text); CREATE TABLE teledb create index tuppermc on tupper(mc); CREATE INDEX teledb insert into tupper select t,md5(t::text) from generateseries(1,10000) as t; INSERT 0 10000 teledb analyze tupper; ANALYZE teledb explain select from tupper where upper(mc)md5('1'); QUERY PLAN Remote Fast Query Execution (cost0.00..0.00 rows0 width0) Node/s: dn01, dn02 > Seq Scan on tupper (cost0.00..133.58 rows25 width37) Filter: (upper(mc) 'c4ca4238a0b923820dcc509a6f75849b'::text) (4 rows) teledb create index tuppermc on tupper(upper(mc)); ERROR: relation "tuppermc" already exists teledb create index tuppermc1 on tupper(upper(mc)); CREATE INDEX teledb explain select from tupper where upper(mc)md5('1'); QUERY PLAN Remote Fast Query Execution (cost0.00..0.00 rows0 width0) Node/s: dn01, dn02 > Bitmap Heap Scan on tupper (cost4.48..50.94 rows25 width37) Recheck Cond: (upper(mc) 'c4ca4238a0b923820dcc509a6f75849b'::text) > Bitmap Index Scan on tuppermc1 (cost0.00..4.47 rows25 width0) Index Cond: (upper(mc) 'c4ca4238a0b923820dcc509a6f75849b'::text) (6 rows)

explain (analyze,buffers) select count(1) from t1 where f21; QUERY PLAN Aggregate (cost7201.23..7201.24 rows1 width8) (actual time9.808..9.808 rows1 loops1) Buffers: shared hit116 > Bitmap Heap Scan on t1 (cost124.87..7177.86 rows9348 width0) (actual time1.312..7.348 rows10051 loops1) Recheck Cond: (f2 1) Heap Blocks: exact85 Buffers: shared hit116 > Bitmap Index Scan on t1f2idx (cost0.00..122.53 rows9348 width0) (actual time1.219..1.219 rows10051 loops1) Index Cond: (f2 1) Buffers: shared hit31 Planning time: 0.696 ms Execution time: 9.969 ms (11 rows)

本文为你介绍合并多个查询结果的方法。 不过滤重复的记录。 plaintext teledb create table teledbpg1(id int, nickname varchar); CREATE TABLE teledb insert into teledbpg1 values(3, 'pg'),(5,'test'); COPY 2 teledb select from teledbpg union all select from teledbpg1; id nickname + 1 teledb 1 hello,pgxc 2 TELEDB 3 pg 4 5 test 3 pg (7 rows) 过滤重复的记录。 plaintext teledb select from teledbpg union select from teledbpg1; id nickname + 1 teledb 1 hello,pgxc 4 5 test 2 TELEDB 3 pg (6 rows) 每个子查询分布在合并结果中的使用。 plaintext teledb select from (select from teledbpg limit 2) as t union all select from (select from teledbpg1 limit 2) teledb ; id nickname + 1 teledb 1 hello,pgxc 5 test 3 pg (4 rows)

本页介绍天翼云TeleDB数据库删除存储过程的语法。 删除不带参数的存储过程 plaintext teledb CREATE OR REPLACE PROCEDURE proc1() AS $$ begin raise notice 'Hello teledbpg'; end; $$ LANGUAGE PLPGSQL; CREATE PROCEDURE teledb drop procedure proc1 ( ); DROP PROCEDURE teledb 删除带参数的存储过程 plaintext teledb CREATE OR REPLACE PROCEDURE proc1(aint int) AS $$ begin raise notice '%',aint; end; $$ LANGUAGE PLPGSQL; CREATE PROCEDURE teledb drop procedure proc1 ( aint int); DROP PROCEDURE teledb 也可以只指定参数的类型即可。 plaintext teledb drop procedure proc1 (int); DROP PROCEDURE teledb

本页介绍天翼云TeleDB数据库存储过程执行的语法。 不带参数 plaintext teledb create or replace procedure pnopara() as $$ begin raise notice 'nopara procedure'; end; $$ language plpgsql; CREATE PROCEDURE teledb call pnopara(); NOTICE: nopara procedure CALL teledb 带参数 plaintext teledb create or replace procedure ppara(aint integer) as $$ begin raise notice 'aint %',aint; end; $$ language plpgsql; CREATE PROCEDURE teledb call ppara(1); NOTICE: aint 1 CALL teledb

应用性能监控APM探针在应用运行时进行字节码增强，实现应用性能管理能力。与其他通过字节码增强技术实现的性能管理方案一样，APM探针会带来一定的应用性能开销，您可以参考本篇分析报告，在接入应用性能监控APM前，基于性能影响进行充分的评估。 测试用例 机器类型: CTyunOS V4.0 25.07 64 位，2C4G 测试时长：10分钟 jvm参数：Xms3500m Xmx3500m XX:MetaspaceSize500m 测试场景：整体架构如下图所示 Java应用基于Spring MVC框架编写，包含 Spring Boot、Spring MVC，Jedis，HikariCP 连接池，每次请求产生5个span（1个 Tomcat 调用、1个 Spring MVC 框架调用、2个 Redis 请求和1个 MySQL 请求），每次请求会随机sleep 2040ms。 基线性能 在不接入 APM 的情况下，性能表现如下： 业务流量（TPS） CPU 利用率 内存利用率 平均响应时间（ms） 500 9.5% 79% 30 1000 15.5% 79% 30 2000 26.5% 80.15% 30 安装探针后的性能指标 应用接入 APM 后，性能表现如下： 业务流量（TPS） CPU 利用率 内存利用率 平均响应时间（ms） 500 25.3% 82% 31 1000 40.1% 82.2% 31.1 2000 66.5% 83.4% 31.2

本章节介绍普通消息的收发方法和示例代码。其中，普通消息发送方式分为同步发送、异步发送、单向发送。 同步发送：同步发送是指消息发送方发出一条消息后，会在收到服务端同步响应之后才发下一条消息的通讯方式。 异步发送：异步发送是指发送方发出一条消息后，不等服务端返回响应，接着发送下一条消息的通讯方式。 单向发送：发送方只负责发送消息，不等待服务端返回响应且没有回调函数触发。 收发消息前，请参考收集连接信息收集RocketMQ所需的连接信息。 准备环境 开源的Java客户端支持连接分布式消息服务RocketMQ版，推荐使用的客户端版本为4.9.7。 通过以下任意一种方式引入依赖： 1. 使用Maven方式引入依赖。 org.apache.rocketmq rocketmqclient 4.9.7 org.apache.rocketmq rocketmqacl 4.9.7 2. 点击下载依赖JAR包：rocketmqall4.9.7binrelease.zip 同步发送 同步发送是最常用的方式，是指消息发送方发出一条消息后，会在收到服务端同步响应之后才发下一条消息的通讯方式，可靠的同步传输被广泛应用于各种场景，如重要的通知消息、短消息通知等。 参考如下示例代码 import org.apache.rocketmq.acl.common.AclClientRPCHook; import org.apache.rocketmq.acl.common.SessionCredentials; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.RPCHook; import org.apache.rocketmq.remoting.common.RemotingHelper; public class ProducerNormalExample { private static RPCHook getAclRPCHook() { return new AclClientRPCHook(new SessionCredentials( "accessKey", // 分布式消息服务RocketMQ控制台用户管理菜单中创建的用户ID "accessSecret" // 分布式消息服务RocketMQ控制台用户管理菜单中创建的密钥 )); } public static void main(String[] args) throws Exception { DefaultMQProducer producer new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook()); // 填入控制台获取NAMESRV接入点地址 producer.setNamesrvAddr("XXX:xxx"); ; // 如果需要开启SSL，请增加此行代码 producer.start(); for (int i 0; i < 128; i++) try { { Message msg new Message("YOUR TOPIC", "TagA", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULTCHARSET)); SendResult sendResult producer.send(msg); System.out.println(sendResult); } } catch (Exception e) { e.printStackTrace(); } producer.shutdown(); } }

在分布式消息服务RocketMQ中，消费者消费消息时，可能会出现消费过快导致下游业务来不及处理的情况，进而影响系统的稳定性。本章节介绍在消费端进行限流的示例代码，以保障系统的稳定。 import java.util.List; import java.util.concurrent.TimeUnit; import com.google.common.util.concurrent.RateLimiter; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.common.consumer.ConsumeFromWhere; import org.apache.rocketmq.common.message.MessageExt; public class ConsumerLimitExample { public static void main(String[] args) throws InterruptedException, MQClientException { DefaultMQPushConsumer consumer new DefaultMQPushConsumer("GroupTest"); consumer.subscribe("TopicTest", ""); consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUMEFROMFIRSTOFFSET); RateLimiter rateLimiter RateLimiter.create(200); consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) { if (!rateLimiter.tryAcquire(msgs.size(),3, TimeUnit.SECONDS)) { return ConsumeConcurrentlyStatus.RECONSUMELATER; } System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs); return ConsumeConcurrentlyStatus.CONSUMESUCCESS; } }); consumer.start(); System.out.printf("Consumer Started.%n"); } }

订阅顺序消息 参考如下示例代码 import org.apache.rocketmq.acl.common.AclClientRPCHook; import org.apache.rocketmq.acl.common.SessionCredentials; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly; import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely; import org.apache.rocketmq.remoting.RPCHook; public class ConsumerFifoExample { private static RPCHook getAclRPCHook() { return new AclClientRPCHook(new SessionCredentials( "accessKey", // 分布式消息服务RocketMQ控制台用户管理菜单中创建的用户ID "accessSecret" // 分布式消息服务RocketMQ控制台用户管理菜单中创建的密钥 )); } public static void main(String[] args) throws Exception { / 创建Consumer，如果想开启消息轨迹，可以按照如下方式创建： DefaultMQPushConsumer consumer new DefaultMQPushConsumer("YOUR GROUP ID", getAclRPCHook(), new AllocateMessageQueueAveragely(), true, null); / DefaultMQPushConsumer consumer new DefaultMQPushConsumer("YOUR GROUP ID", getAclRPCHook(), new AllocateMessageQueueAveragely()); // 填入控制台NAMESRV接入点地址 consumer.setNamesrvAddr("XXX:xxx"); // consumer.setUseTLS(true); // 如果需要开启SSL，请增加此行代码 / 如果想要消费指定TAG的消息，可以按照如下方式订阅： 为订阅所有的TAG pushConsumer.subscribe(TOPICNAME, "Tag1"); / consumer.subscribe("TopicTest", ""); consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) > { System.out.printf("Receive New Messages: %s %n", msgs); return ConsumeOrderlyStatus.SUCCESS; }); consumer.start(); System.out.println("Consumer Started."); } }

本文介绍如何使用Go请求处理程序响应接收到的事件并执行相应的业务逻辑。 请求处理程序 请求处理程序 是您提供的一个方法。当您的函数被调用时，函数计算会运行该方法处理请求。 您可以通过函数计算控制台页面配置请求处理程序，对于Go语言的函数，请求处理程序会被被编译为一个可执行的二进制文件。例如，您的可执行二进制文件名为handler，则请求处理程序直接配置为handler。 在Go语言的代码中，您需要引入官方的SDK库 gitee.com/ctyunfaas/cfruntimegosdk/cf，并实现 handler方法和 main方法。示例如下： package main import ( "log" "gitee.com/ctyunfaas/cfruntimegosdk/cf" ) func main() { cf.Start(HandleRequest) } func HandleRequest() (string, error) { log.Println("hello world!") return "hello world"!", nil } Handler方法签名 下面列举出了有效的Event Handler方法签名，其中 InputType和 OutputType与 encoding/json标准库兼容。 函数计算会使用 json.Unmarshal方法对传入的 InputType进行反序列化，以及使用 json.Marshal方法对返回的 OutputType进行序列化。 func () func () error func (InputType) error func () (OutputType, error) func (InputType) (OutputType, error) func (context.Context) error func (context.Context, InputType) error func (context.Context) (OutputType, error) func (context.Context, InputType) (OutputType, error) Handler的使用需遵循以下规则： Handler必须是一个方法。 Handler支持0～2个输入参数。如果有2个参数，则第一个参数必须是 context.Context。 Handler支持0～2个返回值。如果有1个返回值，则必须是 error类型；如果有2个返回值，则第2个返回值必须是 error。

本文介绍如何使用PHP请求处理程序响应接收到的事件并执行相应的业务逻辑。 请求处理程序 请求处理程序是您提供的一个方法。当您的函数被调用时，函数计算会运行该方法处理请求。 您可以通过函数计算控制台页面配置请求处理程序 ，对于PHP语言的函数，请求处理程序需配置为 [文件名].[方法名]。例如，您的文件名为index.php，方法名为handler，则请求处理程序可配置为 index.handler。 如下是一个事件请求处理程序的方法签名示例： php getBody()>getContents(); $logger $GLOBALS['fcLogger']; $logger>info('request body:' . $body); return new Response( 200, array( 'helloheader1' > 'aaa', 'helloheader2' > ['bbb', 'ccc'], 'ContentType2' > 'application/json' ), 'hello world' ); }

字段 类型 解释说明 errorMessage string 异常信息。 errorType string 异常类型。 stackTrace string[] 异常堆栈。

运行时可为您的函数提供针对不同语言在执行环境中运行的环境。运行时可传递函数调用的事件（event）、上下文信息（context）和响应。天翼云函数计算支持多种语言的标准运行时，同时也可以使用标准运行时构建您自己的运行时，或自行构建容器镜像。 标准运行时 Node.js Python Java Go C PHP 自定义运行时 更多信息，请参考自定义运行时章节。 容器镜像函数 更多信息，请参考容器镜像函数章节。

版本 标识符 操作系统 架构 Python 3.10.12 Python 3.10 Linux (Debian 10) x8664 Python 3.9.16 Python 3.9 Linux (Debian 10) x8664 Python 3.6.15 Python 3.6 Linux (Debian 10) x8664

字段 类型 说明 requestId String 函数请求ID。 credentials Credentials，该结构字段如下： accessKeyId accessKeySecret securityToken 临时密钥信息。 function FunctionMeta，该结构字段如下： name handler memory timeout 当前函数的基本信息。 service ServiceMeta，该结构字段如下： logProject logStore qualifier versionId 当前函数的服务信息。 region String 当前函数的所在地域ID。 accountId String 当前函数的所属账号ID。 logger ContextLog，该结构字段如下： debug info warn error log 日志对象。

本文介绍Python如何实现并应用函数实例生命周期回调方法。 使用说明 当您实现并配置函数实例生命周期回调后，函数计算将在相关实例生命周期事件发生时调用对应的回调程序。函数实例生命周期涉及Initializer和PreStop回调。 Initializer回调 Initializer回调在函数实例启动成功后，请求处理程序（Handler）之前执行。 PreStop回调 PreStop回调在函数实例销毁前执行。 Initializer回调和PreStop回调的方法签名一样，入参只有一个context参数，提供在调用时的运行上下文信息。回调方法定义如下： coding: utf8 import logging def initialize(context): print("initialize...") def preStop(context): print("preStop...")

本文介绍在函数计算中Context的相关概念和使用示例。 上下文 当函数计算运行您的函数时，会将上下文（Context）对象传递到执行方法中。该对象包含有关调用、服务、函数和执行环境等信息。Context接口定义如下： package com.ctg.faas.runtime; / The context object allows you to access useful information available within the Tianyi cloud function compute execution environment / public interface Context { / Gets the function request ID associated with the request. This is the same ID returned to the client that called invoke(). This ID is reused for retries on the same request. @return The request id / public String getRequestId(); / Gets the credentials of the execution role @return The request Credentials / public Credentials getExecutionCredentials(); / Gets the function related parameters @return The request function params / public FunctionParam getFunctionParam(); / Gets the service related parameters @return The request service info / public Service getService(); / Gets the function compute logger instance associated with the context object @return The fc logger / public FunctionComputeLogger getLogger(); / Gets the function compute opentracing instance associated with the context object @return The fc tracing / public OpenTracing getTracing(); / Gets the function compute retry count associated with the context object @return The retry count / public int getRetryCount(); } 示例：获取函数相关信息 使用Context获取函数Handler和Initializer： package example; import com.ctg.faas.runtime.Context; import com.ctg.faas.runtime.FunctionParam; import com.ctg.faas.runtime.PojoRequestHandler; public class PojoHandler implements PojoRequestHandler { @Override public String handleRequest(String in, Context context) { FunctionParam functionParam context.getFunctionParam(); String initializer functionParam.getFunctionInitializer(); String handler functionParam.getFunctionHandler(); return String.format("Function initializer is: %s, handler is: %s", initializer, handler); } }

本文介绍Java如何实现并应用函数实例生命周期回调方法。 使用说明 当您实现并配置函数实例生命周期回调后，函数计算将在相关实例生命周期事件发生时调用对应的回调程序。函数实例生命周期涉及Initializer和PreStop回调。 Initializer回调 Initializer回调在函数实例启动成功后，请求处理程序（Handler）之前执行。使用Initializer回调需要继承com.ctg.faas.runtime.FunctionInitializer接口，并实现该接口的initialize方法。 PreStop回调 PreStop回调在函数实例销毁前执行。使用PreStop回调需要继承com.ctg.faas.runtime.PreStopHandler接口，并实现该接口的preStop方法。 Initializer回调和PreStop回调的方法签名一样，入参只有一个context参数，提供在调用时的运行上下文信息。一个包含回调方法的示例如下： package example; import com.ctg.faas.runtime.Context; import com.ctg.faas.runtime.FunctionInitializer; import com.ctg.faas.runtime.PojoRequestHandler; import com.ctg.faas.runtime.PreStopHandler; public class PojoHandler implements PojoRequestHandler , FunctionInitializer, PreStopHandler { @Override public String handleRequest(String input, Context context) { context.getLogger().info("Hello:" + input); return input; } @Override public void initialize(Context context) { context.getLogger().info("initialize..."); } @Override public void preStop(Context context) { context.getLogger().info("preStop..."); } }