☰
排队叫号系统源码:基于SQLite与状态机的可靠实现
2026/10/12 11:43:05 网站建设 项目流程

简介:一份C#排队机叫号系统源代码,面向C#初学者以及正在为排队机、叫号机设计而苦恼的开发者,提供一套可整体参考的商用级项目实现。代码经过客户多年实际使用未出故障,其稳定性足可作为学习蓝本;系统模块划分较完整,涵盖取号排队、窗口呼叫、屏幕展示、数据存取等主要环节,能帮助读者快速看懂排队业务的流转过程与WinForm界面交互方式。包体为RAR压缩包,共249个文件、1.29MB,体量精简但结构丰富。172个.cs源文件承载核心业务逻辑,44个.resx文件保存窗体布局、控件属性与本地化资源,9个.csproj工程文件、9个.ico图标以及2个.rdlc报表定义,一并补齐编译、显示和打印所需的模块,目录清晰,便于逐段查阅。目前已有1617人学习下载。和入门级示例不同,它保留了数据库操作封装、公共工具类、屏幕显示窗体等多类可直接改动的源码模块,适合用来自学项目分层与数据访问;默认界面可自行替换,作为排队叫号系统二次开发或课程设计起点都很实用。

1. 排队机叫号到底难在哪:一张小票背后的状态流转

很多人以为排队机叫号就是把号打出来、叫下一个,真正动手写过源码的人才知道,难点从来不在打印小票,而在状态管理。营业厅、政务大厅、医院药房里的那套"取号→窗口叫号→过号重叫→办结"流程,落到代码里就是一张票从生到死的状态流转。这个方向适合要给内部做轻量叫号系统的开发者,也适合想抄一套可靠源码的初学者。反直觉的结论是:别急着上分布式,先把状态机画对。否则并发一上来,不是重复叫号就是丢号,线上翻车时你连问题都复现不出来。下面的实现,是我接过几个类似小需求后沉淀下来的最小可靠方案,全部代码围绕 SQLite 和状态字段展开,换 MySQL 也只改一行连接串。

2. 叫号系统数据模型:为什么不能用内存队列硬扛

2.1 一张 queue_ticket 表,把取号、叫号、过号全装下

常见做法是一上来就用 Redis List 或者 Python 的 deque,认为叫号就是"从队头弹一个"。这个思路在演示 Demo 里跑得通,业务一复杂就露馅:过号的人要重叫,窗口要指定叫某个号,领导要查今天的排队记录,服务一重启队列全丢。队列里存的不该是"一个人",而是一张票的完整状态。

我一般用一张关系表做主存储,选 SQLite 是因为单文件、零运维,一个营业厅十几个窗口的并发量完全扛得住;将来要上 MySQL,SQL 基本不用改,只换连接层。核心字段见下表:

字段类型说明
idINTEGER 主键自增全局有序,取号顺序靠它
ticket_noTEXT 唯一索引展示给客户的号码,如 A048
statusINTEGER0 待叫 / 1 已叫 / 2 过号 / 3 完成 / 4 作废
window_idTEXT当前呼叫的窗口编号
created_atTEXT取号时间,ISO 格式
called_atTEXT最近一次叫号时间
finished_atTEXT办结时间

status 用整数不用字符串,是为了让状态跳转能用一条 UPDATE ... WHERE status=? 完成,天然带并发保护。ticket_no 必须加唯一索引,这是后面防重复取号的最后一道保险。时间存 ISO 字符串而不是时间戳,是为了调试时人眼能直接读,排序在 SQLite 里字符串和数字效果一致。

建表语句如下,注意唯一索引和 WAL 模式:

CREATE TABLE IF NOT EXISTS queue_ticket ( id INTEGER PRIMARY KEY AUTOINCREMENT, ticket_no TEXT NOT NULL, status INTEGER NOT NULL DEFAULT 0, window_id TEXT, created_at TEXT NOT NULL, called_at TEXT, finished_at TEXT ); CREATE UNIQUE INDEX IF NOT EXISTS idx_ticket_no ON queue_ticket(ticket_no); PRAGMA journal_mode=WAL;

