企业官网建设流程全解析

简介：围绕射频识别技术搭建的仓库管理项目源码包，面向物联网、物流管理及软件工程方向的学习者，可用于课程设计、毕业设计或同类系统二次开发参考。系统把RFID自动采集融入到货检验、入库分配库位、库存变动记录、库存查询和出库等流程，着力解决手工录入效率低、库存数据滞后与准确性不足的问题。压缩包共42个文件，约1.46MB，以12个C++头文件、10个源文件、8个PHP脚本及4个Qt工程、4个界面文件为主，另含说明文档与编译库文件，能反映客户端界面、业务逻辑、服务端接口和依赖库的组织方式。已有931人学习下载。读者可借此梳理RFID仓储业务的模块划分、前后端数据交互与库存管理落地思路，并参考界面分层、查询终端及入库管理等场景的实现，形成可复用的系统设计参照。

1. 盘点误差 3% 不是人的问题：RFID 仓库管理系统改掉的是哪一段

某公司的月度盘点，四个人拿着纸质单子扫了两天货架，账实差异还是 3.2%。后来他们换成基于射频识别技术的仓库管理系统，同一批货、同一批人，盘点时间压到 40 分钟，差异降到 0.4%。差别不在人勤快不勤快，而在数据采集那一段被换掉了：条码必须“看见才能读”，RFID 是“进到场里就能读”。

这套仓库管理系统源码包做的事很具体——把电子标签的 EPC 编码规则、UHF 读写器的盘存回调、入库出库移库盘点的业务状态机、库存台账的一致性控制，以及和上层 ERP 的对接接口，串成一条能跑通的链路。它适合三类人：正在做仓储数字化的后端或全栈开发者，想先拿一套参考实现验证方案可行的技术负责人，以及被“读不到、读重复、账不平”折腾过、想看看别人怎么写防抖和幂等的工程师。硬件选型、参数调优、现场踩坑，下面一章一章拆。

2. 系统骨架怎么搭：从标签到台账的五层链路与选型账

先把链路讲清楚，否则后面调参数全是蒙的。一张无源 UHF 标签贴到托盘上，它自己没有电池，靠读写器发出来的射频能量激活芯片，再把 EPC 区里那串比特用背散射的方式反射回去。读写器把反射信号解调成 EPC 字符串，通过网口或串口抛给上位机，上位机的采集服务再把它翻译成“哪个 SKU、哪一箱、在哪个库位、什么时间”。业务层拿到这条事件，才谈得上入库、移库、盘点、出库。

2.1 四段链路与频段选型：为什么仓库几乎只选 UHF

链路的四段是：标签 → 天线 → 读写器 → 采集服务。这四段里任何一段掉链子，表现都一样——“读不到”。所以排查时必须能分段定位：拿一张标签贴在天线正前方 30cm，如果这时候都读不到，问题在读写器或天线上，跟业务代码一点关系没有。

频段选型是第一个要算清楚的账。低频 125kHz 读距只有几厘米，胜在不怕金属和液体，工位防错常用；高频 13.56MHz 读距一米以内，适合托盘和单据绑定、门禁；超高频 860–960MHz 才是仓库群读的主力，读距 3–8 米，一秒能读几百张，标签单价也最低。代价是对金属和液体敏感，贴在金属货架上的标签要做抗金属处理，或者垫一层隔离材料。

选完频段还要选协议。仓库场景基本锁定 EPC Class 1 Gen2（常说的 Gen2），因为它的盘存轮次机制天生为群读设计：读写器发 Query，标签在若干个时隙里随机挑一个回应，Q 值决定时隙数量。Q 值太小，标签一起抢答，冲突多；Q 值太大，时隙空转，盘存一轮时间被拉长。这个参数后面第 4 章会单独讲。

还有一个容易忽略的是 Session。Gen2 有 S0–S3 四个会话，每个会话维护一个标签的 A/B 状态标志。用 S0 时，标签被读到一次就翻状态，读写器必须来回切 A/B 才能连续读；用 S2 或 S3 时标签状态保持时间长，适合“同一批货反复扫”的通道门场景。很多现场“第一遍读得到、第二遍读不到”，八成是 Session 选错了。

2.2 分层结构：源码包里的六个模块各自负责什么

这份源码包按职责切成六层，改动频率从下往上递减。摸清这张表，再打开目录就不会迷路。

