华为云用户手册

前提条件 准备装有1.8及以上版本JDK的Linux服务器。 IntelliJ IDEA版本为：2018.3.5或以上版本，Eclipse版本为：3.6.0或以上版本。 通过Demo（sha256:34c9bc8d99eba4ed193603019ce2b69afa3ed760a452231ece3c89fd7dd74da1）获取RealtimeConnector.rar包。 用户程序向MQS写入消息的TPS不能超过6000/s。

调用API（Node.js）示例 在工程中引入signer.js。 var signer = require('./signer')var http = require('http') 生成一个新的Signer，输入API所授权凭据的Key和Secret，可参考获取API的调用信息获取。 var sig = new signer.Signer()// Directly writing AK/SK in code is risky. For security, encrypt your AK/SK and store them in the configuration file or environment variables. // In this example, the AK/SK are stored in environment variables for identity authentication. Before running this example, set environment variables HUAWEICLOUD_SDK_AK and HUAWEICLOUD_SDK_SK. sig.Key = process.env.HUAWEICLOUD_SDK_AKsig.Secret = process.env.HUAWEICLOUD_SDK_SK 生成一个Request对象，指定方法名、请求url和body，可参考获取API的调用信息获取。 var r = new signer.HttpRequest("POST", "c967a237-cd6c-470e-906f-a8655461897e.apigw.exampleRegion.com/app1?a=1");r.body = '{"a":1}' 给请求添加x-stage头，内容为环境名。如有需要，添加需要签名的其他头域。 r.headers = { "x-stage":"RELEASE" } 进行签名，执行此函数会生成请求参数，用于创建http(s)请求，请求参数中添加了用于签名的X-Sdk-Date头和Authorization头。 var opts = sig.Sign(r) 访问API，查看访问结果。如果使用https访问，则将“http.request”改为“https.request”。 var req=http.request(opts, function(res){ console.log(res.statusCode) res.on("data",function(chunk){console.log(chunk.toString())})})req.on("error",function(err){console.log(err.message)})req.write(r.body)req.end()

准备开发环境 安装开发工具 请根据所使用语言选择适合的开发工具。 安装2018.3.5或以上版本的IntelliJ IDEA，安装包请至IntelliJ IDEA官方网站下载。 安装2019 version 16.8.4及以上版本的Visual Studio，安装包请至Visual Studio官方网站下载。 安装开发语言环境 Java：安装1.8.111或以上版本的Java Development Kit，安装包请至Oracle官方下载页面下载。 Python：安装2.7或3.X版本的Python安装包，安装包请至Python官方下载页面下载。

前提条件 已获取API的调用信息，具体请参见开发准备。 已安装开发工具和JavaScript开发语言环境，具体请参见开发准备。 Nodejs安装后，在命令行中，用npm安装“moment”和“moment-timezone”模块。 npm install moment --savenpm install moment-timezone --save 已在IntelliJ IDEA中安装NodeJS插件，如果未安装，请按照图1所示安装。 图1 安装NodeJS插件 浏览器版本为Chrome 89.0或以上版本。

响应消息 响应参数 参数 类型 说明 handler String 消息handler。 message Object 消息的内容。 表2 message参数 参数 类型 说明 content String 消息体的内容。Base64加密密文。 响应示例 [ { "handler": "NCMxMDAjMTgjMA==", "message": { "content": "ImhlbGxvIGh1YXdlaWNsb3VkLTIi" } }]

URI GET /v1/topic/{topic_name}/group/{group_name}/messages?ack_wait={ack_wait}& time_wait={time_wait}&max_msgs={max_msgs} 表1 参数说明 参数 类型 必选 说明 topic_name String 是 主题名称。 group_name String 是 消费组名称。长度不超过249位的字符串，包含a~z，A~Z，0~9、中划线（-）和下划线（_）。 ack_wait Integer 否 提交确认消费的超时时间，客户端需要在该时间内提交消费确认，如果超过指定时间，没有确认消费，系统会报消息确认超时或handler无效，则默认为消费失败。取值范围：1~300s。默认值：15s time_wait Integer 否 设定队列可消费的消息为0时的读取消息等待时间。 如果在等待时间内有新的消息，则立即返回消费结果，如果等待时间内没有新的消息，则到等待时间后返回消费结果。取值范围：1~30s。 默认值：3s max_msgs Integer 否 获取可消费的消息的条数。取值范围：1~10。默认值：10 max_bytes Integer 否 每次消费的消息总负载最大值。取值范围：1~ 2097152。默认值：524288。