WAL 模式值得多说一句:它让读操作不阻塞写操作,多线程取号和查状态同时进行时不会报 database is locked。代价是会产生 -wal 和 -shm 两个附属文件,备份时要把它们一起带上,否则数据可能不是最新的。

2.2 取号接口的最小实现:编号生成与并发兜底

取号的核心是生成"当天序号"。最常见的写法是 SELECT COUNT(*) + 1,但并发取号时两个请求同时查到同一个 COUNT,就会打出两张一样的票。所以我的做法是:编号生成照做,但数据库唯一索引在底下兜底,撞号就重试。

import sqlite3 from datetime import datetime DB_PATH = "queue.db" def get_conn(): conn = sqlite3.connect(DB_PATH, timeout=5) conn.execute("PRAGMA journal_mode=WAL") return conn def take_number(prefix="A"): """取号:生成并插入一张新票,返回票号""" conn = get_conn() try: today = datetime.now().strftime("%Y-%m-%d") with conn: cur = conn.execute( "SELECT COUNT(*) FROM queue_ticket " "WHERE ticket_no LIKE ? AND created_at LIKE ?", (prefix + "%", today + "%") ) seq = cur.fetchone()[0] + 1 ticket_no = f"{prefix}{seq:03d}" conn.execute( "INSERT INTO queue_ticket (ticket_no, status, created_at) " "VALUES (?, 0, ?)", (ticket_no, datetime.now().isoformat()) ) return ticket_no except sqlite3.IntegrityError: # 并发撞号,重新生成 return take_number(prefix) finally: conn.close()

这段代码的逻辑是:先数当天同前缀有多少张票,再加一得到新序号,插入成功后把票号返回给前端打印。第一遍 SELECT 不是线程安全的,但 INSERT 撞了唯一索引会抛 IntegrityError,递归重试就能拿到下一个可用号。

两个参数需要按现场调:prefix 是号别前缀,常见场景里有 A 普通号、B 老年号、C 预约号,不同队列互不干扰;seq 的位数决定了号面宽度,03d 会补成 A048,超过 999 后会自动变四位数,不用你管。失控点是递归重试——极端情况下可能栈溢出,生产环境请把递归改成 for 循环,最多重试三次,三次还撞说明表里数据异常,要告警而不是闷头重试。

还有一条血泪经验:小票打印必须放在事务提交之后。曾经有个项目把打印机调用放在事务内,打印机卡纸导致事务一直不提交,后面所有取号请求全部堵住。正确顺序是:先 INSERT 提交,拿到票号,再驱动打印机,打印失败就让窗口手动补打,数据库记录不能回滚。

2.3 窗口叫号接口:用条件 UPDATE 避免两个窗口抢同一单

窗口叫号最怕两个窗口同时按下"呼叫下一个",把同一个号叫走。新手写法是先 SELECT 取最早的待叫号,再 UPDATE 改成已叫,这两步之间有间隙,并发时就会抢到同一行。

解法是把"取出+锁定"合并成一条带条件的 UPDATE。SQLite 的 UPDATE 是串行执行的,同一个时刻只有一个请求能把 status 从 0 改成 1,另一个请求的 rowcount 会是 0,说明没抢到,重试即可。

def call_next(window_id, prefix="A"): """窗口叫号:取最早待叫号并标记为已叫,返回票号""" conn = get_conn() try: for _ in range(3): with conn: cur = conn.execute( """UPDATE queue_ticket SET status = 1, window_id = ?, called_at = ? WHERE id = ( SELECT id FROM queue_ticket WHERE status = 0 AND ticket_no LIKE ? ORDER BY id ASC LIMIT 1 ) AND status = 0""", (window_id, datetime.now().isoformat(), prefix + "%") ) if cur.rowcount > 0: cur = conn.execute( "SELECT ticket_no FROM queue_ticket WHERE id = " "(SELECT id FROM queue_ticket WHERE status = 1 " "AND window_id = ? ORDER BY called_at DESC LIMIT 1)", (window_id,) ) return cur.fetchone()[0] return None finally: conn.close()