采集层最忌讳被业务代码污染。我见过不少项目图省事，在串口回调里直接开数据库事务，结果标签密集上报时事务排队，串口缓冲区溢出，一次丢十几张。正确做法是采集层只做“收 + 入队”，把 EPC、天线号、RSSI、时间戳丢进内存队列或 Redis，业务层异步消费。

2.3 环境准备：依赖与初始化

这套项目对运行环境的要求不算高，Python 3.10 以上、PostgreSQL 14 以上、Redis 6 以上就能跑。数据库选 PostgreSQL 是因为后面要用到部分唯一索引和 JSONB 存原始上报，MySQL 在 8.0 之前做不了这类约束。

# 克隆后进入项目根目录，创建虚拟环境 python3 -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate # 安装运行依赖，requirements.txt 里已锁定大版本 pip install -r requirements.txt # 关键包：pyserial（串口读写器）、paho-mqtt（事件总线）、 # psycopg2-binary（PG 驱动）、pydantic（配置与报文校验）、apscheduler（定时盘点任务） # 初始化数据库结构，迁移脚本带版本号，可重复执行 python manage.py migrate --db warehouse # 灌入基础数据：库区、库位、SKU、读写器台账 python manage.py seed --config conf/seed.dev.yaml

参数说明两点：--db warehouse走的是配置文件里预设的连接别名，生产环境换成conf/prod.yaml里的连接串，不要把密码写进命令行；seed的配置文件决定了生成多少个库位和 SKU，开发环境默认 200 个库位够用，压测时调到 5000 才能真正暴露索引问题。

2.4 首次启动：把模拟读写器跑起来

没有硬件也能把整条链路跑通，这是这份源码包最实用的地方——它内置了一个模拟读写器，按设定速率随机吐出 EPC，行为跟真实设备一致。

# conf/reader.dev.yaml readers: - id: R001 name: 收货口通道门 driver: simulator # 无硬件时用 simulator；真实设备填 serial 或 llrp antennas: [1, 2] # 该读写器接了两路天线 power_dbm: 26 # 读功率，先给中等值，后面对照 4x 节扫描 session: 2 # S2，通道门连续盘存用得到 q_value: 4 # 时隙数量级，标签密集时往上调 emit_per_second: 40 # 仅模拟器生效，模拟每秒 40 条上报

# 启动采集服务，只负责收数入队 python manage.py run_reader --config conf/reader.dev.yaml # 另开一个终端启动业务消费者，把事件落成入库流水 python manage.py run_consumer --queue rfid.events --workers 4

power_dbm是读功率，国内常用设备上限多在 30dBm 左右，不要一上来拉满，功率越大读得越远，串读隔壁通道的概率也越大。session填 2 是我们前面说的 S2。emit_per_second只对模拟器有意义，真实设备不认这个键，改成serial驱动时它会自动忽略。

提示：模拟器跑通只能证明代码逻辑对，不能证明现场能读。上真机之前，务必先用设备厂商自带的调试工具确认标签能被读到，再回来接系统，否则你会花半天怀疑自己的代码。

3. 核心链路落地：EPC 编码、盘存回调与库存台账的代码实现

3.1 EPC 编码规则：96 位里到底塞什么

标签的 EPC 区常见容量是 96 位，也就是 24 个十六进制字符。这 24 个字符怎么切分，全项目组必须提前定死并写进文档，因为标签一旦贴到货上、发到全国各仓，改编码规则等于重贴一遍所有标签。

我们采用的切分是：2 字节前缀（区分项目与标签批次）+ 4 字节 SKU 编码 + 4 字节序列号 + 2 字节校验。别把 SKU 用自增主键，因为主键会随环境重建而变，换成固定长度的品类码更稳。