修改配置信息 为了方便，下文分生产与消费两个配置文件介绍。如果ROMA Connect实例开启了SASL认证，在Java客户端的配置文件中必须配置涉及SASL认证的相关信息，否则无法连接。如果没有使用SASL认证，请注释掉相关配置。 生产消息配置文件（对应生产消息代码中的mqs.sdk.producer.properties文件） 以下粗体部分为不同MQS特有的信息，必须修改。客户端其他参数，可以自主添加。 #Topic名称在具体的生产与消费代码中。########################举例：bootstrap.servers=192.168.0.196:9095,192.168.0.196:9096,192.168.0.196:9094bootstrap.servers=ip1:port1,ip2:port2,ip3:port3#发送确认参数acks=all#键的序列化方式key.serializer=org.apache.kafka.common.serialization.StringSerializer#值的序列化方式value.serializer=org.apache.kafka.common.serialization.StringSerializer#producer可以用来缓存数据的内存大小buffer.memory=33554432#重试次数retries=0########################如果不使用SASL认证，以下参数请注释掉。########################设置用户名和密码sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="username" \ password="********";#SASL鉴权方式sasl.mechanism=PLAIN#加密协议，目前支持SASL_SSL协议security.protocol=SASL_SSL#ssl truststore文件的位置ssl.truststore.location=E:\\temp\\client.truststore.jks#ssl truststore文件的密码，固定，请勿修改。配置此密码是为了访问Java生成的jks文件。ssl.truststore.password=dms@kafkassl.endpoint.identification.algorithm= 示例代码中的参数说明，可参考获取MQS连接信息获取参数值。 bootstrap.servers：MQS连接地址和端口。 username和password：开启SASL_SSL认证时所使用的用户名和密码。 ssl.truststore.location：开启SASL_SSL认证时所使用的客户端证书。 消费消息配置文件（对应消费消息代码中的mqs.sdk.consumer.properties文件） 以下粗体部分为不同MQS特有的信息，必须修改。客户端其他参数，可以自主添加。 #Topic名称在具体的生产与消费代码中。########################举例：bootstrap.servers=192.168.0.196:9095,192.168.0.196:9096,192.168.0.196:9094bootstrap.servers=ip1:port1,ip2:port2,ip3:port3#用来唯一标识consumer进程所在组的字符串，请您自行设定。#如果设置同样的group id，表示这些processes都是属于同一个consumer groupgroup.id=1#键的序列化方式key.deserializer=org.apache.kafka.common.serialization.StringDeserializer#值的序列化方式value.deserializer=org.apache.kafka.common.serialization.StringDeserializer#偏移量的方式auto.offset.reset=earliest########################如果不使用SASL认证，以下参数请注释掉。########################设置jaas账号和密码，通过控制台设置sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="username" \ password="********";#SASL鉴权方式sasl.mechanism=PLAIN#加密协议，目前支持SASL_SSL协议security.protocol=SASL_SSL#ssl truststore文件的位置ssl.truststore.location=E:\\temp\\client.truststore.jks#ssl truststore文件的密码，配置此密码是为了访问Java生成的jks文件。ssl.truststore.password=dms@kafkassl.endpoint.identification.algorithm= 示例代码中的参数说明，可参考获取MQS连接信息获取参数值。 bootstrap.servers：MQS连接地址和端口。 group.id：消费组名称。根据业务需要，自定义消费组名称，如果设置的消费组不存在，系统会自动创建。 username和password：开启SASL_SSL认证时所使用的用户名和密码。 ssl.truststore.location：开启SASL_SSL认证时所使用的客户端证书。

方法详情 public static String json2Xml(String json) json转xml 输入参数 json：json格式的字符串 返回信息 返回xml格式字符串 public static String xml2Json(String xml) xml转json 输入参数 xml：xml格式的字符串 返回信息 返回json格式字符串 public static String json2XmlWithoutType(String json) json转xml 输入参数 json：json格式的字符串 返回信息 返回xml格式字符串 public static HashMap jsonToMap(String json) json转map 输入参数 json：json格式的字符串 返回信息 返回map格式字符串

