TCP/IP展厅中控系统：可靠连接、指令闭环与断网韧性设计-平芜编程栈

简介：这是一款面向展厅、会议室等智能中控场景的跨平台软件解决方案，专为非编程人员设计，通过可视化拖曳方式快速构建人机交互界面，解决传统中控系统开发门槛高、UI定制难、多端协同弱等痛点。资源包共373个文件，含272个PNG与56个JPG格式UI素材（位于userData目录，支持完全自定义替换），16个核心DLL动态库及3个Windows端EXE主程序，2个Android APK安装包（含Main.apk与AdobeAir.apk），另有XML配置、INI参数、DB数据库及SWF动画等辅助文件，整体88.01MB，结构完整、即装即用。已有2105人学习下载，涵盖系统集成商、展厅运维工程师及智能硬件开发者。用户可直接部署运行，获得TCP/UDP/串口/PJLink/网络唤醒全协议支持、分组一键开关、键盘指令绑定、灯光延时控制、页面与按钮动态增删、平板与PC无线数据同步等完整中控功能，同时具备Adobe AIR运行环境依赖组件与Arduino扩展能力（含.ane插件），满足中小型智能空间的灵活部署与长期迭代需求。

1. 展厅中控不是“接线盒”，而是要能扛住三小时连续巡检、五路设备并发指令、断网后自动降级的TCP/IP软件系统

很多同学做毕设或课设时，一看到“中控”就默认是串口+红外+继电器的硬件盒子，配个按钮界面完事。但真实展厅场景根本不是这样：布展期间调试阶段，中控要同时响应导览平板、后台管理端、语音唤醒模块、定时任务脚本、甚至第三方预约系统的HTTP回调；正式开放后，它得在无专人值守状态下连续运行72小时以上，中间不能重启，设备掉线要秒级感知，指令发出去必须有ACK确认，超时未回还得自动重试+日志标记；更关键的是——展厅网络常不稳定，Wi-Fi信号穿墙衰减、交换机偶发丢包、甚至临时断电重启路由器，这时候中控不能“死机”，得自动切到本地缓存策略，用预置逻辑维持基础展项运转。这套基于TCP/IP的智能中控软件程序，就是为解决这三类硬需求而设计的：可靠连接、指令闭环、断网韧性。它适合需要交付可演示、可答辩、可实际部署的毕设/课设同学，也适合某公司展厅升级项目中负责落地实施的初级工程师——不堆概念，只讲怎么让TCP连接不掉、指令不丢、故障不懵。

2. 总体架构设计：四层结构与五个核心模块的职责划分

这套中控软件不是单体大程序，而是按“通信底座—指令中枢—设备适配—业务逻辑—人机交互”分层解耦。我一般会强制画一张白板草图给导师看：最底层是TCP长连接池，中间是带状态机的指令调度器，往上是设备驱动抽象层（不是直接写RS232代码），再往上是展厅业务规则引擎（比如“观众进入A区→自动开启投影+调暗灯光+播放导览音频”），最上层才是Web管理界面和串口/蓝牙调试终端。这种分层不是为了炫技，而是为了答辩时能清晰回答“如果投影仪协议换了，改哪？如果要加人脸识别联动，动哪？如果客户说不要Web端只要微信小程序，砍哪？”——答案分别是：设备驱动层、规则引擎层、人机交互层。所有模块通过统一的消息总线通信，消息格式固定为JSON，含msg_id、timestamp、src、dst、cmd_type、payload六字段，避免后期各模块私聊导致耦合爆炸。

2.1 TCP通信层：心跳保活与连接状态机的双保险机制

TCP本身不保证应用层可用，展厅里常见问题是：设备端程序崩溃了，但TCP连接还挂着（TIME_WAIT状态残留）；或者网络抖动导致ACK丢失，客户端以为发成功了，服务端根本没收到。所以我在通信层写了两级防护：