from dataclasses import dataclass PREFIX_BYTES = 2 # 标签批次号，换供应商时递增 SKU_BYTES = 4 # 品类码，全公司唯一，与 ERP 物料编码映射 SERIAL_BYTES = 4 # 同品类下的序列号，够 42 亿个 CRC_BYTES = 2 # 前 10 字节的 CRC16，防手输错、防脏数据 @dataclass class EpcCode: prefix: int sku_code: int serial: int def encode(self) -> str: body = ( self.prefix.to_bytes(PREFIX_BYTES, "big") + self.sku_code.to_bytes(SKU_BYTES, "big") + self.serial.to_bytes(SERIAL_BYTES, "big") ) crc = crc16_ccitt(body) # 只算前 10 字节 return (body + crc.to_bytes(CRC_BYTES, "big")).hex().upper() @classmethod def decode(cls, epc: str) -> "EpcCode": raw = bytes.fromhex(epc) if len(raw) != PREFIX_BYTES + SKU_BYTES + SERIAL_BYTES + CRC_BYTES: raise ValueError(f"EPC 长度不对: {epc}") body, crc = raw[:-CRC_BYTES], raw[-CRC_BYTES:] if crc16_ccitt(body) != int.from_bytes(crc, "big"): raise ValueError(f"EPC 校验失败: {epc}") # 直接丢弃，不进业务表 # 按固定偏移切片，禁止用字符串 index 定位，长度一变就全错 return cls( prefix=int.from_bytes(body[0:2], "big"), sku_code=int.from_bytes(body[2:6], "big"), serial=int.from_bytes(body[6:10], "big"), )

这段代码有两个设计取舍值得说。一是把校验放在 EPC 里而不是数据库里，因为最脏的数据往往在入库之前就产生了——手持机误扫、贴错标签、隔壁仓库的货串读进来，decode抛异常就把它们挡在业务层外面，台账干净得多。二是校验用 CRC16-CCITT 而不是简单的累加和，前者对连续位翻转更敏感，实测能拦住大部分贴标机打印残缺的标签。

sku_code与 ERP 物料编码的映射单独建一张表，不要试图把 13 位物料编码硬塞进 4 字节，塞不下也不该塞。映射表跟着 SKU 主数据一起同步，一天一次全量、增量走消息，量不大。

3.2 读写器接入：把厂商差异关进一个抽象类

现场的读写器往往不止一个品牌，收货口可能是 A 家，叉车上是 B 家，手持机又是第三家。如果业务代码里到处写if brand == 'A'，后面每换一台设备都是一场灾难。做法是定义一个最小接口，只暴露start / stop / on_report，把协议差异关在驱动里。

import queue, threading, time class BaseReader: """所有读写器驱动的最小契约：启动、停止、上报回调。""" def __init__(self, cfg: dict): self.cfg = cfg self.queue: queue.Queue = queue.Queue(maxsize=20000) # 有界队列，防止内存涨爆 self._stop = threading.Event() def start(self): raise NotImplementedError def stop(self): self._stop.set() def _emit(self, epc: str, antenna: int, rssi: float): # 采集层只做一件事：打包原始上报入队，绝不在这里写库 record = {"epc": epc, "reader_id": self.cfg["id"], "antenna": antenna, "rssi": rssi, "ts": int(time.time() * 1000)} try: self.queue.put_nowait(record) except queue.Full: # 队列满说明下游消费跟不上，丢最旧的（盘存是幂等的，丢几条不影响结果） self.queue.get_nowait() self.queue.put_nowait(record) class SerialReader(BaseReader): """串口读写器驱动，多数国产设备走这套 ASCII 协议。""" def start(self): import serial self.port = serial.Serial(self.cfg["port"], self.cfg["baud"], timeout=0.2) self._send_cmd(f"SET_POWER {self.cfg['power_dbm']}") self._send_cmd(f"SET_SESSION {self.cfg['session']}") self._send_cmd(f"SET_Q {self.cfg['q_value']}") self._send_cmd("START") threading.Thread(target=self._loop, daemon=True).start() def _loop(self): while not self._stop.is_set(): line = self.port.readline().decode("ascii", "ignore").strip() if not line.startswith("UPLOAD"): continue # 形如：UPLOAD,0001,E28011700000020A1B2C3D4,E,32 _, ant, epc, _pc, rssi_raw = line.split(",") rssi = (int(rssi_raw) - 128) / 2.0 # 厂商用 0~255 编码，换算成 dBm self._emit(epc.upper(), int(ant), rssi)

这里的关键点有三个。第一，队列必须有界，maxsize=20000是经验值，通道门高峰期一秒几百条，下游偶尔卡顿两秒也不会丢；如果真的满了，丢最旧的而不是阻塞，因为盘存事件天然幂等，同一条 EPC 后面还会重复上报。第二，RSSI 的原始值各厂商编码方式不同，有的给 0–255，有的直接给负 dBm，换算写进驱动，不要让业务层去猜。第三，_loop用独立线程读串口，主线程只负责生命周期，不能用同步轮询，否则定时任务一来就把读串口的节奏卡住。

