简介DMLS协议中文版是一份面向电能量数据采集终端与电能表通讯开发的IEC62056协议族中文说明文档适合电力采集终端研发、测试人员及协议栈学习研究者快速掌握DLMS协议体系。文档共1个doc文件压缩包约620KB属于轻量化的技术手册便于随时查阅。内容从整体协议模型切入系统讲解物理层、链路层、应用层三层结构并针对应用层重点剖析ASN.1语法、BER与AXDR编码、AARQ/AARE连接帧及数据请求流程同时提供请求电量、瞬时量、负荷曲线、时间等实际报文范例帮助读者将抽象规约落到具体抓包与开发场景中。文中还梳理了HDLC链路控制、地址校验、CRC校验以及拆包组包等关键机制对理解电能表数据采集链路很有帮助。目前已有866人学习下载可作为开发调试时的常备参考资料。1. DMLS协议中文版先搞懂它到底解决什么问题说到 DMLS协议做边缘计算和工业数据采集的应该不陌生。它不像 MQTT 那样靠主题发布订阅也不像 HTTP 那样一问一答而是一套带确认、带序号的二进制链路同步规范专门解决设备端与平台端在弱网环境下丢包、重连、乱序的问题。这里说的中文版是指社区把英文原版协议规范和注释整理成了中文并补充了报文示例和字段说明。如果你是网关开发、PLC 二次集成或者物联网平台对接的工程师手头要接入一个只实现了 DMLS 的设备网关那这份中文版就是你快速定位字段含义和排障的最短路径。下面我会把它拆成报文结构、握手流程、实现参数和踩坑点带你从文档一路跑到可验证的客户端。2. 拆解DMLS协议核心机制帧结构、应用层握手与批量确认DMLS 协议的核心不在传输层而在应用层怎么定义“一条消息”。它和 MQTT 最大的区别是DMLS 对每条消息都要求有序、可确认、可重发而不是发布后就不管。这也是很多设备接入平台时既想要 MQTT 的轻量又想要 TCP 的可靠性最后选 DMLS 的原因。中文版文档最有价值的地方就是把“谁先发、谁回、确认到哪个序号”这套交互规则用中文重新讲了一遍省得再对着英文术语猜。2.1 帧结构拆解魔数、序列号、时间戳都在哪里所有二进制协议第一道门槛就是看懂帧。DMLS 固定头部 21 字节后面跟着负载和 2 字节 CRC16。中文版文档最常被粘贴出来的一段就是下面这张表。偏移字段长度说明0magic4固定 DMLS 的 ASCII也就是 0x44 0x4D 0x4C 0x534version1当前协议版本取值 0x015msg_type10x01 注册0x02 数据0x03 ACK0x04 心跳0x05 错误6flags1bit0 负载压缩bit1 表示 token 放在负载首部7seq4小端 uint32按会话单调递增溢出回绕11ts8小端 uint64毫秒级 UTC 时间戳19payload_len2小端 uint16最大 6553521payload可变JSON 或 msgpack 编码末2crc162CRC16-CCITT覆盖头部加负载用 Python 解析这段头部不需要引入任何第三方库struct就够了。下面是我在调试 DMLS 网关时常用的解析函数import struct def parse_dmls_header(header: bytes) - dict: if len(header) 21: raise ValueError(DMLS frame head must be 21 bytes, got %d % len(header)) magic, version, msg_type, flags, seq, ts, payload_len struct.unpack( 4sBBBIQH, header ) if magic ! bDMLS: raise ValueError(bad magic: %r % magic) return { magic: magic, version: version, msg_type: msg_type, flags: flags, seq: seq, ts: ts, payload_len: payload_len, }这里有个很容易看反的地方4sBBBIQH的拆包顺序对应表里的 21 字节。B是 1 字节无符号整数I是 4 字节无符号整数Q是 8 字节无符号整数H是 2 字节无符号整数。如果你写成4sBBBIQH那整个 seq 和 ts 就是从大端解释和规范里的小端编码完全相反。实际上 DMLS 规范写明了所有多字节字段都是小端这也是为什么用而不是。常见做法是先判断 magic 再继续解 payload不要在 magic 不正确时去解后面字段。因为不少设备在裸 TCP 连接上直接发了一个 HTTP 请求你按 DMLS 解就会得到一个莫名其妙的 seq。上表里 seq 是 uint32也就是一般写到 42 亿多就会回绕到 0。这个回绕处理我会在第 4 章展开。2.2 应用层三次消息注册、令牌、数据确认的完整链路DMLS 的会话建立不是 TCP 握手而是三条应用层消息。客户端连上服务端后第一步不是发数据而是发 REGISTER。REGISTER 的 payload 是一个 JSON至少包含client_id。服务端收到后会回一条 REGISTER_ACK里面带上分配给这个设备的token以及服务端允许的window_size和idle_timeout。第三步才是数据消息。数据消息的 payload 必须带上 token否则服务端会把这一帧当成未注册设备直接丢弃并回一个 0x05 错误帧。这个步骤很像 HTTP 里的 Authorization 头区别是 token 不是每次都放在头部而是可以放在 payload 首部通过 flags 的 bit1 标记。这样对帧结构更紧凑。在抓包时你只要看到一段字节流里连续出现 0x44 0x4d 0x4c 0x53就可以认为是 DMLS 流量。我一般会在现场用 tcpdump 先确认设备有没有按预期发注册帧:tcpdump -i eth0 port 35200 -X -n | head -40如果没有看到444d4c53说明设备根本没走 DMLS或者端口配错。-X能同时打印 ASCII 和十六进制-n关掉域名解析避免在弱网现场卡在 DNS 反查上。这个命令不用装额外工具大部分 Linux 发行版自带。注册帧之后服务端返回的 ACK 是 0x03 类型而不是 0x01 类型。很多第一次接的人以为 REGISTER_ACK 也是 0x01结果状态机永远等不到。这一点在中文版里常常被一句话带过但实际代码里我用msg_type 0x03 and seq register_seq来判断注册成功并且还要比对 token 非空。2.3 滑动窗口与批量 ACK为什么 DMLS 在弱网下比 MQTT/HTTP 稳DMLS 的可靠性来自滑动窗口。服务端在注册阶段下发的 window_size 表示“客户端最多可以连续发多少帧而不用等 ACK”。比如 window_size8客户端可以连续发 8 帧然后等待一条累积 ACK。ACK 里带的是“已确认到的最大 seq”表示前面这些全部收到。这比 MQTT QoS1 的逐包确认效率高也比 HTTP 的请求-响应模型少一半的交互次数。在车载、产线这类时延 50ms 以上的链路上逐个确认意味着每帧浪费一个 RTT批量确认则能把 8 帧的往返开销摊薄。而 HTTP 长连接虽然能复用 TCP但对每个业务消息都要构造独立请求队头阻塞问题也更明显。窗口也不能设得太大。window_size 越大发送端缓存越多一旦链路丢包重传风暴也越凶。中文版规范给出的建议是弱网环境下 8~16专线环境 32~64。这个值不是客户端自己能改的必须按服务端 REGISTER_ACK 带下来的值来。如果服务端没下发就用 8 兜底。实现滑动窗口客户端要维护send_base、next_seq和窗口上限。收到 ACK 后send_base max(ack_seq 1, send_base)然后才能从缓存里释放已确认的帧。这里要注意DMLS 允许 ACK 乱序或重复到达所以不能用“收到了 ACK 就清空全部缓存”的简单逻辑。正确做法是等累积 ACK 推进窗口而不是逐个释放。为了便于对照MQTT 的 QoS1 是发送一个包就要等一个 PUBACK窗口恒为 1DMLS 的窗口是服务端动态下发的。如果你在一条高时延链路上做对比会发现同样的 100 条数据MQTT 至少 100 个 RTTDMLS 只需要 13 个 RTT。这也是它常被用在卫星链路和跨地域专线的原因。3. 用 Python 搭一个 DMLS 最小客户端从中文版到可运行代码这一节的核心是让中文版文档变成你能跑起来的代码。下面的脚本我刻意只用标准库这样在任何一台能联网的 Linux 主机上都能直接执行不需要先折腾依赖。3.1 最小客户端帧封装、注册和设备鉴权的完整链路先放一段最基础的帧封装和接收函数。它只负责把 Python 字典变成 DMLS 二进制帧再把收到的二进制帧变回字典。import socket import struct import json import time MAGIC bDMLS def crc16(data: bytes) - int: crc 0xFFFF for byte in data: crc ^ byte 8 for _ in range(8): if crc 0x8000: crc ((crc 1) ^ 0x1021) 0xFFFF else: crc (crc 1) 0xFFFF return crc def build_frame(msg_type: int, seq: int, payload: dict, flags: int 0) - bytes: body json.dumps(payload).encode(utf-8) ts int(time.time() * 1000) header struct.pack(4sBBBIQH, MAGIC, 0x01, msg_type, flags, seq, ts, len(body)) crc crc16(header body) return header body struct.pack(H, crc) def recv_exact(sock: socket.socket, n: int) - bytes: buf b while len(buf) n: chunk sock.recv(n - len(buf)) if not chunk: raise ConnectionError(connection closed by peer) buf chunk return buf def recv_frame(sock: socket.socket, timeout: float 3.0) - dict: sock.settimeout(timeout) header recv_exact(sock, 21) magic, version, msg_type, flags, seq, ts, plen struct.unpack(4sBBBIQH, header) if magic ! MAGIC: raise ValueError(bad magic: %r % magic) body recv_exact(sock, plen 2) payload_bytes body[:-2] return { msg_type: msg_type, seq: seq, payload: json.loads(payload_bytes.decode(utf-8)), }逻辑说明build_frame中ts统一使用毫秒不传时区struct.pack的表示小端与第 2.1 节一致。crc16用的是 CRC16-CCITT 初始值 0xFFFF多项式 0x1021这是 DMLS 中文版规范里给出的默认参数。recv_frame先收 21 字节头再根据plen收负载和 CRC最后把负载当作 JSON 解析。为了控制示例长度这里省略了 CRC 校验逻辑生产代码一定要把body[:-2]与尾部 CRC 做比对否则一个坏包就会让解析器错位。接下来是注册函数。注册的目的是在这条 TCP 连接上拿到服务端下发的 token 和窗口参数。def register(sock: socket.socket, client_id: str, seq: int 0) - dict: sock.sendall(build_frame(0x01, seq, {client_id: client_id})) resp recv_frame(sock, timeout5.0) if resp[msg_type] ! 0x03: raise RuntimeError(register failed, msg_type%s % resp[msg_type]) return resp[payload]参数说明client_id是设备在平台侧的全局唯一标识通常由硬件序列号加型号组成。seq是当前会话的起始序号DMLS 中文版建议每次进程重启都从 0 开始服务端会把 0 当成新会话。如果你在同一个 TCP 连接上重连后强制把 seq 续到之前的尾巴不少服务端反而会当成丢了中间帧直接回0x1003。注册后从返回值里拿出 token、window_size 和 idle_timeout后续所有请求都要带着它们。3.2 心跳与重连这几个参数按现场条件调才不容易半夜翻车DMLS 的会话是有寿命的服务端在注册 ACK 里会下发 idle_timeout表示这条会话多少秒没有消息就会被回收。客户端的心跳周期必须小于这个值常见做法是取它的三分之一。下面这张表是我在多个项目里用下来的参数范围。参数推荐值说明heartbeat_interval30~45s取服务端 idle_timeout 的 1/3recv_timeout5~10s等待 ACK 的超时超过后触发重传reconnect_wait2s 起步指数退避上限 60s避免同时重连造成风暴window_size8按服务端下发为准客户端不能超过服务端限制seq_wrap_threshold0x80000000用于回绕比较见第 4 章心跳帧和普通数据帧走同一个帧结构msg_type 是 0x04。这里有一个容易踩的细节心跳也占用 seq也要递增。很多客户端为了省事把心跳的 seq 写成 0服务端收到后会把之前窗口里的数据帧都当乱序处理表现就是平台侧频繁丢数。def send_heartbeat(sock: socket.socket, token: str, seq: int) - None: sock.sendall(build_frame(0x04, seq, {token: token}))重连逻辑更不能一失败就立刻重连否则几十台设备同时掉线会把服务端打爆。我一般用一个指数退避的循环每次重连间隔翻倍最多 60 秒wait 2 while True: try: sock socket.create_connection((host, port), timeout5) info register(sock, client_id) break except Exception as e: print(reconnect in %.1fs: %s % (wait, e)) time.sleep(wait) wait min(wait * 2, 60)这段代码的问题在于register里已经设置了 5 秒超时所以 create_connection 和 register 的超时叠加最坏可能要 10 秒才发现链路挂了。如果想缩短感知时间可以把 create_connection 的 timeout 改成 3把 recv_frame 的 timeout 改成 4总体上控制在 7 秒内。弱网环境下不建议把 recv_timeout 设成 1 秒一个抖动就会误判反而触发重连风暴。3.3 发送一条带确认的数据阻塞式等 ACK 的写法调试阶段最直观的发送方式是发一条数据等一条 ACK再发下一条。这个逻辑可以写成下面这个函数def send_data_blocking(sock: socket.socket, token: str, seq: int, data: dict) - int: payload {token: token, data: data} sock.sendall(build_frame(0x02, seq, payload)) ack recv_frame(sock, timeout5.0) if ack[msg_type] ! 0x03: raise RuntimeError(expected ACK, got msg_type%s % ack[msg_type]) return ack[seq]说明这里的seq是当前数据帧的序号token来自注册结果。ACK 的seq表示服务端确认到哪个序号如果是累积确认它会大于等于你发送的最新 seq。阻塞式写法逻辑最简单但它一次只能发一帧把 window_size 完全架空。如果现场链路 RTT 是 50ms一帧一停最多只能跑 20 帧/秒遇到高频率采集瞬间背压。3.4 把注册参数存下来窗口和会话状态不该散落在函数里注册返回的window_size、idle_timeout和token是后面所有逻辑的上下文。我建议用一个小的 session 对象将它们打包而不是用全局变量。class DmlsSession: def __init__(self, token: str, window_size: int, idle_timeout: int): self.token token self.window_size window_size self.idle_timeout idle_timeout self.next_seq 1 self.send_base 0这个对象会在滑动窗口确认时用到。每次发数据前从 session 取 token 和 seq收到 ACK 后更新 send_base。如果你把 token、seq 写在多个地方调试时很容易出现“这条用的是旧 token、那条用的新 seq”的串线问题现场排查非常痛苦。4. 常见问题与避坑跑 DMLS 协议时最容易翻车的 5 处地方任何二进制协议都要靠血泪经验填坑DMLS 也一样。下面五条是我在对接设备时反复看到的按现象、原因、解决的顺序列出来。4.1 错误帧当日志读0x05 payload 不是文本而是 JSON 结构现象客户端把 0x05 错误帧的 payload 直接打印成字符串看到一串{err_code:...}又不知道错在哪。原因DMLS 中文版规范里错误帧的 payload 也是一个 JSON但很多示例代码只展示十六进制没说明字段结构。比对新版与旧版时会发现err_code的取值还变过几次。解决先解析 JSON再取字段不要直接打印原始字节。我通常会用一个映射函数ERR_MAP { 0x1001: unregistered, 0x1002: bad token, 0x1003: seq too old, 0x1004: window overflow, } def handle_error(payload: dict): code payload.get(err_code) msg payload.get(err_msg, unknown) raise RuntimeError(%s: %s % (ERR_MAP.get(code, unknown), msg))注意服务端返回的 payload 是小端编码的 JSON解码用 UTF-8不要按 GBK 解。把这条加到客户端入口大部分注册失败都能在十秒内定位到原因。4.2 序列号回绕误判成丢包现象设备连续运行几个月后客户端开始疯狂重传服务端却显示已确认平台侧出现大量重复数据。原因seq 是 uint32回绕后从 0 开始。客户端写的判断是if ack_seq next_seq回绕后0 0xFFFFFFFF不成立于是把 ACK 当旧包丢弃。解决用带符号差比较def seq_gt(lhs: int, rhs: int) - bool: return ((lhs - rhs) 0xFFFFFFFF) 0x80000000seq_gt(ack_seq, next_seq)为真说明 ACK 是新的否则就是旧确认或重复 ACK。这个技巧在 TCP 序列号比较里很成熟DMLS 直接照搬即可。注意两边的整形都必须先按 32 位无符号处理不能混用有符号整数。4.3 时间戳单位不一致日志全对不上现象设备上报时间和平台日志时间相差 8 小时或者差了 1024 倍。原因有人把 ts 直接传秒有人传毫秒还有人传本地时间。DMLS 规范写的是 UTC 毫秒不随系统时区变。解决发送端统一用int(time.time() * 1000)不要用datetime.now().timestamp()后者在部分系统上会受进程时区设置影响。接收端解析后如果 ts 小于 10 位基本说明对方发的是秒需要乘 1000 对齐。日志里统一按毫秒格式化别在 JSON 日志里混用两种单位。更严格的做法是在帧解析入口就做归一化def norm_ts(raw_ts: int) - int: if raw_ts 1_000_000_000: return raw_ts * 1000 return raw_ts这个判断能兜住大多数固件实现差异但不能完全替代协议一致性测试。4.4 多设备共用一条 TCP 连接时 session 隔离没做现象两个设备复用一条连接A 设备的数据包里带了 B 设备的 token平台侧数据串线。原因DMLS 允许一条 TCP 连接多路复用但 session 以 token 为标识。上报数据帧里的 token 没有与服务端当前连接的注册记录校验客户端复用了同一个套接字又没有正确切换上下文。解决每个数据帧的 token 从本次注册的返回值里取不要写死配置。如果只有一个 TCP 连接但逻辑上有多个设备需要为每个设备单独维护一个 session 字典发送前明确知道当前套接字对应哪个 sessionsessions {} def register_device(sock, client_id): info register(sock, client_id) sessions[client_id] DmlsSession(info[token], info[window_size], info[idle_timeout])服务端收到 token 不匹配时应当回 0x1002 并断开连接防止污染后续帧。客户端这边连接断开后也要及时把 session 从字典里移除。4.5 中文版术语不统一flag / flagsbody / payload现象按中文版字段名去写结构体结果字段对不上抓包也看不懂。原因中文版来自不同维护者有的地方把flags翻译成flag把payload翻译成body字段名在示例代码里混用。解决以英文原版字段名为准把中文版当注释。程序里统一用flags和payload。如果历史代码用了body可以在协议层加别名转换但解析器入口只认规范名FIELD_ALIASES { flag: flags, body: payload, }这条不算技术坑却是团队协作里最容易被忽略的坑。中文版读得越久反而越容易把两种叫法混在代码里。我在 code review 时遇到混用字段名的改动都会直接打回去。5. 进阶先跑通本地回环验证再连真机进阶阶段我给自己的硬性要求是不拿真机当试验田。DMLS 这种二进制协议一旦字段偏移错了真机日志根本没法看。所以我会先做一个本地回环测试用同一个 crc16 和帧结构在内存里互相发一帧。5.1 回环测试模拟服务端验活下面这个 mock server 只做注册响应不接受数据帧但对验证客户端是不是一上来就乱发数据已经够了。import socket import threading def mock_server(host, port, token): srv socket.socket() srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srv.bind((host, port)) srv.listen(1) def handler(conn): frame recv_frame(conn, timeout5) if frame[msg_type] 0x01: payload {token: token, window_size: 8, idle_timeout: 90} conn.sendall(build_frame(0x03, frame[seq], payload)) conn.close() t threading.Thread(targethandler, args(srv.accept()[0],)) t.start() return srv然后跑一个最简断言srv mock_server(127.0.0.1, 35200, tok-test) client socket.create_connection((127.0.0.1, 35200), timeout5) info register(client, gw-test) assert info[token] tok-test print(loopback register OK) srv.close()这段代码的前提是你已经把第 3 章的build_frame、recv_frame、register放到同一个模块里。我实际做的时候还会在断言里加window_size 8因为服务端下发的窗口和客户端后续的发送策略强相关。顺序一定要先 register 再发数据不能在连接建立后立刻 sendall 数据帧否则 mock server 会等不到注册帧直接超时。我还习惯在回环里加一个 seq 回绕的模拟把 mock server 返回的 ACK 的 seq 改成 0xFFFFFFFF再验证客户端的seq_gt能正确判断新 ACK。这个技巧能提前暴露很多生产环境跑几个月才会出现的问题。每次接新设备我都会先把这套回环跑一遍再把客户端连到真机。养成这个习惯之后我在现场排错的次数少了一大半。希望帮到你。本文还有配套的精品资源点击获取