IoT-波bollg个人页面-阿里云开发者社区

阿里云物联网平台C-SDK 4.x版本网关子设备如何上报物模型？

业务场景 1、子设备与子设备、网关与子设备之间通过业务协议进行通信（ZigBee、wifi、蓝牙等）2、网关设备连接物联网平台，创建物理通道，并代理子设备上线创建逻辑通道3、子设备通过逻辑通道与物联网平台进行通信，但是实际不建立任何物理上的连接。原理介绍1、子设备上报物模型，首先得满足网关-子设备架构，先要创建网关和子设备产品关于网关产品和子设备的产品创建这里就不过多赘述。物模型上报，首先肯定得有物模型的定义吧。 2、子设备和网关要建立一种关联关系，建立网关与子设备的topo结构。(这样网关和平台才知道，谁是这个网关的子设备)（1）可以直接在控制台上进行操作（2）可以通过云端API进行操作（3）可以通过SDK的接口进行操作 C-SDK 4.x 接口 components/subdev/aiot_subdev_api.h 3、网关设备创建与物联网平台的链路，再代理子设备进行上线(注意：这些操作都是在网关上完成) 网关设备如同普通直连设备一样，与物联网平台建立连接,即网关-子设备架构中的物理通道代理子设备上线，帮助子设备与物联网平台建立逻辑通道 components/subdev/aiot_subdev_api.h 4、子设备通过网关与物联网平台的物理链路上报消息到物联网平台（注意：这些操作都是在网关上完成，子设备只需要提供三元组信息） *不管是MQTT API还是物模型API，他们的底层原理都是一样的，网关设备和物联网平台进行直连，存在唯一的一条物理通道，而子设备通过网关代理上线，使用逻辑通道与平台进行通信，复用的是网关的物理通道 * 官网说明：可以看到这里有两种方式，分别是基于MQTT API的接口和基于物模型API的接口。那么aiot_mqtt_pub和aiot_dm_send又是如何体现物理通道和逻辑通道呢？(1) MQTT API aiot_mqtt_pub()接口方式 core/aiot_mqtt_api.h 直接调用aiot_mqtt_pub传入带子设备信息的Topic (2) 物模型API aiot_dm_send()接口方式在aiot_dm_send()的两个参数中handle使用网关代理上线后的那个handle，msg携带子设备的信息，完成逻辑通道的消息传输。 components/data-model/aiot_dm_api.h 这个接口里面优先使用的是msg里面携带的product key和devicename来组装topic这个msg结果如下aiot_dm_msg_t 操作步骤1、创建网关产品和子设备的产品&&设备，并建立Topo结构，这里使用控制台方式方便快捷（此篇文章主要介绍子设备的物模型上报）子设备产品定义一个测试用的物模型属性 2、打开C-SDK 4.x Demos demos/subdev_basic_demo.c (1) 修改网关设备三元组信息以及url在demo_mqtt_start()函数中(2)修改子设备的三元组信息全局变量aiot_subdev_dev_t g_subdev[](3)添加demo_subdev_send_property_post接口，上报子设备物模型数据增加头文件包含编写demo_subdev_send_property_post接口 (4) 在main函数中，子设备代理上线成功之后，添加子设备上报物模型逻辑，调用demo_subdev_send_property_post接口 (5)在main函数中，在物模型API上报子设备的物模型逻辑之后，再添加一个通过MQTT API接口上报物模型的逻辑 (6)在C-SDK 4.x主目录 make cd output./subdev-basic-demo 我们可以看到三个子设备的物模型数据都上报成功温度分别是0.9 、1.9、 2.9音量分别是88、89、90 控制台中查看物模型数据：子设备1 子设备2 子设备3 修改后的Demo代码源码：demos/subdev_basic_demo.c /* * 这个例程适用于`Linux`这类支持pthread的POSIX设备, 它演示了...... * * 需要用户关注或修改的部分, 已经用 TODO 在注释中标明 * */ #include <stdio.h> #include <string.h> #include <unistd.h> #include <pthread.h> #include "aiot_state_api.h" #include "aiot_sysdep_api.h" #include "aiot_mqtt_api.h" #include "aiot_subdev_api.h" #include "aiot_dm_api.h" /* 位于portfiles/aiot_port文件夹下的系统适配函数集合 */ extern aiot_sysdep_portfile_t g_aiot_sysdep_portfile; /* 位于external/ali_ca_cert.c中的服务器证书 */ extern const char *ali_ca_cert; static pthread_t g_mqtt_process_thread; static pthread_t g_mqtt_recv_thread; static uint8_t g_mqtt_process_thread_running = 0; static uint8_t g_mqtt_recv_thread_running = 0; //TODO：修改子设备三元组信息 aiot_subdev_dev_t g_subdev[] = { { "a1H****0mgG", "IoT******bDemo1", "e98c465******6dcb35cfa75", "prGp****OjLS7GI" }, { "a1H*****mgG", "IoTDe*****Demo2", "92d719*****67f65bfe4e4d895", "prG*****jLS7GI" }, { "a1H9wUf0mgG", "IoTDeviceSubDemo3", "b2d20a9bb7dbe34e97a4c56f47aed6ee", "prG*****LS7GI" } /* { "a13FN5TplKq", "subdev_basic_demo_01", "768XBgQwgOakz3K4uhOiLeeh9xjJQx6h", "y7GSILD480lBSsP8" }, { "a13FN5TplKq", "subdev_basic_demo_02", "iwTZrbjbgNVChfuJkihjE5asekoyKoYv", "y7GSILD480lBSsP8" }, { "a13FN5TplKq", "subdev_basic_demo_03", "fdutq35iKMYdcWWBuIINY26hsNhgFXWE", "y7GSILD480lBSsP8" }, { "a13FN5TplKq", "subdev_basic_demo_04", "HCKv50YqgwdKhy5cE0Vz4aydmK2ojPvr", "y7GSILD480lBSsP8" } */ }; /* TODO: 如果要关闭日志, 就把这个函数实现为空, 如果要减少日志, 可根据code选择不打印 * * 例如: [1577589489.033][LK-0317] subdev_basic_demo&a13FN5TplKq * * 上面这条日志的code就是0317(十六进制), code值的定义见core/aiot_state_api.h * */ /* 日志回调函数, SDK的日志会从这里输出 */ static int32_t demo_state_logcb(int32_t code, char *message) { printf("%s", message); return 0; } /* MQTT事件回调函数, 当网络连接/重连/断开时被触发, 事件定义见core/aiot_mqtt_api.h */ void demo_mqtt_event_handler(void *handle, const aiot_mqtt_event_t *event, void *userdata) { switch (event->type) { /* SDK因为用户调用了aiot_mqtt_connect()接口, 与mqtt服务器建立连接已成功 */ case AIOT_MQTTEVT_CONNECT: { printf("AIOT_MQTTEVT_CONNECT\n"); /* TODO: 处理SDK建连成功, 不可以在这里调用耗时较长的阻塞函数 */ } break; /* SDK因为网络状况被动断连后, 自动发起重连已成功 */ case AIOT_MQTTEVT_RECONNECT: { printf("AIOT_MQTTEVT_RECONNECT\n"); /* TODO: 处理SDK重连成功, 不可以在这里调用耗时较长的阻塞函数 */ } break; /* SDK因为网络的状况而被动断开了连接, network是底层读写失败, heartbeat是没有按预期得到服务端心跳应答 */ case AIOT_MQTTEVT_DISCONNECT: { char *cause = (event->data.disconnect == AIOT_MQTTDISCONNEVT_NETWORK_DISCONNECT) ? ("network disconnect") : ("heartbeat disconnect"); printf("AIOT_MQTTEVT_DISCONNECT: %s\n", cause); /* TODO: 处理SDK被动断连, 不可以在这里调用耗时较长的阻塞函数 */ } break; default: { } } } /* MQTT默认消息处理回调, 当SDK从服务器收到MQTT消息时, 且无对应用户回调处理时被调用 */ void demo_mqtt_default_recv_handler(void *handle, const aiot_mqtt_recv_t *packet, void *userdata) { switch (packet->type) { case AIOT_MQTTRECV_HEARTBEAT_RESPONSE: { printf("heartbeat response\n"); /* TODO: 处理服务器对心跳的回应, 一般不处理 */ } break; case AIOT_MQTTRECV_SUB_ACK: { printf("suback, res: -0x%04X, packet id: %d, max qos: %d\n", -packet->data.sub_ack.res, packet->data.sub_ack.packet_id, packet->data.sub_ack.max_qos); /* TODO: 处理服务器对订阅请求的回应, 一般不处理 */ } break; case AIOT_MQTTRECV_PUB: { printf("pub, qos: %d, topic: %.*s\n", packet->data.pub.qos, packet->data.pub.topic_len, packet->data.pub.topic); printf("pub, payload: %.*s\n", packet->data.pub.payload_len, packet->data.pub.payload); /* TODO: 处理服务器下发的业务报文 */ } break; case AIOT_MQTTRECV_PUB_ACK: { printf("puback, packet id: %d\n", packet->data.pub_ack.packet_id); /* TODO: 处理服务器对QoS1上报消息的回应, 一般不处理 */ } break; default: { } } } /* 执行aiot_mqtt_process的线程, 包含心跳发送和QoS1消息重发 */ void *demo_mqtt_process_thread(void *args) { int32_t res = STATE_SUCCESS; while (g_mqtt_process_thread_running) { res = aiot_mqtt_process(args); if (res == STATE_USER_INPUT_EXEC_DISABLED) { break; } sleep(1); } return NULL; } /* 执行aiot_mqtt_recv的线程, 包含网络自动重连和从服务器收取MQTT消息 */ void *demo_mqtt_recv_thread(void *args) { int32_t res = STATE_SUCCESS; while (g_mqtt_recv_thread_running) { res = aiot_mqtt_recv(args); if (res < STATE_SUCCESS) { if (res == STATE_USER_INPUT_EXEC_DISABLED) { break; } sleep(1); } } return NULL; } int32_t demo_mqtt_start(void **handle) { int32_t res = STATE_SUCCESS; void *mqtt_handle = NULL; char *url = "iot-as-mqtt.cn-shanghai.aliyuncs.com"; /* 阿里云平台上海站点的域名后缀 */ char host[100] = {0}; /* 用这个数组拼接设备连接的云平台站点全地址, 规则是 ${productKey}.iot-as-mqtt.cn-shanghai.aliyuncs.com */ uint16_t port = 443; /* 无论设备是否使用TLS连接阿里云平台, 目的端口都是443 */ aiot_sysdep_network_cred_t cred; /* 安全凭据结构体, 如果要用TLS, 这个结构体中配置CA证书等参数 */ /* TODO: 替换为自己设备的三元组 */ char *product_key = "a1L****pp"; char *device_name = "Io*****Demo1"; char *device_secret = "68b10******a4b052a197538"; /* 创建SDK的安全凭据, 用于建立TLS连接 */ memset(&cred, 0, sizeof(aiot_sysdep_network_cred_t)); cred.option = AIOT_SYSDEP_NETWORK_CRED_SVRCERT_CA; /* 使用RSA证书校验MQTT服务端 */ cred.max_tls_fragment = 16384; /* 最大的分片长度为16K, 其它可选值还有4K, 2K, 1K, 0.5K */ cred.sni_enabled = 1; /* TLS建连时, 支持Server Name Indicator */ cred.x509_server_cert = ali_ca_cert; /* 用来验证MQTT服务端的RSA根证书 */ cred.x509_server_cert_len = strlen(ali_ca_cert); /* 用来验证MQTT服务端的RSA根证书长度 */ /* 创建1个MQTT客户端实例并内部初始化默认参数 */ mqtt_handle = aiot_mqtt_init(); if (mqtt_handle == NULL) { printf("aiot_mqtt_init failed\n"); return -1; } /* TODO: 如果以下代码不被注释, 则例程会用TCP而不是TLS连接云平台 */ /* { memset(&cred, 0, sizeof(aiot_sysdep_network_cred_t)); cred.option = AIOT_SYSDEP_NETWORK_CRED_NONE; } */ snprintf(host, 100, "%s.%s", product_key, url); /* 配置MQTT服务器地址 */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_HOST, (void *)host); /* 配置MQTT服务器端口 */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_PORT, (void *)&port); /* 配置设备productKey */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_PRODUCT_KEY, (void *)product_key); /* 配置设备deviceName */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_DEVICE_NAME, (void *)device_name); /* 配置设备deviceSecret */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_DEVICE_SECRET, (void *)device_secret); /* 配置网络连接的安全凭据, 上面已经创建好了 */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_NETWORK_CRED, (void *)&cred); /* 配置MQTT默认消息接收回调函数 */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_RECV_HANDLER, (void *)demo_mqtt_default_recv_handler); /* 配置MQTT事件回调函数 */ aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_EVENT_HANDLER, (void *)demo_mqtt_event_handler); /* 与服务器建立MQTT连接 */ res = aiot_mqtt_connect(mqtt_handle); if (res < STATE_SUCCESS) { /* 尝试建立连接失败, 销毁MQTT实例, 回收资源 */ aiot_mqtt_deinit(&mqtt_handle); printf("aiot_mqtt_connect failed: -0x%04X\n", -res); return -1; } /* 创建一个单独的线程, 专用于执行aiot_mqtt_process, 它会自动发送心跳保活, 以及重发QoS1的未应答报文 */ g_mqtt_process_thread_running = 1; res = pthread_create(&g_mqtt_process_thread, NULL, demo_mqtt_process_thread, mqtt_handle); if (res < 0) { printf("pthread_create demo_mqtt_process_thread failed: %d\n", res); aiot_mqtt_deinit(&mqtt_handle); return -1; } /* 创建一个单独的线程用于执行aiot_mqtt_recv, 它会循环收取服务器下发的MQTT消息, 并在断线时自动重连 */ g_mqtt_recv_thread_running = 1; res = pthread_create(&g_mqtt_recv_thread, NULL, demo_mqtt_recv_thread, mqtt_handle); if (res < 0) { printf("pthread_create demo_mqtt_recv_thread failed: %d\n", res); g_mqtt_process_thread_running = 0; pthread_join(g_mqtt_process_thread, NULL); aiot_mqtt_deinit(&mqtt_handle); return -1; } *handle = mqtt_handle; return 0; } int32_t demo_mqtt_stop(void **handle) { int32_t res = STATE_SUCCESS; void *mqtt_handle = NULL; mqtt_handle = *handle; g_mqtt_process_thread_running = 0; g_mqtt_recv_thread_running = 0; pthread_join(g_mqtt_process_thread, NULL); pthread_join(g_mqtt_recv_thread, NULL); /* 断开MQTT连接 */ res = aiot_mqtt_disconnect(mqtt_handle); if (res < STATE_SUCCESS) { aiot_mqtt_deinit(&mqtt_handle); printf("aiot_mqtt_disconnect failed: -0x%04X\n", -res); return -1; } /* 销毁MQTT实例 */ res = aiot_mqtt_deinit(&mqtt_handle); if (res < STATE_SUCCESS) { printf("aiot_mqtt_deinit failed: -0x%04X\n", -res); return -1; } return 0; } void demo_subdev_recv_handler(void *handle, const aiot_subdev_recv_t *packet, void *user_data) { switch (packet->type) { case AIOT_SUBDEVRECV_TOPO_ADD_REPLY: case AIOT_SUBDEVRECV_TOPO_DELETE_REPLY: case AIOT_SUBDEVRECV_TOPO_GET_REPLY: case AIOT_SUBDEVRECV_BATCH_LOGIN_REPLY: case AIOT_SUBDEVRECV_BATCH_LOGOUT_REPLY: case AIOT_SUBDEVRECV_SUB_REGISTER_REPLY: case AIOT_SUBDEVRECV_PRODUCT_REGISTER_REPLY: { printf("msgid : %d\n", packet->data.generic_reply.msg_id); printf("code : %d\n", packet->data.generic_reply.code); printf("product key : %s\n", packet->data.generic_reply.product_key); printf("device name : %s\n", packet->data.generic_reply.device_name); printf("message : %s\n", (packet->data.generic_reply.message == NULL)?("NULL"):(packet->data.generic_reply.message)); printf("data : %s\n", packet->data.generic_reply.data); } break; case AIOT_SUBDEVRECV_TOPO_CHANGE_NOTIFY: { printf("msgid : %d\n", packet->data.generic_notify.msg_id); printf("product key : %s\n", packet->data.generic_notify.product_key); printf("device name : %s\n", packet->data.generic_notify.device_name); printf("params : %s\n", packet->data.generic_notify.params); } break; default: { } } } /* 子设备物模型属性上报函数演示 */ int32_t demo_subdev_send_property_post(void *dm_handle, char *params, char *subdev_product_key,char *subdev_device_name) { aiot_dm_msg_t msg; memset(&msg, 0, sizeof(aiot_dm_msg_t)); msg.type = AIOT_DMMSG_PROPERTY_POST; msg.data.property_post.params = params; msg.product_key = subdev_product_key; msg.device_name = subdev_device_name; return aiot_dm_send(dm_handle, &msg); } int main(int argc, char *argv[]) { int32_t res = STATE_SUCCESS; void *mqtt_handle = NULL, *subdev_handle = NULL; /* 配置SDK的底层依赖 */ aiot_sysdep_set_portfile(&g_aiot_sysdep_portfile); /* 配置SDK的日志输出 */ aiot_state_set_logcb(demo_state_logcb); res = demo_mqtt_start(&mqtt_handle); if (res < 0) { printf("demo_mqtt_start failed\n"); return -1; } subdev_handle = aiot_subdev_init(); if (subdev_handle == NULL) { printf("aiot_subdev_init failed\n"); demo_mqtt_stop(&mqtt_handle); return -1; } aiot_subdev_setopt(subdev_handle, AIOT_SUBDEVOPT_MQTT_HANDLE, mqtt_handle); aiot_subdev_setopt(subdev_handle, AIOT_SUBDEVOPT_RECV_HANDLER, demo_subdev_recv_handler); res = aiot_subdev_send_topo_add(subdev_handle, g_subdev, sizeof(g_subdev)/sizeof(aiot_subdev_dev_t)); if (res < STATE_SUCCESS) { printf("aiot_subdev_send_topo_add failed, res: -0x%04X\n", -res); aiot_subdev_deinit(&subdev_handle); demo_mqtt_stop(&mqtt_handle); return -1; } sleep(2); // aiot_subdev_send_topo_delete(subdev_handle, g_subdev, sizeof(g_subdev)/sizeof(aiot_subdev_dev_t)); // if (res < STATE_SUCCESS) { // printf("aiot_subdev_send_topo_delete failed, res: -0x%04X\n", -res); // aiot_subdev_deinit(&subdev_handle); // demo_mqtt_stop(&mqtt_handle); // return -1; // } // sleep(2); // aiot_subdev_send_topo_get(subdev_handle); // if (res < STATE_SUCCESS) { // printf("aiot_subdev_send_topo_get failed, res: -0x%04X\n", -res); // aiot_subdev_deinit(&subdev_handle); // demo_mqtt_stop(&mqtt_handle); // return -1; // } // sleep(2); // aiot_subdev_send_sub_register(subdev_handle, g_subdev, sizeof(g_subdev)/sizeof(aiot_subdev_dev_t)); // if (res < STATE_SUCCESS) { // printf("aiot_subdev_send_sub_register failed, res: -0x%04X\n", -res); // aiot_subdev_deinit(&subdev_handle); // demo_mqtt_stop(&mqtt_handle); // return -1; // } // sleep(2); // aiot_subdev_send_product_register(subdev_handle, g_subdev, sizeof(g_subdev)/sizeof(aiot_subdev_dev_t)); // if (res < STATE_SUCCESS) { // printf("aiot_subdev_send_product_register failed, res: -0x%04X\n", -res); // aiot_subdev_deinit(&subdev_handle); // demo_mqtt_stop(&mqtt_handle); // return -1; // } // sleep(2); aiot_subdev_send_batch_login(subdev_handle, g_subdev, sizeof(g_subdev)/sizeof(aiot_subdev_dev_t)); if (res < STATE_SUCCESS) { printf("aiot_subdev_send_batch_login failed, res: -0x%04X\n", -res); aiot_subdev_deinit(&subdev_handle); demo_mqtt_stop(&mqtt_handle); return -1; } sleep(2); // aiot_subdev_send_batch_logout(subdev_handle, g_subdev, sizeof(g_subdev)/sizeof(aiot_subdev_dev_t)); // if (res < STATE_SUCCESS) { // printf("aiot_subdev_send_batch_logout failed, res: -0x%04X\n", -res); // aiot_subdev_deinit(&subdev_handle); // demo_mqtt_stop(&mqtt_handle); // return -1; // } /* 子设备上报物模型属性基于物模型API接口*/ for(int i = 0; i < sizeof(g_subdev)/sizeof(aiot_subdev_dev_t) ; i++) { /* *********************************************************************************** * 替换为子设备的ProductKey 和子设备的deviceName */ char *product_key = g_subdev[i].product_key; char *device_name = g_subdev[i].device_name; static float cnt = 0.9; char property_payload[30] = {0}; snprintf(property_payload, sizeof(property_payload), "{\"Temperature\": %f}", cnt++); //HAL_Snprintf(property_payload, sizeof(property_payload), "{\"Temperature\": {\"value\": %d,\"time\":1603248304}}", cnt++); //snprintf(property_payload, sizeof(property_payload), "{\"Temperature\": {\"value\": 77.77,\"time\":1603248304000}}"); demo_subdev_send_property_post(subdev_handle,property_payload,product_key,device_name); //demo_subdev_send_property_post(mqtt_handle,property_payload,product_key,device_name); /*********************************************************************************************/ sleep(1); } /* 子设备上报物模型属性基于MQTT API接口*/ for(int i = 0; i < sizeof(g_subdev)/sizeof(aiot_subdev_dev_t) ; i++) { /* *********************************************************************************** * 替换为子设备的ProductKey 和子设备的deviceName */ char *product_key2 = g_subdev[i].product_key; char *device_name2 = g_subdev[i].device_name; static int cnt2 = 88; char property_payload2[30] = {0}; snprintf(property_payload2, sizeof(property_payload2), "{\"params\":{\"Volume\": %d}}", cnt2++); //HAL_Snprintf(property_payload, sizeof(property_payload), "{\"Temperature\": {\"value\": %d,\"time\":1603248304}}", cnt++); //snprintf(property_payload, sizeof(property_payload), "{\"Temperature\": {\"value\": 77.77,\"time\":1603248304000}}"); char pub_topic[128] = {0}; snprintf(pub_topic,sizeof(pub_topic),"/sys/%s/%s/thing/event/property/post",product_key2,device_name2); res = aiot_mqtt_pub(mqtt_handle, pub_topic, (uint8_t *)property_payload2, (uint32_t)strlen(property_payload2), 0); //res = aiot_mqtt_pub(subdev_handle, pub_topic, (uint8_t *)property_payload2, (uint32_t)strlen(property_payload2), 0); if (res < 0) { printf("aiot_mqtt_sub failed, res: -0x%04X\n", -res); return -1; } /*********************************************************************************************/ sleep(1); } res = aiot_subdev_deinit(&subdev_handle); if (res < STATE_SUCCESS) { printf("aiot_subdev_deinit failed: -0x%04X\n", res); } res = demo_mqtt_stop(&mqtt_handle); if (res < 0) { printf("demo_start_stop failed\n"); return -1; } return 0; }

