企业官网建设流程全解析

12-总线冲突与轮询机制：多设备485总线分时轮询、防阻塞方案

一条 485 总线上挂了 20 个传感器、10 个继电器模块，主站一开口，好几个设备同时回应——场面立刻变成"群口相声"，谁也听不清谁。这就是总线冲突。

本篇讲透 RS485 半双工的冲突原理、主从轮询机制怎么设计、超时怎么跳、如何防阻塞，最后给出可落地的 Python/C 轮询调度代码和总线负载评估方法。

一、RS485 半双工：为什么必须"排队说话"

RS485 是半双工总线，物理上只有两条差分线（A/B），同一时刻只能有一个设备在线上发送。它没有"冲突检测"（不像以太网有 CSMA/CD），谁先抢到线谁发，撞车就是比特流互踩，产生乱码帧。

Modbus RTU 用主从模式从根本上规避冲突：只有主站（Master）能主动发请求，从站（Slave）收到合法的、带自己地址的请求才应答。从站之间、从站与从站之间永不直接通信。

主站 ──→ 从站1: "1号，报告湿度" 从站1 ──→ 主站: "湿度80%" 主站 ──→ 从站5: "5号，打开电磁阀" 从站5 ──→ 主站: "已打开"

一问一答，天然单通道。规则简单，但实现上全是细节坑：轮询顺序、超时、离线设备、应答延迟，处理不好照样把总线堵死。

二、轮询机制：主站挨个问话

主站按顺序向每个从站发请求，等应答，然后问下一个：

轮询循环: 问从站1 → 等应答 → 问从站2 → 等应答 → ... → 问从站N → 回到从站1

关键时间参数

9600 波特率下 1 个字符约 1.04ms，3.5 字符 ≈ 3.6ms。用 9600 波特率传一个"读 2 个寄存器"的帧约 8 字节 + 应答 9 字节 ≈ 20ms 一个往返。

三、轮询周期设计：按重要性分优先级

全部设备按"重要性 + 数据变化频率"排队：

实现上用"加权轮询"：高优先级点每次轮询都发，低优先级点每 N 轮才发一次。