RSSI 这个值在系统里的作用被很多人低估。它不参与业务判断，但它是“距离指纹”：同一个天线读到同一个 EPC，RSSI 从 -40 掉到 -70，说明货被挪远了或者标签被遮挡了。后面 4.2 节的去重和 5.2 节的误读排查都要靠它。

3.3 入库事务：一条 EPC 只能落一次

业务层最容易翻车的地方，是把“读到了”直接等同于“入库了”。实际现场一条 EPC 在通道门会被重复上报几十次，网络抖动还会导致采集服务重启后重放。所以入库逻辑必须做到：同一条 EPC 对同一张单据，重复调用没有任何副作用。

from datetime import datetime from sqlalchemy import text def inbound_scan(session, doc_no: str, epc: str, operator: str): """单条扫码入库。幂等键 = (doc_no, epc)。""" code = EpcCode.decode(epc) # 校验失败的标签在这里被挡掉 # 先落原始事件，即使后续业务校验失败也有据可查 session.execute(text(""" INSERT INTO scan_event(doc_no, epc, sku_code, serial, operator, created_at) VALUES (:doc, :epc, :sku, :sn, :op, :ts) ON CONFLICT (doc_no, epc) DO NOTHING -- 重复上报直接忽略 """), {"doc": doc_no, "epc": epc, "sku": code.sku_code, "sn": code.serial, "op": operator, "ts": datetime.utcnow()}) # 库存扣减用条件更新，避免并发超卖/超收 updated = session.execute(text(""" UPDATE stock SET qty = qty + 1, version = version + 1, updated_at = :ts WHERE location_id = :loc AND sku_code = :sku AND version = :ver """), {"loc": resolve_location(session, doc_no), "sku": code.sku_code, "ver": current_version(session, doc_no)}).rowcount if updated == 0: raise ConcurrencyConflict(f"库存版本冲突 doc={doc_no} sku={code.sku_code}") session.execute(text(""" INSERT INTO inbound_line(doc_no, epc, qty, status) VALUES (:doc, :epc, 1, 'RECEIVED') """), {"doc": doc_no, "epc": epc})

逻辑说明：scan_event上的唯一索引(doc_no, epc)是幂等的第一道闸门，重复上报在数据库层被吞掉，业务代码不用写去重判断；stock表用version字段做乐观锁，条件更新返回 0 行就说明有人抢先改了，抛冲突让上层重试。resolve_location从单据头拿目标库位，综合作业单也可以改成从上报的天线号反推库区。

参数上要盯的是重试策略：冲突重试最多三次，间隔 50ms、150ms、450ms 退避，三次都失败说明这条 SKU 在这个库位上被高频争抢，通常是有人把同一个货位分配给了两张并行作业单，得回去查分配逻辑，而不是把重试次数调大。

3.4 表结构与库存一致性：三条约束撑住台账

SQL 层面真正管用的约束就三条：事件表的(doc_no, epc)唯一、库存表的(location_id, sku_code)唯一、快照表的按日分区。

-- 原始扫码事件，只增不改，出问题靠它复盘 CREATE TABLE scan_event ( id BIGSERIAL PRIMARY KEY, doc_no VARCHAR(32) NOT NULL, epc VARCHAR(24) NOT NULL, sku_code INTEGER NOT NULL, serial INTEGER NOT NULL, reader_id VARCHAR(16), antenna SMALLINT, rssi NUMERIC(5,1), operator VARCHAR(32), created_at TIMESTAMPTZ NOT NULL DEFAULT now(), CONSTRAINT uq_scan UNIQUE (doc_no, epc) -- 幂等靠它，不靠应用层判断 ); CREATE INDEX idx_scan_epc_time ON scan_event (epc, created_at DESC); -- 追溯单件货用 -- 实时库存，一行一个库位一个品类 CREATE TABLE stock ( location_id INTEGER NOT NULL, sku_code INTEGER NOT NULL, qty INTEGER NOT NULL DEFAULT 0 CHECK (qty >= 0), version INTEGER NOT NULL DEFAULT 0, updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY (location_id, sku_code) ); -- 每日快照，盘点差异和报表都从这走，避免全表扫事件表 CREATE TABLE stock_snapshot ( snap_date DATE NOT NULL, location_id INTEGER NOT NULL, sku_code INTEGER NOT NULL, qty INTEGER NOT NULL, PRIMARY KEY (snap_date, location_id, sku_code) ) PARTITION BY RANGE (snap_date);

