简介：面向中国电信物联网开放平台及天翼NB-IoT应用开发者的API Demo例程，用于快速验证北向RESTful接口的开放能力，帮助使用者体验业务功能并熟悉整体流程。Demo核心覆盖应用鉴权、设备管理、消息订阅、数据采集、命令下发、批量任务六大模块，基本对应物联网平台北向接入中的典型操作场景。资源包大小约4.59MB，内部主要包含src源码目录与Open source components依赖jar包：前者按功能模块组织可直接运行的示例代码，方便跟踪调用链路；后者汇总工程所需的第三方组件，减少环境配置时间。整体目录结构清晰，便于按需查阅。目前已有1389人学习，适合物联网平台对接初期的工程师作为参考。通过本Demo，读者可以直观理解北向接口的请求构造、鉴权参数、回调订阅及返回数据解析方式，并在此基础上扩展自己的业务逻辑，从而降低平台联调门槛，加速功能验证与落地。
1. 一套跑得通的 API Demo 例程，能帮你省掉大半踩坑时间
一个刚从 MCU 开发转过来的工程师，拿到一批物联网模组之后的第一件事，往往不是写驱动，而是翻接入文档。文档厚得跟字典一样，真正能让设备连上平台的核心逻辑，其实都浓缩在一份示例代码里。标题里说的“运营商物联网开放平台 API Demo 例程”，指的就是平台提供给设备开发者的接入示例：一个拿得到、跑得通的最小工程，能把设备注册、数据上行、命令下行这三件事完整串起来。
这套例程的价值不在代码写得多优雅，而在于把一个平台的连接细节显式化：MQTT 接入地址怎么拼、密码是不是要做二次变换、topic 怎么命名、物模型字段对类型的要求有多严格。只要它跑通了，你的设备就拿到了进入平台的合法入场券；后续换成自己的业务逻辑，只是工作量问题。适合它的人也很明确：一类是做设备端固件的嵌入式工程师，要把模组快速跑进平台；另一类是系统集成方，需要通过例程来理解整个平台的 API 边界，才能设计好自己的云端业务。
2. 动手之前先拆清楚：Demo 背后的 API 体系
2.1 设备接入类 API 和北向 OpenAPI 是两套东西
不少人在初看这个方向时，会默认“API 就是 HTTP 接口”。实际上这里藏着最容易看走眼的分岔路。
第一类是设备接入类 API。这类 API 不以 REST 风格出现在业务系统面前，而是以协议形式作用于设备本身。设备用 MQTT、LWM2M、TCP 或 HTTP 与平台通信，对接时需要对上的信息包括服务地址、端口、topic、消息体里的数据模型。平台通常把这部分写进《设备接入协议》，里面讲的是“设备往哪个 topic 发什么、字段叫什么、ACK 怎么回”。API Demo 里的设备侧工程，就是按这套协议写出来的最小实现。
第二类是平台开放 OpenAPI，也就是通常说的北向 API。你云端的业务系统要主动获取设备状态，或者给设备下发命令，就得用你的业务服务器去调用这些 REST 接口。典型流程是“先登录拿 access token，再带着 token 调用设备查询、命令下发”。控制台上能干的大部分操作，OpenAPI 也都提供了一遍。Demo 里的服务端工程或者一组演示脚本，通常就是做这个调用流程的样例。
第三类是消息订阅。平台在设备产生事件时，根据你在控制台配置的订阅地址，把数据主动推送到你的服务器，类似 webhook。设备侧 Demo 往往不会把订阅后端写全，但会在文档里给一个“如何订阅、如何验签、如何回 ACK”的示例。
把这三类分开之后，你才不会拿着设备侧 Demo 的代码去问“为什么我用 HTTP 调用不了”。设备接入协议解决的是“设备怎么说话”，北向 OpenAPI 解决的是“业务系统怎么问平台”，消息订阅解决的是“平台怎么把话推送给你”。三者鉴权方式、超时语义、字段类型都不一样，混在一起调一定会翻车。
2.2 Demo 帮你串起来的三条主链路
一份能用的接入 Demo，不管用什么语言写，都绕不开三条链路。我拿到例程后的习惯，是按这个顺序先对齐。
第一是设备认证链路。设备在控制台的“产品-设备”里被创建，得到一个设备 ID 和一个设备密钥。设备接入平台时，MQTT 参数会要三样东西：clientId、username、password。这三样不是随手填的，平台会按统一规则校验“这个设备是否存在、密码是否正确、是否有权限接入”。Demo 例程最容易踩坑的地方在这：设备密钥并不总是作为 MQTT 密码直接使用，可能还有一层算法变换。平台官方提供“MQTT 接入参数生成工具”，会把可用的接入三件套算出来，你直接拿去用就能通。
第二是属性上报链路。设备连上平台之后，把采集数据写成 JSON，发布到平台定义的属性上报 topic。平台侧已经按“物模型”把产品的数据结构定好了，Demo 里一般会给一个默认属性，比如温度、湿度，提示你后续换业务属性时，字段名和类型必须跟物模型对齐。
第三是命令下行链路。平台给设备下发命令，在设备端体现为一条 MQTT 消息。设备收到之后，要按协议回执，平台才能把这次下发的状态更新为“命令成功”。这一条是 Demo 里最容易被忽略的，很多示例只打印收到的命令，没有走完“读取、执行、回执”的闭环。
这三条链路只要有一条断了，排查方向就完全不同：认证问题是连接阶段，上报问题是消息解析，下行问题是回执状态机。所以我看 Demo 的第一件事，就是找出这三段各自对应的代码位置。
2.3 一机一密和一型一密，创建产品前就要选好
接入认证方式，在创建产品时就得决定，因为它直接影响后面所有 Demo 部署方式。平台里最常见的就是一机一密和一型一密两种。
一机一密，每一台设备在出厂前有自己唯一的密钥，设备 ID 和密钥是一对一关系。设备遗失、报废、需要重置时，单独处理这一台就行，不影响其他设备，安全隔离性最好。代价是密钥管理成本高，适合行业设备和企业资产，比如水表、电表、物流定位器，数量相对可控，客户也需要知道每台设备的明细。
一型一密，同一产品下的设备共用一套产品级密钥。设备首次接入时，用自己的设备 ID 加产品密钥，动态申请一个当次接入凭证。好处是批量发货不用逐台烧录独立密钥，生产效率高；坏处是产品密钥一旦泄露，同型号设备可能被仿冒接入。
我通常在两种之间这样取舍：出货量不高、单品价值高、客户会单独维护设备的，选一机一密；海量消费品、后续有大量首次激活场景的，选一型一密，同时用好平台侧的产物密钥管理能力。两种方式的差异可以看这张表：
| 对比项 | 一机一密 | 一型一密 |
|---|---|---|
| 密钥分布 | 每台设备独立 | 同型号共用产品级密钥 |
| 批量生产 | 逐台写入 | 统一烧录，激活效率高 |
| 单台处置 | 可单独吊销重置 | 重置方便，但要防止产品密钥泄露 |
| 安全风险 | 单点泄露影响小 | 产品密钥泄露影响面大 |
选型完了再跑 Demo，别等设备和数据都建好了再想换。那种返工不仅是代码改动，产品和设备都要重建，属于典型的后悔药场景。
3. 在自己电脑上把 Demo 最小闭环跑通
3.1 建产品、加设备，先把三组关键信息拎出来
动手第一步，是把控制台里的地基打牢。登录平台控制台，进入“产品开发”模块，新建一个产品。需要注意的字段通常有四个：产品分类，按实际场景选智能家居或工业传感；接入协议，MQTT 场景居多，NB 模组也有走 LWM2M 的情况；认证方式，按上一节的一机一密或一型一密选择；联网方式，按 SIM 卡类型和模组制式选。
建好产品之后，进入物模型定义。一般至少添加一个属性和一个事件。属性可以叫 temperature，类型选小数，范围设 -50 到 150；事件可以设一个“告警上报”。物模型保存后，再进“设备列表”创建设备。创建设备会返回设备 ID 和设备密钥，这三组信息是要写进 Demo 配置文件的：
| 信息名称 | 从哪里拿 | 作用 |
|---|---|---|
| 产品 ID | 产品基本信息页 | 组成 MQTT clientId 和 topic 前缀 |
| 设备 ID | 设备详情页 | 设备唯一标识，也是 MQTT 用户名的一部分 |
| 设备密钥 | 设备详情页，一机一密 | 作为接入凭证的原始材料 |
设备详情页里一般还会有一个“接入参数生成器”。点开之后，平台会直接算出一组完整的 MQTT clientId、username、password 和接入地址。把这组数据跟 Demo 配置文件里的对应值对一遍，就能看出代码里哪一步做了二次处理，哪一步需要替换成你自己的。
提示：接入地址通常带区域段信息，不要把它理解成固定的官方域名来死记。以控制台生成的接入点为准，换区域运营时地址前缀会变。
3.2 用 Python 把设备接入和数据上报跑通
常见的验证方式是用 Python 跑 MQTT Demo。设备侧工程一般会带一份基于 paho-mqtt 的示例代码，我习惯先把平台工具生成的接入三件套固定填进去跑一次，确认通路没问题，再改回代码自动生成。下面这段是示意写法，用来对照理解整条链路：
import json import hashlib import paho.mqtt.client as mqtt # 从控制台取得的三个基础信息 PRODUCT_ID = "your_product_id" DEVICE_ID = "your_device_id" DEVICE_KEY = "your_device_key" # 注意：多数平台的 MQTT password 不是 DEVICE_KEY 明文， # 而是由平台工具生成的动态凭证。hashlib 这行只是结构示意， # 真正要用的值，请以控制台“接入参数生成器”的输出为准。 mqtt_password = hashlib.sha256(DEVICE_KEY.encode("utf-8")).hexdigest() broker = "mqtt.{region}.{platform_domain}" # 以控制台接入信息为准 port = 1883 client = mqtt.Client(client_id=f"{PRODUCT_ID}_{DEVICE_ID}") client.username_pw_set(DEVICE_ID, mqtt_password) def on_connect(client, userdata, flags, rc): if rc == 0: print("connect success") # 等连接成功回调后再发布，避免消息在 TCP 会话建立期就丢包 payload = { "temperature": 26.8, "humidity": 60.0 } topic = f"/{PRODUCT_ID}/{DEVICE_ID}/thing/property/post" client.publish(topic, json.dumps(payload), qos=0) else: print("connect fail, rc =", rc) client.on_connect = on_connect client.connect(broker, port, keepalive=60) client.loop_forever()逻辑说明：client_id 使用“产品ID_设备ID”的组合，这是很多 MQTT 平台验证设备归属的通用做法。如果你手里的平台在接入文档里给出的 clientId 格式不是这个，必须以控制台生成的值为准。topic 里的thing/property/post也只是一种常见样式，不同产品协议版本可能写成properties/report或者v1/devices/...，所以复制到正式工程前，一定要从控制台“设备 topic 列表”那一页复制真实值。
参数说明：keepalive=60表示每 60 秒发一次心跳包。如果设备是电池供电，可以按平台允许范围把心跳拉长到 300 秒，或者使用平台的低功耗模式；但也不能调得太长，否则运营商网络侧的连接映射会老化，造成“设备在线但实际失联”的假象。qos=0适合高频采集、丢一两条无所谓的场景；如果是告警或命令类消息，建议改qos=1，并在发布回执返回后再认为消息投递成功。
运行后看到connect success，再回控制台看设备状态是否变为“在线”，这一步就算通了。
3.3 命令下行：收到指令和执行完成是两回事
设备侧 Demo 里最常见的缺口，是只打印下行命令，不回复。平台统一下发的命令消息，payload 大概长这样：
{ "commandId": "780a3f20-xxxx-xxxx-xxxx", "params": { "switch": 1 } }设备处理完业务之后，要发布一条命令响应消息，并且带上同一个 commandId。示例代码如下：
# 放在 mqtt 的 on_message 回调里 def on_message(client, userdata, msg): print("recv:", msg.payload.decode()) cmd = json.loads(msg.payload) response_topic = f"/{PRODUCT_ID}/{DEVICE_ID}/thing/service/property/set_reply" ack_payload = { "commandId": cmd["commandId"], "code": 0, "message": "success" } client.publish(response_topic, json.dumps(ack_payload), qos=1) print("replied ack")逻辑说明：这里回执的语义是“成功收到命令并已执行完成”。如果业务动作是打开继电器、切换视频流，一定要在真正执行完之后再回这个值，不要提前回执。否则平台会把命令标记成“成功”，而设备端实际执行失败，两边都看不出问题，这就成了黑匣子。部分平台还把“收到命令”和“执行完成”拆成两个不同 topic，需要分别回复两次，具体以协议文档里的 commandId 状态机为准。
参数说明：响应消息里的code字段，不同平台可能叫result、status或error_code，命名要以协议定义为准。响应 topic 是否需要带产品 ID 前缀、是否需要带设备 ID，也要对照协议文档，不能凭 Demo 里的上下文推断着用。
4. 避坑：从 Demo 到生产最容易翻车的 5 个问题
4.1 设备一直认证失败，密码大概率不是发下来的那个
现象：MQTT 客户端反复报Connection Refused: not authorized，把控制台上展示的设备密钥原样填进 password，依然连不上。
原因：不少平台的 MQTT 接入密码不是设备密钥明文，而是由“设备密钥 + 时间戳 + 产品维度算法”推导出来的动态凭证。控制台上展示的是“设备密钥”，是生成 MQTT 密码的原料，不是密码本身。Demo 里的生成函数如果版本落后，或者密钥体系更新过，就会对不上。
解决：先不要和代码里的算法纠缠。去控制台设备详情页，打开“MQTT 接入参数生成工具”，把生成的三件套直接填入连接参数试跑。跑通之后，再回头比对这个工具的输出和 Demo 算法结果是否一致，决定改哪边。这样能省掉一整晚的瞎试。
4.2 上报成功但控制台看不到数据，先查类型和 topic
现象：publish 调用没有报错，服务端也收到了上行消息，但控制台的产品数据页里就是没有最新属性。
原因：一是属性名和物模型定义不一致，比如模型里是temperature，代码里写成temp；二是类型不一致，模型定义整数类型，代码上传了浮点数；三是发布 topic 是自己拼的，平台根本不做物模型解析。
解决：按三个步骤排查。第一，去“设备日志”或“消息追踪”页面，打开原始上报消息，看服务端实际收到的 JSON 是什么。第二，把这个 JSON 和产品物模型导出的定义做逐字段对比，键名、类型、单位都不能错。第三，直接用控制台的“在线调试”功能改发一条属性消息，能成功，再回来对比代码里的 topic。这套流程走完，多数“看不到数据”的问题都能定位。
4.3 设备在线但命令总超时，因为你只回了第一个 ACK
现象：设备端 MQTT 日志里已经打印出下行的命令内容，但控制台的命令记录一直显示“超时”。
原因：平台对“命令下发成功”的判定，通常要看最终的执行结果，而不是 TCP 层的消息确认。你只回了一条“已收到”，却没有把“执行完成”的结果发上去，平台等到超时时间就判失败。
解决：确认设备在完成真实业务动作后，把带 commandId 的最终执行结果发出。如果协议要求“收到”和“完成”两次回执，就分两个状态分别发布。业务执行出现异常时，不要硬回成功码，要回一个失败码，让平台知道命令执行失败，由业务系统决定是否重试。
4.4 时间戳校验反复失败，先校准所有设备时钟
现象：接入日志里报“签名已过期”或“timestamp invalid”，但代码逻辑没有改过，昨天还能连上。重跑一次又成功，设备一多就零散掉线。
原因：平台的接入凭证通常带时间窗口，只要设备本地 RTC 和服务器时间偏差超过允许范围，就会被判定无效。PC 上跑不容易撞上，因为桌面系统会自动同步时间；换到电池型模组或者刚补电的开发板上，RTC 恢复默认时间就会出问题。
解决：在正式工程里，把“网络时间已同步”作为接入前置检查。设备启动时先用 NTP 或运营商网络对一次时，确保本地时间落在平台允许的偏差窗口内。如果产品形态不支持校时，看平台是否提供免时间戳的接入方式；不支持的话，只能定期让云端下发标准时间覆盖设备时间。这不是玄学，就是最基础的时间同步问题。
注意：TLS 方式接入时，如果证书握手失败，命令行里一般会打印
handshake failure。要先确认模组固件里带的是否是完整的根证书链，再谈网络问题。
4.5 从 PC 环境换到真机就连不上，先怀疑模组和网络
现象：同样的代码、同样的接入参数，在 PC 上轻松连接，烧到开发板上却 connect timeout 或者握手失败。
原因：真机网络路径比 PC 复杂。NB 信号弱、模组 APN 配置不完整、设备使用省电模式长时间断连，都会导致连接建立缓慢。启用 TLS 时，模组固件证书不全也会失败，而 PC 端 Python 会自动加载系统证书，所以差别一下就出来了。
解决：按顺序排查。先用 1883 明文端口验证能连通，再换 8883 加密端口。核对模组的 APN 是否匹配当前制式，检查设备心跳间隔和平台的在线判断逻辑是否匹配。如果有低功耗策略，评估“上传后立即休眠”的模式是否真的适合你的产品。换用模组厂商提供的 AT 指令连接脚本，逐步抓日志，这一步不能跳过。PC 跑通只是起点，真机多花一小时，上线后少熬夜一次。
5. 从 Demo 到业务系统：把北向 OpenAPI 串起来
5.1 为什么不能只依赖设备侧 Demo
设备侧 Demo 解决的是“设备能把数据存到平台”。但你的产品通常是一个完整业务系统：运维后台要读设备温度历史，用户 App 要开关设备，告警系统要收到离线通知。这些功能不可能让人打开控制台逐台点。
所以需要在业务系统里引入平台提供的北向 OpenAPI。从控制台里找到 API 文档，在代码里加上“获取 token”这一环，就能把平台当成后端来接：查询设备列表、查设备详情、下发命令、删除设备。这一切用代码完成，比人工操作控制台高效，也更安全。
5.2 获取访问令牌并调用查询设备接口
北向调用流程一般是先认证、再访问资源。下面是一段常用的 Python 请求示意：
import requests app_key = "your_app_key" app_secret = "your_app_secret" base_url = "https://openapi.{platform_domain}" # 第 1 步：换取访问令牌 r = requests.post( f"{base_url}/v1/token", json={"appId": app_key, "appSecret": app_secret}, timeout=10 ) token_data = r.json().get("data") or {} token = token_data.get("accessToken", "") print("token:", token[:12] + "...") # 第 2 步：用令牌查设备状态 headers = {"Authorization": f"Bearer {token}"} resp = requests.get( f"{base_url}/v1/devices", params={"deviceId": "your_device_id"}, headers=headers, timeout=10 ) print(resp.json())逻辑说明：accessToken 是有有效期的，通常只有几个小时。业务系统应该把 token 缓存到内存或 Redis 里，快过期时再统一刷新，而不是每个请求都重新登录，否则很容易触发平台的调用频控限制。设备查询接口正常会返回在线状态、最近在线时间、固件版本这些信息，足够业务侧做基础展示和判断。
参数说明：appId/appSecret来自控制台“应用管理-新建应用”，不是控制台的登录账号密码。这两个值要放在服务端保存，绝对不能写进前端代码。新建应用一般默认没有其他产品的访问权限，记得在控制台把目标产品授权给这个应用，否则接口会直接报权限错误。
5.3 开放 API 和消息订阅怎么配合
单纯用北向 OpenAPI 定时查询设备状态，会有两个问题：查太频繁浪费配额，查太慢实时性不够。所以平台一般还提供消息订阅能力。到控制台配置一个 HTTPS 订阅回调地址，平台会在设备上报、离线、告警等事件发生时，把消息推送到你的服务器。
我实际项目里的搭配方法是：设备实时告警走消息订阅，后台立刻感知；业务页面展示历史数据和曲线走 OpenAPI 查询；需要主动下发的控制指令走 OpenAPI。两套机制不是替代关系，而是事件驱动与按需读取之间的互补。
做消息订阅要注意：平台推送到你的回调地址时，一般会带签名。你不能直接把收到的 JSON 盲目入库，必须先验签。回调接口收到消息后，要尽快返回 HTTP 200，直接把消息投进消息队列，异步去处理业务。如果同步处理耗时过长，平台可能在超时后重复推送，造成重复消费。把“接收、验签、入队、异步处理、落库查询”这个链路写顺了，业务系统才算真正接稳平台。只靠设备侧 Demo，你在实时性和可维护性上会一直差一步。
6. 把 Demo 改造成回归工具：上线前的最后一层防护
Demo 跑通一次，不代表环境一直稳定。我习惯把整个接入链路自动化成一个回归脚本，每次固件版本发布前先跑一遍。回归动作只有三个：设备登录一次；上报一组测试属性；等待命令下行并确认收到回执。全部通过再打包发布，失败就立刻定位是平台侧变更、证书过期，还是固件改了字段类型。
我写过的最小回归逻辑大概是这样：
def regression_round(): assert mqtt_login() is True assert report_property({"temperature": 25.0}) is True cmd = wait_command(timeout=30) assert cmd["commandId"] is not None assert reply_command(cmd["commandId"]) is True这里的timeout=30是给命令下行留出的等待窗口：太短容易误报，太长会拖慢回归节奏。运行方式很简单，连续跑 50 次，统计失败率。如果出现偶发失败，再打开enable_logger()看是重连慢还是 topic 对不上。把重连失败也算进统计，你会发现很多平台侧的问题，其实是设备端没有做指数退避重连导致的。
我记得有一次教训：发布新固件前只验证了“上报成功”，没验证“命令下行”。结果现场反映设备在控制台显示在线，却控制不了。最后查下来，是命令响应的 topic 格式按照旧平台的写法代入了新环境，固件迁移时只改了上行链路，下行一直没对上。从那以后，我的回归验证里永远把下行命令回执列为必测项，即使本次改动只涉及采集逻辑。若时间实在不够，至少也要保留这一行“命令下行回执”，因为它最能兜住影响用户感知的那一类错误。
这条习惯帮我躲过不少半夜被叫醒的麻烦。每次版本更新前跑一遍回归链路，比在平台日志里手动翻找要节省太多时间。希望帮到你。
本文还有配套的精品资源，点击获取