这条 SQL 的关键在子查询和 WHERE status = 0 的叠加:子查询选出最早的待叫号 ID,外层 UPDATE 再确认这一行确实还是待叫状态。两个窗口并发时,只有一个 UPDATE 的 rowcount 是 1,另一个是 0,循环重试就会叫到下一个号。

窗口只能用自己所里配置的 prefix,避免 A 窗口把 B 队列的号叫走,这个过滤条件放在 ticket_no LIKE 里。如果你的现场所有窗口都能叫所有队列,去掉 LIKE 条件即可。叫号动作完成后,大厅屏那边通过查询 status = 1 且 called_at 最新的记录刷新"正在呼叫"。

3. 过号与重叫的完整状态机:源码里最容易写崩的两个动作

3.1 状态跳转表:先定规矩再写代码

服务端所有接口,本质上都是对 status 字段的受控修改。我每次动手前会先把跳转表画出来贴在代码注释里,所有函数照着表写,不要自由发挥。这个表也是评审代码时第一件要核对的东西。

当前状态允许的动作目标状态触发场景
0 待叫窗口叫号1 已叫正常呼叫
1 已叫标记过号2 过号叫了三遍没人来
1 已叫办理完成3 完成业务办结
2 过号窗口重叫1 已叫客户回来了
2 过号办理完成3 完成过号后直接办
0/1/2作废4 作废系统误操作、票面损坏

为什么一定要这张表?因为没有它,你会自然写出"过号就是把这条记录删掉"的代码。删掉一时爽,查账时发现号不连续,领导问你 A023 去哪了,你只能瞪眼。正确做法是过号只改状态,记录永远留在表里,今天的业务数据随时能回溯。

3.2 过号标记与重叫:绑定窗口,防止二次抢号

过号和重叫是这套源码里最容易写崩的两个动作。很多人把"重叫"实现成"把这张票重新插回队列尾",这是错的:客户人是回来了,但他的号一旦回到队尾,就会排到所有新取号的人后面,实体的队伍规则被代码改掉了,现场会吵起来。正确做法是把状态从 2 改回 1,让原窗口"再播报一次",号码不挪位置。

def mark_no_show(ticket_no): """过号:仅允许已叫状态的票变成过号""" conn = get_conn() try: with conn: cur = conn.execute( "UPDATE queue_ticket SET status = 2 " "WHERE ticket_no = ? AND status = 1", (ticket_no,) ) return cur.rowcount > 0 finally: conn.close() def recall(ticket_no, window_id): """过号重叫:绑定窗口,仅允许原窗口把过号票叫回""" conn = get_conn() try: with conn: cur = conn.execute( """UPDATE queue_ticket SET status = 1, window_id = ?, called_at = ? WHERE ticket_no = ? AND status = 2""", (window_id, datetime.now().isoformat(), ticket_no) ) return cur.rowcount > 0 finally: conn.close()

mark_no_show 的 WHERE 条件带 status = 1,保证只有"正在被呼叫"的票能过号。如果窗口对一个还没叫过的号点过号,rowcount 会是 0,接口直接告诉前端操作无效,这就拦住了一半误操作。

recall 的玄学在于必须绑定 window_id。真实场景里 A 窗口过号的客户,可能跑到 B 窗口问"轮到我没",B 窗口一查确实过号了,顺手就点重叫,结果人还在 A 那边,B 窗口白喊三遍。我在表结构里没有单独加归属字段,直接用 window_id 当归属,recall 里带上当前窗口号,谁过号谁重叫,跨窗口重叫交给现场人工协调。

重叫不改动队列顺序还有个额外好处:called_at 更新成当前时间,大厅屏按 called_at 排序展示时,这个号会跳到"正在呼叫"最前面,视觉上就是原窗口重新喊了一遍,符合直觉。