构造器详情 public RabbitMqConfig(ConnectionConfig connectionConfig, QueueConfig queueConfig, ExchangeConfig exchangeConfig) 构造一个RabbitMQ客户端配置。 参数： connectionConfig表示客户端连接配置。 queueConfig表示队列配置。 exchangeConfig表示交换器配置。

方法列表 返回类型 方法和说明 static String json2Xml(String json) json转xml static String xml2Json(String xml) xml转json static String json2XmlWithoutType(String json) json转xml static HashMap jsonToMap(String json) json转map

示例 { "devices": [{ "deviceId": "D68NZxB4", "services": [{ "data": { "key": "value" }, "eventTime": "20191023T173625Z", "serviceId": "serviceName" }] }]}

数据API中的执行语句说明 数据API调用存储过程时，参数可通过后端服务请求的Headers、Parameters或者Body传递，参数名的语法为：{参数名}.{数据类型}.{传输类型}。 数据类型包括String和int。 传输类型指入参或出参声明，入参使用in，出参使用out。 数据API中调用存储过程的执行语句示例： call sb_test(${nname.String.in},${nsal.int.out}) 该脚本示例中，nname为字符串类型的入参，参数名为nname.String.in ，value则是你要查询的参数值。nsal为数值类型的出参，参数名为nsal.int.out，由于格式限定，出参的value也需要填写，可填写符合数据类型的任意值，不影响输出结果。 数据API中对存储过程的调用，用String和int来区分字符串和数值，无需加单引号，这一点和SQL要求不一样。 在后端服务的Headers、Parameters或者Body中定义的参数名不能相同，否则将被覆盖。 Body传递参数示例： 后端服务请求的Body内容 { "nname.String.in": "zhang", "nsal": 0} 响应结果 { "test": [ 5000 ]} Parameters传递参数示例： 后端服务请求的Parameters内容 https://example.com?nname.String.in=zhang&nsal=0 响应结果 { "test": [ 5000 ]}

数据源说明 假设数据库里面有一张表，表结构如下建表语句所示： create table sp_test(id number,name varchar2(50),sal number); 往表中并插入数据，数据集如下表所示： 表1 sp_test表数据集 ID NAME SAL 1 ZHANG 5000 2 LI 6000 3 ZHAO 7000 4 WANG 8000 在Oracle数据库中调用存储过程，根据name查询sal的值。 create or replace procedure APICTEST.sb_test(nname in varchar, nsal out number) asbeginselect sal into nsal from sp_test where name = nname;end;

准备开发环境 安装开发工具 请根据所使用语言选择适合的开发工具。 安装2018.3.5或以上版本的IntelliJ IDEA，安装包请至IntelliJ IDEA官方网站下载。 安装3.0.3及以上版本Apache Maven，安装包请至Maven官方下载页面下载。 安装2019 version 16.8.4及以上版本的Visual Studio，安装包请至Visual Studio官方网站下载。 安装开发语言环境 Java：安装1.8.111或以上版本的Java Development Kit，安装包请至Oracle官方下载页面下载。 Python：安装2.7或3.X版本的Python，安装包请至Python官方下载页面下载。 Go：安装1.14及以上版本的Go，安装包请至Go官方下载页面下载。 C#：安装6.0及以上版本的.NET，安装包请至.NET官方下载页面下载。

开发规范 开发工具版本要求： IntelliJ IDEA：2018.3.5或以上版本。 Eclipse：3.6.0或以上版本。 开发语言环境版本要求： 1.8.111或以上版本的Java Development Kit。 设备开发要求： 通过MQTT协议接入设备时，当前仅支持MQTT协议中的QoS0和QoS1。 为防止因网络不稳定或实例升级等原因导致设备连接中断，建议在设备开发过程中增加自动重连机制。如果使用ROMA Connect提供的设备demo，则默认已开启重连机制。如果使用开源MQTT客户端，则需要根据开源代码配置重连机制。开启自动重连后，如果连接丢失，客户端会一直自动重连服务器，直到连接成功为止。 父主题： 开发说明