# tcp_client.py 核心心跳逻辑（Python示例） import socket import threading import time import json class TCPSession: def __init__(self, host, port): self.host = host self.port = port self.sock = None self.is_connected = False self.last_heartbeat = time.time() self.heartbeat_interval = 15 # 秒 self.ack_timeout = 8 # 等ACK超时时间 def connect(self): try: self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.settimeout(5) self.sock.connect((self.host, self.port)) self.sock.settimeout(None) # 后续用非阻塞+select self.is_connected = True self.last_heartbeat = time.time() threading.Thread(target=self._heartbeat_loop, daemon=True).start() return True except Exception as e: print(f"[ERROR] Connect failed: {e}") return False def _heartbeat_loop(self): while self.is_connected: if time.time() - self.last_heartbeat > self.heartbeat_interval: # 发送心跳包（不含业务数据，仅用于探测） hb_pkt = {"type": "HEARTBEAT", "seq": int(time.time())} try: self.sock.sendall(json.dumps(hb_pkt).encode('utf-8') + b'\n') self.last_heartbeat = time.time() except Exception as e: print(f"[WARN] Heartbeat send failed: {e}") self._on_disconnect() break time.sleep(2) def send_with_ack(self, cmd_data: dict) -> bool: """发送带ACK确认的指令，失败自动重试3次""" for attempt in range(3): try: # 添加唯一请求ID，服务端回包必须带此ID cmd_data["req_id"] = f"REQ_{int(time.time()*1000)}_{attempt}" payload = json.dumps(cmd_data).encode('utf-8') + b'\n' self.sock.sendall(payload) # 启动ACK监听（非阻塞方式，避免卡主线程） ack_received = self._wait_for_ack(cmd_data["req_id"], timeout=self.ack_timeout) if ack_received: return True except Exception as e: print(f"[WARN] Send attempt {attempt} failed: {e}") time.sleep(0.3) return False def _wait_for_ack(self, req_id: str, timeout: float) -> bool: # 实际项目中此处用select或epoll监听socket读事件 # 为简化示例，用短时阻塞模拟（生产环境禁用！） start = time.time() while time.time() - start < timeout: try: self.sock.settimeout(0.1) data = self.sock.recv(1024) if data: msg = json.loads(data.decode('utf-8').strip()) if msg.get("type") == "ACK" and msg.get("req_id") == req_id: return True except socket.timeout: continue except Exception as e: print(f"[ERROR] ACK parse failed: {e}") break return False

这段代码的关键不在语法，而在设计意图：

HEARTBEAT包不携带业务数据，纯探测，避免被设备端误解析；
每条业务指令带req_id，服务端ACK必须原样返回，杜绝“发了以为成功，其实丢了”的玄学问题；
send_with_ack()封装重试逻辑，但重试间隔0.3秒而非立即重发，防止网络拥塞雪崩；
_wait_for_ack()用短时阻塞+循环，是学生项目可接受的折中方案（工业级需用select或asyncio）。

提示：很多同学把心跳和业务指令混在一起发，结果设备端协议解析器一看到陌生字段就丢包。务必让心跳包格式极简，且与业务包严格区分。

2.2 指令调度中心：基于有限状态机（FSM）的指令生命周期管理

中控最怕“指令发出去石沉大海”。比如发了“打开投影仪”，但投影仪正在自检，10秒后才响应；此时若用户又点“关闭”，系统得知道前一条还没完成，不能简单覆盖。所以我用FSM管理每条指令的状态流转：PENDING → SENT → ACK_RECEIVED → EXECUTING → COMPLETED / FAILED / TIMEOUT。状态变更全部记日志，并暴露查询接口（如GET /api/cmd/status?req_id=REQ_1234567890_0）。

-- 数据库表：command_log（SQLite轻量级，适合毕设） CREATE TABLE command_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, req_id TEXT NOT NULL UNIQUE, cmd_type TEXT NOT NULL, -- 'POWER_ON', 'VOLUME_UP', 'PLAY_VIDEO' device_id TEXT NOT NULL, -- 'projector_a', 'light_zone_b' payload TEXT, -- JSON原始负载 status TEXT NOT NULL DEFAULT 'PENDING', -- 'PENDING','SENT','ACK_RECEIVED',... created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, error_msg TEXT, retry_count INTEGER DEFAULT 0 ); -- 状态更新SQL（示例：收到ACK后更新） UPDATE command_log SET status = 'ACK_RECEIVED', updated_at = CURRENT_TIMESTAMP WHERE req_id = 'REQ_1234567890_0' AND status = 'SENT';

这个表不只是为了查日志，更是为了实现“指令幂等性”：同一req_id重复提交，调度器直接返回历史状态，不二次下发。某高校毕设答辩时，老师故意狂点按钮测试，靠这个表+状态机稳住了全场。

2.3 设备驱动抽象层：协议无关的设备注册与命令映射

展厅设备五花八门：投影仪用PJLink协议（TCP文本）、灯光控制器走DMX512（需USB转串口）、音响用Crestron串口指令、LED屏走UDP。如果每个设备都写一套通信代码，维护成本爆炸。我的做法是定义统一设备接口：