3.3 完成与作废:终态处理和大厅屏刷新

办结动作在代码里最不起眼,但它决定了这张票能否进入统计报表。完成和作废都是终态,终态之后任何状态跳转都必须失败,这是状态机设计的底线。

def finish_ticket(ticket_no): """办结:已叫或过号的票进入终态""" conn = get_conn() try: with conn: cur = conn.execute( "UPDATE queue_ticket SET status = 3, finished_at = ? " "WHERE ticket_no = ? AND status IN (1, 2)", (datetime.now().isoformat(), ticket_no) ) return cur.rowcount > 0 finally: conn.close() def void_ticket(ticket_no): """作废:未完成的票直接作废,保留记录""" conn = get_conn() try: with conn: cur = conn.execute( "UPDATE queue_ticket SET status = 4 " "WHERE ticket_no = ? AND status IN (0, 1, 2)", (ticket_no,) ) return cur.rowcount > 0 finally: conn.close()

finish_ticket 允许从已叫或过号直接办结,是因为现场经常出现"客户过号后直接来办,不需要再喊一遍"的情况。void_ticket 允许从待叫作废,用于取号机多打了一张空白票。两个函数都把终态排除在 WHERE 条件之外,所以已经完成或作废的票再调这些接口会返回 False,前端拿到 False 提示"该号已结束",不会产生脏数据。

大厅屏的数据源我一般不做复杂的推送,让窗口屏每三秒轮询一次接口,查最近一条 status = 1 的记录。几百个号以内的量级完全够用,轮询接口记得按 called_at 倒序,因为过号重叫会刷新 called_at,倒序才能让最新呼叫顶到最前面。

4. 排队叫号常见问题排查:编号重复、重复叫号、重启丢号

4.1 现象:并发取号时打出了两张相同票号

现象是两台取号机同时出票,一张 A048,另一张也是 A048,客户拿着号互相瞪眼。原因很直接:票号由"SELECT COUNT + INSERT"两步组成,两个请求在同一时刻读到相同 COUNT,生成相同序号。

解决靠三道防线:第一,调用端加分布式锁只能锁单机,多台取号机锁不住,所以不推荐;第二,靠数据库唯一索引兜底,撞号时 INSERT 抛异常,捕获后重新生成;第三,真上了高并发,把票号的序号来源换成数据库自增 ID,插入后拿 cur.lastrowid 格式化,彻底告别 COUNT 方式。我现在的主力做法是第三种,自增 ID 永远不会重复,连重试逻辑都省了。

4.2 现象:两个窗口同时叫到了同一个号

窗口端 A 和 B 同时按"呼叫下一个",语音播报同时喊出 A048。原因就是先 SELECT 再 UPDATE 的经典竞态,两窗口查到的都是最早的待叫号,然后各自 UPDATE 成功。

解决方法是第二章里 call_next 的写法:UPDATE 语句里带 WHERE status = 0,并把子查询放进 UPDATE,两个并发请求只有一个能命中,另一个 rowcount 为 0 后重新取号。这里有个排查技巧:如果线上还出现重复叫号,先看窗口端代码是不是绕过了服务端接口,自己用本地列表维护队列,那属于客户端逻辑和数据源不一致,必须强制窗口屏的所有状态变更都走服务端。

4.3 现象:过号重叫后,又被别的窗口抢走

W1 窗口叫了 A050 没人来,标记过号。五分钟后客户回来了,W1 点"重新呼叫",这时 W2 窗口正好也在点叫号,结果 A050 被 W2 叫走。

原因是重叫动作没有绑定原窗口,任何窗口都能对过号票执行重叫。解决方法是 recall 接口强制传入 window_id,并且只允许把状态为 2 且 window_id 等于当前窗口的票改成已叫。我的表结构里没有单独的"归属窗口"字段,window_id 第一次叫号时被写入,过号时保留,重叫时校验一致性,这是最简单可靠的方案。如果现场要求"谁都可以重叫",那就要在界面上加二次确认弹窗,否则这种抢号投诉会一直出现。