阿里云物联网平台如何订阅异步服务调用的返回结果？

业务场景 1、云端向设备发送的下行消息或者异步服务调用到平台也算结束，平台再向设备进行一个透传2、设备端向平台发送的上行消息到平台就算结束3、云端通过服务端订阅来获取设备上行的消息原理1、要想获取异步服务调用的返回结果，首先设备得有返回(1) 设备接收平台端透传过来的异步服务调用 (2)设备端收到数据后进行响应2、这里的设备响应结果发送给平台之后，平台可通过云产品流传或服务端订阅，再将消息发送给云端方式一：云平台流转，注意选择Topic:/${productKey}/${deviceName}/thing/downlink/reply/message这个为什么可行？可参考官方文档说明然后添加操作为发布到AMQP消费组方式二:AMQP服务端订阅订阅什么呢？勾选设备上报消息即可（前提是设备端有返回，即满足原理1的前提）操作步骤不多说了，直接上步骤1、准备测试用的产品和设备主要是定义一个异步服务：这里不详细阐述 2、准备测试用的云端调试工具可以是集成云端SDK的Demo，可以是业务逻辑应用调用云端API，最简单的直接使用云端API在线调试工具具体参数填写规范，这里也不做详细阐述 3、物联网平台控制台上配置好规则引擎（1）云平台流转选择好产品设备和topic 注意SQL语句的编写，这里的字段就是要发送给AMQP客户端的消息内容，可以事先进行调试。这里要注意AMQP客户端都是按照既定的协议格式进行过滤数据的，所以这里的消息内容需要按照协议进行配置确定好消息内容后 SQL语句： SELECT timestamp('yyyy-MM-dd\'T\'HH:mm:ss\'Z\'') as 云平台流转至AMQP测试,deviceName() as deviceName, code as code,data as data,topic() as topic,messageId() as requestId,id as id,topic(1) as productKey,iotId as iotId FROM "/a16hDZJpRCl/IoTDeviceDemo1thing/downlink/reply/message" WHERE （2）服务端订阅勾选设备上报消息即可，具体消费组怎么创建就不详细阐述实测效果： 4、设备端接收消息+响应reply 代码示例：pom.xml: <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>org.example</groupId> <artifactId>MQTTClient</artifactId> <version>1.0-SNAPSHOT</version> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <configuration> <source>6</source> <target>6</target> </configuration> </plugin> </plugins> </build> <dependencies> <dependency> <groupId>org.eclipse.paho</groupId> <artifactId>org.eclipse.paho.client.mqttv3</artifactId> <version>1.1.0</version> </dependency> <dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>23.0</version> </dependency> <dependency> <groupId>com.aliyun</groupId> <artifactId>aliyun-java-sdk-core</artifactId> <version>3.5.1</version> </dependency> </dependencies> </project> AliyunIoTSignUtil： package com.alibaba.taro; import javax.crypto.Mac; import javax.crypto.SecretKey; import javax.crypto.spec.SecretKeySpec; import java.util.Arrays; import java.util.Map; /** * AliyunIoTSignUtil */ public class AliyunIoTSignUtil { public static String sign(Map<String, String> params, String deviceSecret, String signMethod) { //将参数Key按字典顺序排序 String[] sortedKeys = params.keySet().toArray(new String[] {}); Arrays.sort(sortedKeys); //生成规范化请求字符串 StringBuilder canonicalizedQueryString = new StringBuilder(); for (String key : sortedKeys) { if ("sign".equalsIgnoreCase(key)) { continue; } canonicalizedQueryString.append(key).append(params.get(key)); } try { String key = deviceSecret; return encryptHMAC(signMethod,canonicalizedQueryString.toString(), key); } catch (Exception e) { throw new RuntimeException(e); } } /** * HMACSHA1加密 * */ public static String encryptHMAC(String signMethod,String content, String key) throws Exception { SecretKey secretKey = new SecretKeySpec(key.getBytes("utf-8"), signMethod); Mac mac = Mac.getInstance(secretKey.getAlgorithm()); mac.init(secretKey); byte[] data = mac.doFinal(content.getBytes("utf-8")); return bytesToHexString(data); } public static final String bytesToHexString(byte[] bArray) { StringBuffer sb = new StringBuffer(bArray.length); String sTemp; for (int i = 0; i < bArray.length; i++) { sTemp = Integer.toHexString(0xFF & bArray[i]); if (sTemp.length() < 2) { sb.append(0); } sb.append(sTemp.toUpperCase()); } return sb.toString(); } } Demo： package com.alibaba; import com.alibaba.taro.AliyunIoTSignUtil; import org.eclipse.paho.client.mqttv3.*; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import org.json.JSONObject; import java.util.HashMap; import java.util.Map; public class CustomTopicMessageDemo2 { public static String productKey = "a16hD*****"; public static String deviceName = "IoTDevice****"; public static String deviceSecret = "0895205d*********"; public static String regionId = "cn-shanghai"; // 物模型-属性上报topic //private static String pubTopic = "/sys/" + productKey + "/" + deviceName + "/thing/event/property/post"; //private static String subTopic = "/sys/" + productKey + "/" + deviceName + "/thing/service/property/set"; // 自定义topic，在产品Topic列表位置定义 //private static String pubTopic = "/"+productKey + "/" + deviceName+"/user/DemoTest"; //private static String subTopic = "/"+productKey + "/" + deviceName+"/user/DemoTest"; private static String pubTopic = "/"+productKey + "/" + deviceName+"/user/get"; private static String subTopic = "/"+productKey + "/" + deviceName+"/user/get"; private static MqttClient mqttClient; public static void main(String [] args){ initAliyunIoTClient(); // ScheduledExecutorService scheduledThreadPool = new ScheduledThreadPoolExecutor(1, // new ThreadFactoryBuilder().setNameFormat("thread-runner-%d").build()); // // scheduledThreadPool.scheduleAtFixedRate(()->postDeviceProperties(), 10,10, TimeUnit.SECONDS); // 汇报属性 //String payloadJson = "{\"params\":{\"MasterLightSwitch\":0,\"LivingLightSwitch\":0,\"SecondaryLightSwotch\":0,\"MasterCurtainSwitch\":1,\"SecondaryCurtainSwitch\":1,\"LivingCurtainSwitch\":1}}"; //String payloadJson = "{\"params\":{\"Temp\":77,\"yyy\":{\"tttt\":\"123\"}}}"; String payloadJson = "{\"params\":{\"Temp\":77,\"yyy\":\"8888\"}}"; //String payloadJson = "{\"tts\":\"ss\"}"; //String payloadJson = "34454545"; postDeviceProperties(payloadJson); try { mqttClient.subscribe(subTopic); // 订阅Topic } catch (MqttException e) { System.out.println("error:" + e.getMessage()); e.printStackTrace(); } // 设置订阅监听 mqttClient.setCallback(new MqttCallback() { @Override public void connectionLost(Throwable throwable) { System.out.println("connection Lost"); } @Override public void messageArrived(String s, MqttMessage mqttMessage) throws Exception { String payload = new String(mqttMessage.getPayload()); System.out.println(" 接收消息："); System.out.println("Topic : " + s); System.out.println(payload); //打印输出消息payLoad System.out.println("================================================================="); // String subTopic = "/sys/" + productKey + "/" + deviceName + "/thing/service/property/set"; // if(s.equals(subTopic)) { // JSONObject jsonProperty = new JSONObject(payload); // if(jsonProperty.has("params")) // { // String paramsJson = jsonProperty.get("params").toString(); // System.out.println("test paramsJson is:\n" + paramsJson); // String params = "{\"params\": " + paramsJson + "}"; // System.out.println("test params is:\n" + params); // System.out.println("收到属性设置后，再上报一次属性："); // postDeviceProperties(params); // } // } //收到服务调用，给予返回reply // 下行（Alink JSON）： // 请求Topic：/sys/{productKey}/{deviceName}/thing/service/{tsl.service.identifier} // 响应Topic：/sys/{productKey}/{deviceName}/thing/service/{tsl.service.identifier}_reply String subTopic = "/sys/" + productKey + "/" + deviceName + "/thing/service/StartP2PStreaming"; String replyTopic = "/sys/" + productKey + "/" + deviceName + "/thing/service/StartP2PStreaming_reply"; if(s.equals(subTopic)) { JSONObject jsonProperty = new JSONObject(payload); if(jsonProperty.has("id")) { String id = jsonProperty.get("id").toString(); String replyJson = "{\"data\":{},\"code\":200,\"id\":\""+ id +"\"}"; //System.out.println("test replyJson is:\n" + replyJson); //String replys = "{\"params\": " + replyJson + "}"; //System.out.println("test reply is:\n" + replys); System.out.println("收到服务调用后，给予返回"); postServiceReply(replyJson,replyTopic); } } } @Override public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) { } }); } /** * 初始化 Client 对象 */ private static void initAliyunIoTClient() { try { // 构造连接需要的参数 String clientId = "java" + System.currentTimeMillis(); Map<String, String> params = new HashMap<String, String>(16); params.put("productKey", productKey); params.put("deviceName", deviceName); params.put("clientId", clientId); String timestamp = String.valueOf(System.currentTimeMillis()); params.put("timestamp", timestamp); // cn-shanghai String targetServer = "tcp://" + productKey + ".iot-as-mqtt."+regionId+".aliyuncs.com:1883"; String mqttclientId = clientId + "|securemode=3,signmethod=hmacsha1,timestamp=" + timestamp + "|"; String mqttUsername = deviceName + "&" + productKey; String mqttPassword = AliyunIoTSignUtil.sign(params, deviceSecret, "hmacsha1"); connectMqtt(targetServer, mqttclientId, mqttUsername, mqttPassword); } catch (Exception e) { System.out.println("initAliyunIoTClient error " + e.getMessage()); } } public static void connectMqtt(String url, String clientId, String mqttUsername, String mqttPassword) throws Exception { MemoryPersistence persistence = new MemoryPersistence(); mqttClient = new MqttClient(url, clientId, persistence); MqttConnectOptions connOpts = new MqttConnectOptions(); // MQTT 3.1.1 connOpts.setMqttVersion(4); connOpts.setAutomaticReconnect(false); connOpts.setCleanSession(false); //connOpts.setCleanSession(true); connOpts.setUserName(mqttUsername); connOpts.setPassword(mqttPassword.toCharArray()); connOpts.setKeepAliveInterval(60); mqttClient.connect(connOpts); } /** * 汇报属性 */ private static void postDeviceProperties(String payloadJson) { try { //上报数据 //高级版物模型-属性上报payload System.out.println("上报属性值:"); //String payloadJson = "{\"params\":{\"Status\":0,\"Data\":\"15\"}}"; //String payloadJson = "{\"GeoLocation\":{\"Longitude\":120.99,\"Latitude\":30.13,\"Altitude\":39.01},\"BatteryPercentage\":40.703533, \"Temperature\":2.233362}"; //String payloadJson = "{\"id\":\"3\",\"version\":\"1.0\",\"params\":{\"GeoLocation\":{\"Longitude\":120.999,\"Latitude\":30.13,\"Altitude\":39.01},\"BatteryPercentage\":42.99999, \"Temperature\":2.233362}}"; //String payloadJson = "{\"params\":{\"MasterLightSwitch\":0,\"LivingLightSwitch\":0,\"SecondaryLightSwotch\":0,\"MasterCurtainSwitch\":1,\"SecondaryCurtainSwitch\":1,\"LivingCurtainSwitch\":1}}"; System.out.println(payloadJson); MqttMessage message = new MqttMessage(payloadJson.getBytes("utf-8")); message.setQos(0); mqttClient.publish(pubTopic, message); System.out.println("================================================================="); } catch (Exception e) { System.out.println(e.getMessage()); } } /** * 服务返回 */ private static void postServiceReply(String payloadJson,String relpyTopic) { try { //上报数据 //高级版物模型-属性上报payload System.out.println("服务调用返回:"); //String payloadJson = "{\"params\":{\"Status\":0,\"Data\":\"15\"}}"; System.out.println("Topic:"); System.out.println(relpyTopic); System.out.println(payloadJson); MqttMessage message = new MqttMessage(payloadJson.getBytes("utf-8")); message.setQos(0); mqttClient.publish(relpyTopic, message); System.out.println("================================================================="); } catch (Exception e) { System.out.println(e.getMessage()); } } } 实测效果： 5、云端使用AMQP客户端登录，并接收消息参考官方文档，这里就不作详细阐述。https://help.aliyun.com/document_detail/143601.html?spm=a2c4g.11186623.6.624.304e354e2OEGFh 代码示例：pom.xml <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>org.example</groupId> <artifactId>Test</artifactId> <version>1.0-SNAPSHOT</version> <dependencies>        <dependency> <groupId>commons-codec</groupId> <artifactId>commons-codec</artifactId> <version>1.10</version> </dependency>  <dependency> <groupId>org.apache.qpid</groupId> <artifactId>qpid-jms-client</artifactId> <version>0.47.0</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-simple</artifactId> <version>1.7.25</version> <scope>compile</scope> </dependency> <dependency> <groupId>org.apache.maven.surefire</groupId> <artifactId>maven-surefire-common</artifactId> <version>2.12.4</version> </dependency>       </dependencies> </project> Demo package com.alibaba; import org.apache.commons.codec.binary.Base64; import org.apache.qpid.jms.JmsConnection; import org.apache.qpid.jms.JmsConnectionListener; import org.apache.qpid.jms.message.JmsInboundMessageDispatch; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.crypto.Mac; import javax.crypto.spec.SecretKeySpec; import javax.jms.*; import javax.naming.Context; import javax.naming.InitialContext; import java.net.URI; import java.util.Hashtable; import java.util.concurrent.ExecutorService; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; public class AmqpJavaClientDemo { private final static Logger logger = LoggerFactory.getLogger(AmqpJavaClientDemo.class); //业务处理异步线程池，线程池参数可以根据您的业务特点调整，或者您也可以用其他异步方式处理接收到的消息。 private final static ExecutorService executorService = new ThreadPoolExecutor( Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(50000)); public static void main(String[] args) throws Exception { //参数说明，请参见AMQP客户端接入说明文档。 String accessKey = "LTAI4G2*****"; String accessSecret = "Mp2f4qopmULI6*****"; String consumerGroupId = "e0oRIYMSOYwQ*****"; //iotInstanceId：购买的实例请填写实例ID，公共实例请填空字符串""。 String iotInstanceId = ""; long timeStamp = System.currentTimeMillis(); //签名方法：支持hmacmd5、hmacsha1和hmacsha256。 String signMethod = "hmacsha1"; //控制台服务端订阅中消费组状态页客户端ID一栏将显示clientId参数。 //建议使用机器UUID、MAC地址、IP等唯一标识等作为clientId。便于您区分识别不同的客户端。 String clientId = "yangboClientId"; //userName组装方法，请参见AMQP客户端接入说明文档。 String userName = clientId + "|authMode=aksign" + ",signMethod=" + signMethod + ",timestamp=" + timeStamp + ",authId=" + accessKey + ",iotInstanceId=" + iotInstanceId + ",consumerGroupId=" + consumerGroupId + "|"; //计算签名，password组装方法，请参见AMQP客户端接入说明文档。 String signContent = "authId=" + accessKey + "&timestamp=" + timeStamp; String password = doSign(signContent,accessSecret, signMethod); //接入域名，请参见AMQP客户端接入说明文档。 String connectionUrl = "failover:(amqps://1875496626634053.iot-amqp.cn-shanghai.aliyuncs.com:5671?amqp.idleTimeout=80000)" + "?failover.reconnectDelay=30"; Hashtable<String, String> hashtable = new Hashtable<String, String>(); hashtable.put("connectionfactory.SBCF",connectionUrl); hashtable.put("queue.QUEUE", "default"); hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory"); Context context = new InitialContext(hashtable); ConnectionFactory cf = (ConnectionFactory)context.lookup("SBCF"); Destination queue = (Destination)context.lookup("QUEUE"); // Create Connection Connection connection = cf.createConnection(userName, password); ((JmsConnection) connection).addConnectionListener(myJmsConnectionListener); System.out.println("connection success"); // Create Session // Session.CLIENT_ACKNOWLEDGE: 收到消息后，需要手动调用message.acknowledge()。 // Session.AUTO_ACKNOWLEDGE: SDK自动ACK（推荐）。 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); connection.start(); // Create Receiver Link MessageConsumer consumer = session.createConsumer(queue); consumer.setMessageListener(messageListener); } private static MessageListener messageListener = new MessageListener() { @Override public void onMessage(Message message) { try { //1.收到消息之后一定要ACK。 // 推荐做法：创建Session选择Session.AUTO_ACKNOWLEDGE，这里会自动ACK。 // 其他做法：创建Session选择Session.CLIENT_ACKNOWLEDGE，这里一定要调message.acknowledge()来ACK。 // message.acknowledge(); //2.建议异步处理收到的消息，确保onMessage函数里没有耗时逻辑。 // 如果业务处理耗时过程过长阻塞住线程，可能会影响SDK收到消息后的正常回调。 executorService.submit(new Runnable() { public void run() { processMessage(message); } }); } catch (Exception e) { logger.error("submit task occurs exception ", e); } } }; /** * 在这里处理您收到消息后的具体业务逻辑。 */ private static void processMessage(Message message) { try { byte[] body = message.getBody(byte[].class); String content = new String(body); String topic = message.getStringProperty("topic"); String messageId = message.getStringProperty("messageId"); System.out.println("receive message" + ", topic = " + topic + ", messageId = " + messageId + ", content = " + content); logger.info("receive message" + ", topic = " + topic + ", messageId = " + messageId + ", content = " + content); } catch (Exception e) { logger.error("processMessage occurs error ", e); } } private static JmsConnectionListener myJmsConnectionListener = new JmsConnectionListener() { /** * 连接成功建立。 */ @Override public void onConnectionEstablished(URI remoteURI) { logger.info("onConnectionEstablished, remoteUri:{}", remoteURI); } /** * 尝试过最大重试次数之后，最终连接失败。 */ @Override public void onConnectionFailure(Throwable error) { logger.error("onConnectionFailure, {}", error.getMessage()); } /** * 连接中断。 */ @Override public void onConnectionInterrupted(URI remoteURI) { logger.info("onConnectionInterrupted, remoteUri:{}", remoteURI); } /** * 连接中断后又自动重连上。 */ @Override public void onConnectionRestored(URI remoteURI) { logger.info("onConnectionRestored, remoteUri:{}", remoteURI); } @Override public void onInboundMessage(JmsInboundMessageDispatch envelope) {} @Override public void onSessionClosed(Session session, Throwable cause) {} @Override public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {} @Override public void onProducerClosed(MessageProducer producer, Throwable cause) {} }; /** * 计算签名，password组装方法，请参见AMQP客户端接入说明文档。 */ private static String doSign(String toSignString, String secret, String signMethod) throws Exception { SecretKeySpec signingKey = new SecretKeySpec(secret.getBytes(), signMethod); Mac mac = Mac.getInstance(signMethod); mac.init(signingKey); byte[] rawHmac = mac.doFinal(toSignString.getBytes()); return Base64.encodeBase64String(rawHmac); // return Arrays.toString(Base64.encodeBase64(rawHmac)); } } 实测效果：云平台流转方式的返回结果： AMQP订阅方式返回结果：

IoT-波bollg个人页面-阿里云开发者社区

个人介绍

擅长的技术

阿里云物联网平台之Iot Studio中的web应用域名绑定

阿里云物联网平台之云端API调用(即云端开发)

阿里云物联网平台 “物模型属性” 的分析&&易错点&&上报属性时Payload如何正确组装？

阿里云物联网平台C-SDK 4.x版本网关子设备如何上报物模型？

阿里云物联网平台如何订阅异步服务调用的返回结果？

阿里云物联网平台C-SDK 4.x OTA升级示例

阿里云物联网平台之Iot Studio中的web应用域名绑定

阿里云物联网平台之利用云平台流转如何实现同一款产品下任意俩个设备的通信？

阿里云物联网平台之云端API调用(即云端开发)

阿里云物联网平台 “物模型属性” 的分析&&易错点&&上报属性时Payload如何正确组装？

阿里云物联网平台C-SDK 4.x版本网关子设备如何上报物模型？

阿里云物联网平台如何订阅异步服务调用的返回结果？

阿里云物联网平台C-SDK 4.x OTA升级示例

阿里云物联网平台之Iot Studio中的web应用域名绑定

阿里云物联网平台之利用云平台流转如何实现同一款产品下任意俩个设备的通信？

阿里云物联网平台之云端API调用(即云端开发)

阿里云物联网平台 “物模型属性” 的分析&&易错点&&上报属性时Payload如何正确组装？

阿里云物联网平台C-SDK 4.x版本网关子设备如何上报物模型？

阿里云物联网平台如何订阅异步服务调用的返回结果？