# device_driver.py from abc import ABC, abstractmethod class DeviceDriver(ABC): @abstractmethod def connect(self) -> bool: pass @abstractmethod def send_command(self, cmd: dict) -> bool: """cmd格式统一：{'action': 'power_on', 'params': {...}}""" pass @abstractmethod def parse_response(self, raw_data: bytes) -> dict: """返回标准化响应：{'status': 'success', 'data': {...}}""" pass # 具体实现示例：PJLink投影仪驱动 class PJLinkDriver(DeviceDriver): def __init__(self, ip, port=4352): self.ip = ip self.port = port self.sock = None def connect(self) -> bool: try: self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.connect((self.ip, self.port)) # PJLink握手：发送%1INFO\r\n，等待%1INFOSUCCESS\r\n self.sock.send(b'%1INFO\r\n') resp = self.sock.recv(1024) return b'SUCCESS' in resp except: return False def send_command(self, cmd: dict) -> bool: if not self.sock: return False # 将标准cmd映射为PJLink指令 action_map = { 'power_on': b'%1POWR1\r\n', 'power_off': b'%1POWR0\r\n', 'input_hdmi': b'%1INPT1\r\n', } pj_cmd = action_map.get(cmd['action']) if pj_cmd: self.sock.send(pj_cmd) return True return False def parse_response(self, raw_data: bytes) -> dict: # PJLink响应示例：%1POWR=1\r\n → {'status': 'success', 'power': 'on'} if raw_data.startswith(b'%1POWR='): state = raw_data[7:8].decode() return {'status': 'success', 'power': 'on' if state == '1' else 'off'} return {'status': 'unknown'}

关键设计点：

所有设备驱动继承DeviceDriver抽象类，上层调度器只认这个接口；
send_command()接收标准化cmd字典，内部做协议转换，外部完全不知PJLink细节；
parse_response()把原始字节流转成统一JSON结构，供规则引擎消费；
驱动实例在配置文件中注册，增删设备只需改配置，不动核心代码。

3. 关键模块实现：展厅业务规则引擎与Web管理界面

规则引擎是中控的“大脑”，它决定“什么条件下触发什么动作”。很多同学用if-else硬编码，结果加一个新联动就要改代码、重新编译。我采用JSON规则配置+解释执行的方式，既安全又易扩展。

3.1 基于JSON Schema的可视化规则配置体系

规则存为JSON文件（如rules/exhibition_flow.json），结构如下：

{ "rule_id": "entry_trigger", "name": "观众入场自动启动", "enabled": true, "triggers": [ { "type": "sensor_event", "sensor_id": "entrance_sensor_a", "event": "motion_detected", "debounce_ms": 3000 } ], "conditions": [ { "type": "time_range", "start": "09:00", "end": "17:00" } ], "actions": [ { "device_id": "projector_main", "command": "power_on" }, { "device_id": "light_zone_1", "command": "dim_to", "params": {"level": 30} }, { "device_id": "audio_system", "command": "play", "params": {"file": "welcome.mp3"} } ] }

规则加载器用Pythonjsonschema校验格式，确保必填字段存在、时间格式合法、设备ID在注册列表中。校验通过后，由规则执行器（RuleExecutor）解析并绑定到事件总线：

# rule_executor.py import json import threading from datetime import datetime, time class RuleExecutor: def __init__(self, device_manager, event_bus): self.device_manager = device_manager self.event_bus = event_bus self.rules = {} self._load_rules() def _load_rules(self): # 从rules/目录加载所有.json文件 for rule_file in Path("rules").glob("*.json"): try: rule = json.load(rule_file.open()) if rule.get("enabled"): self.rules[rule["rule_id"]] = rule self._bind_triggers(rule) except Exception as e: print(f"[ERROR] Load rule {rule_file}: {e}") def _bind_triggers(self, rule): for trigger in rule.get("triggers", []): if trigger["type"] == "sensor_event": # 订阅传感器事件（假设event_bus已实现发布订阅） self.event_bus.subscribe( topic=f"sensor/{trigger['sensor_id']}/{trigger['event']}", callback=lambda data, r=rule: self._evaluate_rule(r, data), debounce_ms=trigger.get("debounce_ms", 0) ) def _evaluate_rule(self, rule, event_data): # 检查时间条件 now = datetime.now().time() for cond in rule.get("conditions", []): if cond["type"] == "time_range": start_t = datetime.strptime(cond["start"], "%H:%M").time() end_t = datetime.strptime(cond["end"], "%H:%M").time() if not (start_t <= now <= end_t): return # 执行动作 for action in rule.get("actions", []): device = self.device_manager.get(action["device_id"]) if device: cmd = { "action": action["command"], "params": action.get("params", {}) } device.send_command(cmd)