方法详情 public static String JSON2XML(String json, boolean returnFormat) json转xml 输入参数 json：json格式的字符串 returnFormat：返回格式 返回信息 返回xml格式字符串 public static String XML2JSON(String xml) xml转json 输入参数 xml：xml格式的字符串 返回信息 返回xml格式字符串

方法列表 返回类型 方法和说明 static String toString(byte[] bytes, String encoding) 将字节数组转换为字符串 static String toString(byte[] bytes) 将字节数组转换成UTF-8编码字符串 static String toHexString(byte[] data) 将字节数组转成十六进制小写字符串 static byte[] hexToByteArray(String hex) 将十六进制字符串转换成字节数组

方法详情 public static String toString(byte[] bytes, String encoding) 将字节数组转换为字符串 输入参数 bytes：需要转换的字节数组 encoding：编码 返回信息 返回转换后的字符串 public static String toString(byte[] bytes) 将字节数组转换成UTF-8编码字符串 输入参数 bytes：需要转换的字节数组 返回信息 返回转换后的字符串 public static String toHexString(byte[] data) 将字节数组转成十六进制小写字符串 输入参数 data：需要转换的字节数组 返回信息 返回转换后的十六进制字符串 public static byte[] hexToByteArray(String hex) 将十六进制字符串转换成字节数组 输入参数 hex：需要转换的十六进制字符串 返回信息 返回转换后的字节数组

示例 { "data": [{ "deviceInfo": { "manufacturerId": "Test_n", "name": "n-device", "model": "A_n", "nodeId": "n-device", "deviceId": "D59eGSxy" }, "statusCode": 0 }], "mid": 7, "statusCode": 0}

使用示例 在使用CacheUtils之前，需要new一个新对象方可调用。 CacheUtils的get方法只允许使用白名单里的几项作为键： "DICT:api_gw_rest_addr"、"DICT:api_gw_rest_float_addr"、"DICT:api_gw_rest_eip_addr"、"DICT:livedata_private_address" 示例： importClass(com.huawei.livedata.lambdaservice.util.CacheUtils);function execute(data) { var cacheUtils = new CacheUtils var value = cacheUtils.get("DICT:livedata_private_address") return value} 返回的结果为livedata的内网地址。 putCache方法和getCache方法可以实现缓存的使用示例： importClass(com.huawei.livedata.lambdaservice.util.CacheUtils);function execute(data) { var cacheUtils = new CacheUtils code = cacheUtils.putCache("age", "20") if (code != true) { return code } var name = cacheUtils.getCache("age") return name} 返回的结果为20。

使用示例 importClass(com.roma.apic.livedata.client.v1.SoapClient);importClass(com.roma.apic.livedata.config.v1.SoapConfig);importClass(com.roma.apic.livedata.common.v1.XmlUtils);function execute(data) { var soap = new SoapConfig(); soap.setUrl("http://test.webservice.com/ws"); soap.setNamespace("http://spring.io/guides/gs-producing-web-service"); soap.setOperation("getCountryRequest"); soap.setNamespacePrefix("ser"); soap.setBodyPrefix("ser"); soap.setEnvelopePrefix("soapenv"); var content = { "getCountryRequest": { "ser:name": "Spain" }, }; soap.setContent(content); var client = new SoapClient(soap); var result = client.execute(); var body = result.getBody(); return XmlUtils.toJson(body);}

方法列表 返回类型 方法和说明 static boolean putCache(String key, String value) 存入缓存信息 static boolean putCache(String key, String value, int time) 存入带超时时间缓存信息 static String getCache(String key) 获取缓存信息 static long removeCache(String key) 移除缓存信息 static String get(String key) 获取字典缓存信息

方法详情 public static boolean putCache(String key, String value) 存入缓存信息 输入参数 key：缓存信息的key值。 value：缓存的信息。 返回信息 返回对应的boolean值 public static boolean putCache(String key, String value, int time) 存入带超时时间缓存信息 输入参数 key：缓存信息的key值。 value：缓存的信息。 time：超时时间，单位秒。超时后缓存的信息将被删除，获取该缓存信息将返回空值。 返回信息 返回对应的boolean值 public static String getCache(String key) 获取缓存信息 输入参数 key：缓存信息的key值 返回信息 返回key值对应的缓存信息 public static long removeCache(String key) 移除缓存信息 输入参数 key：待移除缓存信息的key值 返回信息 返回执行结果 public static String get(String key) 获取字典缓存信息 输入参数 key：字典缓存信息的key值 返回信息 返回key值对应的字典缓存信息