CHECK (qty >= 0)这条约束看着简单，能救很多次事故：出库逻辑写错、并发扣减没加锁，都会被数据库当场拦下，而不是等到月末账实不符才发现。分区表按月建分区，stock_snapshot一年也就几百万行，不分区的话报表查询会越来越慢。

注意：不要用SELECT ... FOR UPDATE锁整行去做入库累加，这在高并发扫码下会直接把连接池打满。乐观锁加重试在这个场景下吞吐更好，因为真正冲突的比例很低。

3.5 出库复核与防错发

出库比入库危险，因为发错货的代价是真金白银。这里的做法是把“应该发什么”和“实际发了什么”做集合比对，而不是逐条判断。

def outbound_verify(session, doc_no: str) -> dict: expected = set(session.execute(text( "SELECT epc FROM outbound_line WHERE doc_no=:d AND status='PICKING'" ), {"d": doc_no}).scalars()) scanned = set(session.execute(text( "SELECT epc FROM scan_event WHERE doc_no=:d AND event_type='OUTBOUND'" ), {"d": doc_no}).scalars()) missing = expected - scanned # 该发没发，拦住不放行 extra = scanned - expected # 不该发却扫了，几乎都是串读 return {"ok": not missing and not extra, "missing": sorted(missing), "extra": sorted(extra)}

missing非空说明漏拣，extra非空大概率是通道门把隔壁月台的货串读了进来。这两种情况的处理方式完全不同：漏拣要回去找货，串读要做软件侧的时间窗过滤（见 4.2 节），不要混为一谈。

4. 参数调优与联调：把读取率从 82% 拉到 99.2% 的六个旋钮

读取率是这套系统唯一的硬指标。我们在一间 800 平米的成品仓做过对照：默认参数下通道门读取率 82%，逐项调完到 99.2%，两天时间。这一章就是那两天的记录。

4.1 六个旋钮：功率、Q 值、Session、天线角度、标签位置、移动速度

表格里最容易做错的是功率。几乎所有人的第一反应都是“读不到就加大功率”，结果通道门功率拉到 30dBm，把二十米外月台上的货全读进来，系统里多出一堆莫名其妙的入库记录。正确顺序是反过来：先把功率压到最低，保证一张标签贴在天线正前方 50cm 能稳定读到，再逐步加到覆盖整个门宽，找到临界点后回退 1dBm 留余量。

Q 值的调整要配合实测。写个小脚本把不同 Q 值下的单轮盘存耗时和读取张数打出来，比拍脑袋强。

import time from statistics import mean def sweep_q(reader, tags_expected: int, rounds: int = 20): """对每个 Q 值跑若干轮盘存，输出读取率与平均耗时。""" results = [] for q in (3, 4, 5, 6, 7): reader.set_q(q) rates, costs = [], [] for _ in range(rounds): reader.clear() # 清空本地缓存，避免上一轮残留 t0 = time.perf_counter() seen = reader.inventory_once(timeout=2.0) costs.append(time.perf_counter() - t0) rates.append(len(seen) / tags_expected) results.append({"q": q, "read_rate": round(mean(rates), 4), "avg_cost_ms": int(mean(costs) * 1000)}) # 读取率相同的情况下优先选耗时短的，通道门场景时间就是吞吐 return sorted(results, key=lambda r: (-r["read_rate"], r["avg_cost_ms"]))

inventory_once是驱动里封装的单轮盘存，超时 2 秒是为了防止 Q 值过大时空转太久。tags_expected是这一轮实际放进场的标签数，读取率才有分母。这个脚本跑完你会看到一个很典型的拐点：Q 从 3 到 5 读取率明显上升，5 到 7 基本不变但耗时翻倍，最优值就在拐点上。

4.2 去重与防抖：时间窗、RSSI 阈值、状态机三重过滤

一条 EPC 在通道门被上报几十次是常态，但业务上只应该产生一次“通过”事件。光靠数据库唯一索引能防重复入库，防不住“同一托盘在门口晃了三次产生三条出库记录”这种逻辑重复。

第一重是时间窗。同一个 EPC 在 N 秒内只认第一次，N 取多少取决于作业节奏：通道门 3 秒够用，手持盘点要 1 秒以内，不然连着扫两箱同款会漏。