这套设计的好处是：

规则修改无需重启服务，RuleExecutor支持热重载（检测文件mtime变化）；
条件支持组合（未来可加AND/OR字段），当前先做单条件；
debounce_ms防抖是血泪经验——某展厅红外传感器受空调气流干扰，1秒内触发17次，没防抖直接把投影仪开关了17遍。

3.2 Web管理界面：Vue3 + Element Plus的轻量级控制台

毕设答辩必须有可视化界面。我选Vue3（Composition API）+ Element Plus（组件成熟、文档全），打包后整个前端仅380KB，Nginx静态托管即可，不依赖Node.js后端。关键不是炫酷，而是信息密度高、操作路径短、状态反馈强。

核心页面：

设备监控页：表格列出所有注册设备，实时显示状态（在线/离线/错误）、最后心跳、指令成功率（近10分钟），点击设备行展开详情（最近5条指令日志）；
手动控制页：按区域分组（A区/B区/中央通道），每个设备有快捷按钮（电源、音量、输入源），按钮旁实时显示当前状态（如“电源：ON”绿色，“音量：65%”进度条）；
规则管理页：JSON编辑器（用vue-json-pretty），左侧树形展示规则列表，右侧编辑内容，保存时自动校验并提示错误位置；
日志查询页：支持按时间范围、设备ID、指令类型筛选，结果支持导出CSV（答辩时导出一份“过去一小时所有指令”给老师看）。

注意：Web界面所有API调用都带X-Request-ID头，后端日志可关联前后端请求，排查问题时不用猜“用户点的哪个按钮”。

3.3 断网降级模式：本地缓存策略与离线指令队列

展厅最怕断网。我的方案是：当TCP连接中断超过30秒，自动启用降级模式——

本地缓存设备状态：将最后一次成功获取的状态（如投影仪电源=ON）存入SQLite，断网期间所有状态查询返回缓存值；
离线指令队列：用户在断网时发出的指令，存入本地队列（offline_queue.db），恢复联网后按FIFO重发，每条带retry_count字段，超3次失败则标记FAILED并告警；
预置应急规则：配置emergency_rules.json，如“断网持续5分钟 → 自动关闭所有非必要设备（仅保留照明和出口指示）”，该规则由本地定时器触发，不依赖网络。

# offline_manager.py import sqlite3 import json import time from threading import Lock class OfflineManager: def __init__(self, db_path="offline_queue.db"): self.db_path = db_path self.lock = Lock() self._init_db() def _init_db(self): with sqlite3.connect(self.db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS offline_commands ( id INTEGER PRIMARY KEY AUTOINCREMENT, req_id TEXT, cmd_json TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, retry_count INTEGER DEFAULT 0, status TEXT DEFAULT 'QUEUED' ) """) def enqueue(self, cmd_data: dict): with self.lock: with sqlite3.connect(self.db_path) as conn: conn.execute( "INSERT INTO offline_commands (req_id, cmd_json) VALUES (?, ?)", (cmd_data.get("req_id"), json.dumps(cmd_data)) ) def get_pending(self, limit=10) -> list: with self.lock: with sqlite3.connect(self.db_path) as conn: cur = conn.execute( "SELECT id, cmd_json FROM offline_commands WHERE status='QUEUED' ORDER BY created_at LIMIT ?", (limit,) ) return [{"id": r[0], "cmd": json.loads(r[1])} for r in cur.fetchall()] def mark_sent(self, cmd_id: int): with self.lock: with sqlite3.connect(self.db_path) as conn: conn.execute( "UPDATE offline_commands SET status='SENT', retry_count=retry_count+1 WHERE id=?", (cmd_id,) )

这个模块在答辩时是加分项：我当场拔掉网线，演示“断网后点开灯→界面显示‘已发送（离线）’→插回网线→日志显示‘重发成功’”，老师立刻点头。

4. 部署与联调：Windows/Linux双平台打包及设备对接实操指南

写完代码只是开始，部署和联调才是检验是否真懂的环节。很多同学卡在“为什么我的程序连不上投影仪”，其实90%是串口权限、防火墙、IP配置问题。以下是我带过三个展厅项目的标准化流程。