4.4 现象:服务一重启,排队数据全没了

重启之后所有客户的号都没了,这是用内存队列做存储的典型翻车现场。虽然很多人觉得"现场就几十个人,存内存够了",但营业厅的排队数据要留存到当天结束,中途服务发布、断电都要能恢复。

解决思路只有一个:所有状态变更落库,取号成功即提交事务。落库后还有另一个坑——恢复现场。我习惯在启动时加一段自检逻辑:查询 status IN (0,1,2) 的记录,把已叫状态全部改成过号,并记录一条日志"服务重启,未完成呼叫全部转过号"。这样重启后客户回来还能重叫,不会出现窗口喊了但库里没有的状态。

4.5 现象:大厅屏显示的号码和窗口实际叫的不一致

屏上写着"正在呼叫 A048",窗口语音已经在叫 A049。这种问题十有八九不在服务端,而在前端刷新时序:大厅屏轮询时正好赶上 A049 刚被叫出但还没提交,或者前端用了本地缓存排序。

排查顺序我建议先看数据库,查 status = 1 的记录按 called_at 倒序取第一条,如果库里是 A049 而屏幕显示 A048,那是前端渲染问题;如果库里也是 A048,那是叫号逻辑问题。解决前端问题,让显示端每次轮询都带上本地时间戳,服务端返回的数据如果比本地旧就丢弃,或者干脆不用缓存、每次都重新拉取。另外注意过号重叫会刷新 called_at,显示端判断"当前呼叫"必须用 called_at 而不是 ticket_no 大小,否则重叫的号永远排不到最前面。

5. 跑完代码这样验证:用并发脚本检查重复号和状态跳转

5.1 并发取号压测脚本

写完核心源码,别急着接界面,先用脚本把并发场景压一遍。下面的脚本模拟 10 个线程同时取号,每个线程取 20 个,然后查重复号:

import threading import queue_service as qs def worker(_): for _ in range(20): qs.take_number("A") threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)] for t in threads: t.start() for t in threads: t.join() conn = qs.get_conn() rows = conn.execute( "SELECT ticket_no, COUNT(*) c FROM queue_ticket GROUP BY ticket_no HAVING c > 1" ).fetchall() print("重复号检查:", rows if rows else "通过,无重复")

脚本跑完会输出两种结果:有重复号说明唯一索引兜底逻辑出了问题;无重复则说明并发取号稳定。10 个线程 200 个号对 SQLite 来说压力不大,建议再跑一轮 50 线程版本,观察取号耗时是否线性增长。

5.2 一条命令把叫号全流程走一遍

并发只是第一关,状态跳转的完整性要用一条脚本按真实业务路径走到底:取号、叫号、过号、重叫、完成,每个环节检查返回值。

t = qs.take_number("A") print("取号:", t) called = qs.call_next("W1", "A") print("叫号:", called) ok = qs.mark_no_show(t) print("过号:", ok) ok = qs.recall(t, "W1") print("重叫:", ok) ok = qs.finish_ticket(t) print("完成:", ok) # 验证终态不可再跳 ok = qs.finish_ticket(t) print("终态二次完成应失败:", ok)

最后一行是关键:finish_ticket 对已完成的状态应该返回 False。如果你发现终态还能被改,说明 WHERE 条件写漏了,这是状态机最常见的破口。我习惯再加一个 SQL 做全面巡检,统计每个状态的数量,期望结果是从大到小递减:待叫 0、已叫 0、过号 0、完成 200,任何状态出现负数或票号断档都要查。

这套验证方法本身不复杂,但它逼着我把所有接口的边界条件都过了一遍。后来我养成的习惯是:改任何状态相关代码,先跑一遍全流程脚本,再跑一遍并发脚本。状态机这种东西,黑匣子式的"看起来正常"最害人,等线上冒出重复号再回头查,成本是写代码时的十倍。希望帮到你。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询