import time from collections import OrderedDict class Deduper: """滑动时间窗去重。用 OrderedDict 维护插入顺序，天然适合做过期清理。""" def __init__(self, window_sec: float = 3.0, max_keys: int = 50000): self.window = window_sec self.max_keys = max_keys self._seen: OrderedDict[str, float] = OrderedDict() def accept(self, epc: str, ts: float | None = None) -> bool: now = ts or time.time() # 先清理过期键，避免内存一直涨 while self._seen: oldest_epc, oldest_ts = next(iter(self._seen.items())) if now - oldest_ts <= self.window: break self._seen.popitem(last=False) if epc in self._seen: return False # 窗口内重复，丢弃 self._seen[epc] = now if len(self._seen) > self.max_keys: # 兜底，防止异常流量打爆内存 self._seen.popitem(last=False) return True

第二重是 RSSI 阈值。通道门场景下，真正“过门”的标签 RSSI 通常在 -55dBm 以上，而隔壁通道串过来的多在 -70dBm 以下。把阈值卡在 -65dBm，能砍掉大部分串读。阈值不能写死，因为不同天线的增益不一样，上现场第一天就该拿实际数据画个分布图再定。

第三重是状态机。入库单有CREATED → RECEIVING → RECEIVED三个状态，出库单有PICKING → CHECKED → SHIPPED。事件只在特定状态下才被接受，比如单据已经SHIPPED了还收到出库扫码，直接进异常表等人处理，而不是悄悄追加一条。

三重的顺序不能乱：先时间窗（最便宜，纯内存），再 RSSI（一次数值比较），最后状态机（要查库）。反过来会把数据库打成瓶颈。

4.3 与 ERP 的接口契约：幂等键和补偿队列

和上层系统对接，出事基本都出在“重复推送”和“推送丢失”上。约定三条：每条业务事件带全局唯一event_id；接收方必须按event_id幂等；发送方必须有本地补偿队列，失败重试。