4.1 双平台可执行文件打包：PyInstaller多环境构建

项目用Python开发，但交付不能要求客户装Python。我用PyInstaller打包，但绝不直接pyinstaller main.py——那会打出几百MB的包，且Windows/Linux行为不一致。

Windows打包（Win10/11）：

# 安装PyInstaller（虚拟环境内） pip install pyinstaller==6.2.0 # 构建spec文件（关键！控制排除项） pyinstaller --onefile --console \ --add-data "rules;rules" \ --add-data "config;config" \ --exclude-module tkinter \ --exclude-module matplotlib \ --name "exhibition-control-win" \ main.py

Linux打包（Ubuntu 22.04 LTS）：

# 在目标系统版本上构建（避免glibc版本冲突） docker run -v $(pwd):/workspace -w /workspace python:3.10-slim bash -c " pip install pyinstaller==6.2.0 && pyinstaller --onefile --console \ --add-data 'rules:rules' \ --add-data 'config:config' \ --exclude-module tkinter \ --name 'exhibition-control-linux' \ main.py "

提示：--add-data参数Windows用分号;，Linux用冒号:，这是学生最容易翻车的点。打包后用ldd exhibition-control-linux检查动态库依赖，确保无not found。

4.2 设备对接实操：从协议文档到第一条成功指令的七步法

对接新设备（如某品牌LED屏）不是写代码，而是读文档+抓包+试错。我总结七步法，某同学按此三天搞定原本两周没通的设备：

找协议文档：官网搜“LED屏型号 + protocol manual”，重点看“通信接口”（RS232/RS485/TCP）、“默认波特率”、“校验方式”；
确认物理连接：LED屏是网口？查IP；是串口？用USB转串口线，dmesg | grep tty看Linux分配的/dev/ttyUSB0；
测通断：Windows用ping，Linux用nc -zv IP PORT，不通先查网线、交换机、防火墙；
抓原始流量：用Wireshark（TCP）或cat /dev/ttyUSB0（串口，需先stty -F /dev/ttyUSB0 9600 raw -echo）；
发最简指令：协议文档里找“获取固件版本”指令（通常最稳定），如0x55 0xAA 0x01 0x01，用Pythonserial.write()或socket.send()发；
比对响应：抓包看设备回了什么，是否符合文档描述的“成功响应格式”；
写驱动：按2.3节的DeviceDriver接口封装，send_command()发指令，parse_response()解析回包。

某次对接LED屏，文档写“响应长度固定16字节”，但实测有时15字节（末尾\x00被截断），我在parse_response()里加了resp = resp.ljust(16, b'\x00')兜底，问题解决。

4.3 联调避坑：网络、权限、时序三大高频翻车点

现象1：程序能连上设备IP，但发指令无响应

原因：设备端开启了“指令来源IP白名单”，只允许展厅管理PC的IP访问，而你的开发机IP不在白名单。
解决：联系设备供应商获取白名单配置入口，或让运维将开发机IP加入；临时调试可在设备端关白名单（仅限测试环境）。

现象2：Linux下串口设备权限拒绝（PermissionError: [Errno 13]）

原因：普通用户无权访问/dev/ttyUSB0，需加入dialout组。
解决：

sudo usermod -a -G dialout $USER # 退出终端重登，或执行： newgrp dialout

现象3：多设备并发指令时，某设备响应延迟高达5秒

原因：串口设备（如灯光控制器）是半双工，指令间需加delay_ms，但代码里忘了加。
解决：在对应设备驱动的send_command()末尾加time.sleep(0.1)，或在配置文件中为该设备设置min_interval_ms: 100，调度器统一管控。

现象4：Web界面显示设备在线，但实际指令发不出

原因：TCP连接池中该设备连接已断，但心跳包因网络抖动延迟到达，状态未及时更新。
解决：在TCPSession._heartbeat_loop()中，收到心跳ACK后，不仅更新last_heartbeat，还要检查sock.fileno() != -1，否则强制_on_disconnect()。

现象5：规则触发后，部分动作执行了，部分没执行

原因：规则中多个actions并发执行，但某设备驱动不是线程安全的（如串口对象被多线程共用）。
解决：在DeviceDriver基类中加锁机制，或为每个设备驱动实例化独立线程，禁止跨线程共享socket/serial对象。

5. 验收与答辩：一份可演示、可复现、可答辩的交付清单