SENSORS=[{"name":"air_temp","addr":0x01,"period":1},# 每轮{"name":"air_hum","addr":0x02,"period":1},{"name":"soil_ec","addr":0x03,"period":3},# 每3轮{"name":"soil_ph","addr":0x04,"period":3},{"name":"valve_fb","addr":0x05,"period":1},# 阀反馈，闭环用{"name":"light","addr":0x06,"period":5},]defnext_poll_cycle(round_no):"""返回本轮要轮询的设备列表"""return[sforsinSENSORSifround_no%s["period"]==0]

四、轮询超时与跳过机制：离线设备不拖后腿

致命错误示范：某个设备断电了，主站发请求后傻等 5 秒超时，再问下一个。20 个设备里死 3 个，一轮周期从 2 秒拖到 17 秒，其余正常设备的实时性全被拖垮。

正确做法是分级超时 + 快速跳过：

1. 发请求 2. 等待应答，超时 = 300ms（可配置） 3. 超时未收到 → 立即放弃，进入下一个设备 4. 连续离线 N 次（如3次）→ 标记该设备离线，切换为"低频探活" 5. 探活模式：每 10 秒发一次探测帧，恢复应答即回到正常轮询

OFFLINE_THRESHOLD=3# 连续3次无应答判离线PROBE_INTERVAL=5# 离线后每5轮探活1次fail_count={}defpoll_device(ser,dev):"""返回 True 收到应答 / False 超时"""frame=build_read_frame(dev["addr"],dev["reg"],dev["len"])ser.write(frame)resp=wait_response(ser,timeout_ms=300)returnrespisnotNonedefround_robin(ser):round_no=0whileTrue:round_no+=1fordevinSENSORS:# 离线设备降频探活ifdev["addr"]inOFFLINE:ifround_no%PROBE_INTERVAL!=0:continueifpoll_device(ser,dev):fail_count[dev["addr"]]=0OFFLINE.discard(dev["addr"])parse_and_store(dev,last_resp)else:fail_count[dev["addr"]]=fail_count.get(dev["addr"],0)+1iffail_count[dev["addr"]]>=OFFLINE_THRESHOLD:OFFLINE.add(dev["addr"])# 进入低频探活log(f"设备{dev['addr']}离线，转入探活模式")

五、防阻塞方案：别让轮询把 CPU 卡死

主站程序要是"发一帧等一帧"地阻塞，几十个设备就会让程序卡在串口 I/O 上，其它逻辑（存储、上报、报警）全停摆。三个工程级方案：

1. 非阻塞轮询（状态机）

不在wait_response里死等，而是用状态机 + 定时器驱动：

状态机: SEND: 发送请求 → 记录发送时刻 → 进入WAIT WAIT: 检查串口缓冲区，有数据则解析校验 → 收到完整帧 → 处理 → 进入NEXT → 超时(300ms) → 记录失败 → 进入NEXT NEXT: 选下一个设备 → 回到SEND

核心是主循环不阻塞，WAIT状态每次都查一次缓冲区就返回，主循环继续跑别的任务。

2. 异步收发 + 看门狗

串口用独立接收线程，收完一帧交给轮询状态机；主线程同时跑一个看门狗定时器：

classWatchdog:def__init__(self,timeout_ms=1000):self.deadline=time.monotonic()+timeout_ms/1000deffeed(self):self.deadline=time.monotonic()+0.3defexpired(self):returntime.monotonic()>self.deadline wd=Watchdog()# 每次轮询正常推进就 wd.feed()# 主循环里检查：if wd.expired(): 强制重置轮询状态机

一旦某帧异常导致状态机卡死，看门狗强制复位轮询逻辑，总线不至于永久瘫痪。

3. 轮询与业务解耦

串口轮询只负责"取数"，取到的数据丢进队列，业务逻辑（自动控制、云端上报）从队列消费。串口线程慢不影响业务线程：

importqueue data_queue=queue.Queue()defpoll_loop(ser):whileTrue:# ... 轮询取数 ...data_queue.put({"addr":addr,"values":parsed})defcontrol_loop():whileTrue:data=data_queue.get()ifdata["values"]["temp"]>35:turn_on_fan()# 业务逻辑独立运行

六、轮询调度表设计：一个可落地的调度器

把调度逻辑抽象成"调度表"，运维改配置不用改代码：

# 调度表: poll_schedule.yamlpoll_interval_ms:100# 每帧最小间隔timeout_ms:300# 单帧超时offline_threshold:3# 离线判定次数probe_interval_rounds:5# 探活周期slaves:-{addr:1,name:空气温湿度,regs:[0x0000,0x0001],priority:high,period:1}-{addr:2,name:土壤EC/PH,regs:[0x0000,0x0001],priority:medium,period:3}-{addr:5,name:水肥机状态,regs:[0x0014],priority:high,period:1}-{addr:9,name:电磁阀反馈,regs:[0x0000],priority:high,period:1}

调度器通用伪代码：

schedule = load_yaml("poll_schedule.yaml") round_no = 0 while True: round_no += 1 for slave in schedule.slaves: if round_no % slave.period != 0: continue if is_offline(slave) and round_no % probe_interval != 0: continue resp = request(slave, timeout=schedule.timeout_ms) if resp: on_success(slave, resp) else: on_timeout(slave)

七、C 语言版本：嵌入式网关轮询核心

嵌入式 Linux 网关常用 C 实现，核心就是"状态机 + 非阻塞 select"：

#include<stdio.h>#include<sys/select.h>#include<time.h>#defineTIMEOUT_MS300/* 轮询状态机 */enum{ST_SEND,ST_WAIT,ST_NEXT}state=ST_SEND;structtimespecsend_time;voidpoll_loop(intfd){while(1){if(state==ST_SEND){build_and_send(fd,current_slave);clock_gettime(CLOCK_MONOTONIC,&send_time);state=ST_WAIT;}if(state==ST_WAIT){fd_set fds;FD_ZERO(&fds);FD_SET(fd,&fds);structtimevaltv={0,100000};/* 100ms非阻塞检查 */if(select(fd+1,&fds,NULL,NULL,&tv)>0){if(parse_frame(fd,&buf)){/* 收到完整合法帧 */handle_data(current_slave,&buf);state=ST_NEXT;}}if(elapsed_ms(send_time)>TIMEOUT_MS){handle_timeout(current_slave);/* 超时跳下一个 */state=ST_NEXT;}}if(state==ST_NEXT){current_slave=next_slave(current_slave);state=ST_SEND;}/* 主循环还可以跑 watchdog 和其他业务 */watchdog_check();}}

select带 100ms 超时，主循环每 100ms 至少醒一次，永远不会死等。这就是嵌入式网关里"非阻塞轮询"的经典写法。

八、多设备总线负载评估

挂多少设备、用多少波特率，先算算账：

每帧往返时间 T = 请求帧字节数×10位/波特率 + 响应帧字节数×10位/波特率 + 设备响应延迟 + 帧间隔

示例：9600 波特率，请求 8 字节、应答 9 字节、设备延迟 20ms：

T = (8+9)×10/9600 ×1000 + 20 ≈ 17.7 + 20 ≈ 38ms

20 个设备、每轮每设备 1 帧 → 一轮约 0.76 秒。若要求 1 秒内全部扫完，9600 勉强够；20 个以上设备或要控制实时性更好，建议38400 波特率（约快 4 倍）或升级为Modbus TCP（以太网全双工，不存在冲突问题）。

总线负载率计算

负载率 = 有效数据时间 / 总线可用时间

设计建议：

超时时间也要反着算：总线越忙，超时设置应越短，否则离线设备会进一步放大阻塞。

九、小结

485 总线的核心纪律就两条：主从一问一答，绝不两个设备同时开口。工程实现上把握四件事：按优先级做加权轮询、超时快速跳过离线设备、用状态机/异步实现非阻塞、按负载率决定波特率和设备数。把这套机制搭好，几十个设备在一条线上也能跑得又快又稳。