import json, uuid def build_inbound_event(doc_no: str, lines: list[dict]) -> dict: return { "event_id": str(uuid.uuid4()), # 接收方按它做幂等，重推不重复记账 "event_type": "INBOUND_CONFIRMED", "doc_no": doc_no, "occurred_at": "2024-06-01T10:32:11+08:00", "source": "rfid-wms", "lines": [{"epc": l["epc"], "sku_code": l["sku_code"], "qty": l["qty"], "location": l["location"]} for l in lines], } def publish_with_retry(mq, event: dict, max_retry: int = 5): for attempt in range(max_retry): try: mq.publish("erp.inbound", json.dumps(event), qos=1) return True except Exception as e: # 指数退避，最后一次失败落本地补偿表，由定时任务捞起来重发 time.sleep(min(2 ** attempt * 0.2, 5)) save_to_compensation(event) return False

qos=1表示至少一次投递，配合event_id幂等，效果就是“不丢不重”。save_to_compensation落的是本地表，定时任务每五分钟扫一次，超过 24 小时还没成功的报警。这套组合拳比“调大超时时间”靠谱得多，血泪经验是超时调得再大也挡不住对面系统停机维护。

4.4 联调顺序：别一上来就上量

我一般按这个顺序推进，每一步都能单独验证，出错时定位范围小：先用模拟器跑通全链路，确认事件能从驱动流到数据库；再上单台真机、单张标签，验证编码解码和 RSSI 换算；然后十张标签静态摆放，扫 Q 值和功率；接着动态过门，测读取率和重复率；再并联两台读写器，看串读；最后全量 SKU 压测，跑一天看台账是否平。

跳过其中任何一步都可能返工。最常见的是直接上量，结果读到一半发现 RSSI 换算是错的，前面所有调参记录全部作废。

5. 避坑与排查：六个让上线延期两周的现场问题

5.1 硬件与现场类问题

现象一：同一批货在收货口读得到，到了成品仓通道门就漏读三成。原因：成品仓的货架是金属的，标签贴在托盘侧面正对货架，射频能量被金属面反射抵消，形成了驻波盲区。这不是功率不够，加大功率只会让串读更严重。 解决：标签改贴托盘对角两侧，或者换抗金属标签；天线不要正对货架面，改成从两侧 45 度斜打。调整后同一个门读取率从 71% 回到 97%。

现象二：通道门每小时产生几十条莫名其妙的入库记录，货根本没来过。原因：两台读写器装在相距 6 米的两个门，功率都设成 27dBm，波束互相覆盖。加上 Session 用了 S0，标签状态反复翻转，被反复上报。 解决：先把功率降到 22dBm 测覆盖，再把 Session 改成 S2，最后在接入层加 RSSI 阈值 -65dBm。三条一起上，误读降到每班次 2 条以内。

现象三：叉车快速通过时读取率断崖式下跌，慢走就正常。原因：盘存一轮需要时间，Q 值设成 6 时一轮要 300ms 以上，叉车 8km/h 通过 3 米宽的门口只用 1.3 秒，理论上够两轮，但标签在金属货叉附近被遮挡，实际有效轮次不到一轮。 解决：门口装地感或红外，触发后才开始盘存并且把 Q 降到 4 缩短单轮时间；叉车限速 5km/h。读取率从 68% 到 95%。

5.2 软件与数据类问题

现象四：采集服务重启后，前一分钟的数据全部重复入库。原因：采集层用了 Redis List 做队列，消费端LPOP之后才处理，处理到一半崩了，消息已经出队，重启后只能靠上游重推，而重推没有幂等键。 解决：换成RPOPLPUSH到处理中的临时队列，处理成功再LREM，崩了重启从临时队列恢复；同时把入库的幂等键(doc_no, epc)唯一索引补上，双保险。

现象五：月末盘点，系统账面比实物多出 200 多件。原因：stock表的qty只在入库加、出库减，但移库时只写了流水没更新两个库位的余额，日积月累就飘了。 解决：移库做成一个事务里的两条UPDATE，加CHECK (qty >= 0)约束；再补一个每日对账任务，用事件表重算余额和stock比对，不一致就告警。上线后第一时间发现了这个 bug，避免了几万块的损失。

现象六：手持机盘点时，同款商品连着扫两箱只记了一箱。原因：去重时间窗设成了 3 秒，跟通道门共用一套配置。手持机连续扫同款，两条 EPC 不同但业务上是两箱，被窗口误杀了。 解决：去重窗口按设备类型分开配置，手持机 0.8 秒、通道门 3 秒；并且去重的键用 EPC 而不是 SKU，这是当初设计就定对的，没改。

排查这类问题的通用套路是：先在scan_event里按时间范围把原始上报捞出来，看是“没上报”还是“上报了没落业务表”。前者查硬件和驱动，后者查幂等和状态机。这一步能把排查范围砍掉一半。

6. 验收与进阶：用压测脚本把系统跑穿的三个指标

上线前我最不愿意省的一步是压测，因为仓库系统的故障几乎都发生在“量突然上来”的时候——大促前集中收货、月末集中出库、盘点日全员同时扫。写一个能重复跑的压测脚本，比开十次评审会管用。

import random, time, threading def stress(reader, duration_sec=300, workers=4, epc_pool_size=5000): """用模拟器灌数据，观察事件吞吐与库存一致性。""" pool = [EpcCode(prefix=1, sku_code=random.randint(1000, 9999), serial=i).encode() for i in range(epc_pool_size)] stats = {"sent": 0, "accepted": 0, "conflict": 0, "dropped": 0} lock = threading.Lock() def worker(): end = time.time() + duration_sec while time.time() < end: epc = random.choice(pool) # 有意制造重复，验证幂等 ok = reader.inject(epc) with lock: stats["sent"] += 1 stats["accepted" if ok else "dropped"] += 1 time.sleep(0.005) threads = [threading.Thread(target=worker) for _ in range(workers)] [t.start() for t in threads] [t.join() for t in threads] return stats

epc_pool_size=5000模拟 5000 个不同 SKU 的标签，workers=4对应四路天线同时上报，sleep(0.005)让单线程每秒发 200 条，四路合计 800 条/秒，比真实通道门峰值高一倍。刻意从池子里随机取，是为了让重复率接近现场。

跑完之后看三个指标，比看 CPU 占用有意义得多：

第三条最容易被忽略。压测结束后一定要拿scan_event里去重后的 EPC 数量，和stock各库位累加值做一次全量对账，差一件都要查到底。我吃过这个亏：压测跑得漂漂亮亮，延迟和吞吐全部达标，结果上线第一周盘点差了三十多件，回头才发现是移库分支在并发下的版本冲突被吞掉了异常，日志里只有一行 warning。

从那以后我每次压测都强制走一遍“灌数 → 停服 → 全量对账 → 重启补偿”，不管时间多紧都不跳。这套流程看着笨，但它能把九成的库存一致性 bug 挡在上线之前。希望帮到你。

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