答辩不是讲技术多牛，而是证明你做的东西真能用、真稳定、真解决问题。我带过的某高校毕设，老师第一句话就是：“别讲原理，现在打开电脑，让我看它怎么工作。” 所以验收必须围绕“可演示”展开，以下是我的交付清单和答辩技巧。

5.1 必须包含的五份交付物（缺一不可）

文件名	格式	说明	答辩价值
`exhibition-control-win.exe`	Windows可执行文件	直接双击运行，无需安装	展示“零依赖部署”能力
`exhibition-control-linux`	Linux可执行文件	`chmod +x`后直接运行	证明跨平台能力
`demo_config.json`	JSON	预置3台设备（投影仪、灯光、音响）、2条规则（入场启动、离场关闭）	确保答辩现场5分钟内跑通全流程
`test_report.pdf`	PDF	含压力测试（100条指令并发）、断网测试（拔网线3分钟）、长时间运行（72小时日志摘要）	用数据说话，破除“可能行”质疑
`user_manual.md`	Markdown	5分钟上手指南：如何改设备IP、如何加新规则、如何查日志	展示工程化思维

提示：test_report.pdf不是截图拼凑，而是用脚本自动生成。例如压力测试：python stress_test.py --count 100 --concurrent 10，输出CSV后用Pandas绘图，结论写“99.8%指令在200ms内完成，0失败”。

5.2 答辩现场演示的黄金10分钟脚本

别自由发挥，按脚本走，确保可控：

第0-1分钟：环境介绍
“老师好，这是基于TCP/IP的展厅智能中控软件，运行在Windows 10和Ubuntu 22.04双平台，核心是可靠连接、指令闭环、断网韧性。现在演示Windows版。”
第1-3分钟：基础功能演示
双击exhibition-control-win.exe→ 打开浏览器访问http://localhost:8080→ 进入设备监控页，指出“投影仪在线、灯光在线、音响在线” → 点击“灯光”行的“调暗至30%”按钮 → 界面实时变色，同时用手机拍到真实灯光变暗（提前布好摄像头）。
第3-5分钟：规则联动演示
切换到规则管理页 → 点击“入场启动”规则的“启用”开关 → 回到手动控制页，点击“触发传感器A”按钮（模拟红外感应） → 投影仪亮起、灯光调暗、音响播放欢迎语音（提前录好3秒音频）。
第5-7分钟：断网韧性演示
拔掉笔记本网线 → 点击“关闭投影仪” → 界面显示“已发送（离线）” → 等待10秒 → 插回网线 → 查看日志页，滚动到最新，显示“重发成功，req_id=REQ_...” → 投影仪熄灭。
第7-10分钟：故障排查演示
故意删掉config/devices.json中的投影仪IP → 重启程序 → 打开Web界面，设备列表显示“投影仪：离线（连接失败）” → 打开config/devices.json修复IP → 点击界面右上角“重载配置” → 10秒后状态变为“在线”。

5.3 答辩问答预判与应答策略

老师最爱问三类问题，我让每个学生背熟答案：

Q1：TCP长连接万一断了，你怎么保证指令不丢？
→ 不说“我们用了心跳”，要说：“指令发出去后，客户端启动ACK监听，超时未收到则自动重试3次，每次间隔300ms；所有重试记录进command_log表，可通过API实时查询状态。这是send_with_ack()方法的核心逻辑。”

Q2：如果展厅新增一个设备，你要改多少代码？
→ 不说“改一点”，要说：“只需三步：1. 写一个继承DeviceDriver的新类（如NewScreenDriver），实现connect/send_command/parse_response；2. 在config/devices.json中注册该设备；3. 在rules/下加一条新规则。核心调度器、通信层、Web界面一行代码都不动。”

Q3：你们怎么测试断网场景？
→ 不说“拔网线”，要说：“用iptables在Linux上模拟网络分区：sudo iptables -A OUTPUT -d 192.168.1.100 -j DROP（屏蔽设备IP），持续3分钟，验证离线队列是否正确堆积、恢复后是否重发、日志是否完整记录。”

从那以后我每次带毕设，都强制学生在答辩前一周，用这份清单逐项打钩：可执行文件能跑、配置文件能改、规则能加、断网能测、报告能生成。不是为了应付老师，而是确保当答辩现场网络抽风、设备掉线、老师突然拔网线时，你能笑着点开日志页，指着那行绿色的COMPLETED说：“看，它一直都在工作。”

希望帮到你。

本文还有配套的精品资源，点击获取