方法详情 public String get(String key) 返回redis缓存中key对应的value值 输入参数 key：key值 返回信息 redis缓存中key对应的value值 public String put(String key, int expireTime, String value) 更新redis缓存内容、过期时间 输入参数 key：待更新缓存的key值 expireTime：待更新缓存内容的过期时间，单位为秒 value：待更新缓存的value值 返回信息 返回执行结果 public String put(String key, String value) 更新redis缓存内容 输入参数 key：待更新缓存的key值 value：待更新缓存的value值 返回信息 返回执行结果 public Long remove(String key) 删除指定key值的缓存消息 输入参数 key：待删除缓存的key值 返回信息 返回执行结果

方法列表 返回类型 方法和说明 String get(String key) 返回redis缓存中key对应的value值 String put(String key, int expireTime, String value) 更新redis缓存内容、过期时间，返回执行结果 String put(String key, String value) 更新redis缓存内容，返回执行结果 Long remove(String key) 删除指定key值的缓存消息

使用示例 importClass(com.roma.apic.livedata.client.v1.RedisClient);function execute(data) { var redisClient = new RedisClient; var count = redisClient.get("visit_count") if (!count) { redisClient.put("visit_count", 1); }else { redisClient.put("visit_count", parseInt(count) + 1); } return redisClient.get("visit_count");}

方法列表 返回类型 方法和说明 org.apache.kafka.clients.producer.RecordMetadata produce(String topic, String message) 生产消息 不能直接返回方法produce(String topic, String message)，否则会导致返回信息为空。例如在使用示例中，不能直接使用“return record”句式，否则返回的信息为空。

使用示例 importClass(com.roma.apic.livedata.client.v1.KafkaProducer);importClass(com.roma.apic.livedata.config.v1.KafkaConfig);var kafka_brokers = '1.1.1.1:26330,2.2.2.2:26330'var topic = 'YourKafkaTopic'function execute(data) { var config = KafkaConfig.getConfig(kafka_brokers, null) var producer = new KafkaProducer(config) var record = producer.produce(topic, "hello, kafka.") return { offset: record.offset(), partition: record.partition(), code: 0, message: "OK" }}

前提条件 已获取API的调用信息，具体请参见开发准备。 已安装开发工具和Python开发语言环境，具体请参见开发准备。 Python安装完成后，在命令行中使用pip安装“requests”库。 pip install requests 如果pip安装requests遇到证书错误，请下载并使用Python执行此文件，升级pip，然后再执行以上命令安装。 已在IntelliJ IDEA中安装Python插件，如果未安装，请按照图1所示安装。 图1 安装Python插件

调用API示例 在工程中引入apig_sdk。 from apig_sdk import signerimport requestsimport os 生成一个新的Signer，输入API所授权凭据的Key和Secret，可参考获取API的调用信息获取。 sig = signer.Signer()// Directly writing AK/SK in code is risky. For security, encrypt your AK/SK and store them in the configuration file or environment variables. // In this example, the AK/SK are stored in environment variables for identity authentication. Before running this example, set environment variables HUAWEICLOUD_SDK_AK and HUAWEICLOUD_SDK_SK. sig.Key = os.getenv('HUAWEICLOUD_SDK_AK')sig.Secret = os.getenv('HUAWEICLOUD_SDK_SK') 生成一个Request对象，指定方法名、请求url、header和body，可参考获取API的调用信息获取。 r = signer.HttpRequest("POST", "https://c967a237-cd6c-470e-906f-a8655461897e.apigw.exampleRegion.com/app1?a=1", {"x-stage": "RELEASE"}, "body") 进行签名，执行此函数会在请求参数中添加用于签名的X-Sdk-Date头和Authorization头。 X-Sdk-Date是一个必须参与签名的请求消息头参数。 sig.Sign(r) 访问API，查看访问结果。 //若使用系统分配的子 域名 访问https请求的API时，需要在data=r.body后添加“,verify=False”，用来忽略证书校验resp = requests.request(r.method, r.scheme + "://" + r.host + r.uri, headers=r.headers, data=r.body)print(resp.status_code, resp.reason)print(resp.content)