简介基于Python的网络版中国象棋是一份面向Python初、中级开发者的网络编程实战源码覆盖套接字通信、并发处理、中国象棋规则和简单界面设计多个知识点。项目采用客户端与服务器分离的结构先在局域网内启动服务端程序再由两名玩家分别运行一次客户端并修改配置文件中的地址参数即可完成远程对弈。资源共31个文件压缩包约6.67MB主体为9个Python源文件包括服务端、客户端、配置、棋盘逻辑和界面控制等模块另有14个pyc编译文件以及图片、字体、文本等素材。已有2151人学习下载。解压后可按目录结构快速找到服务器与客户端入口对照学习数据包收发、状态同步和棋子走法判定过程也可直接修改代码扩展功能是巩固Python网络编程与游戏开发能力的实用练习项目。1. 基于Python的网络版中国象棋为什么自己写一局比想象中值很多人以为网络版中国象棋就是把单机游戏加上一个网络发送函数真正动手时才会发现两边的棋盘如何保持一致炮隔着棋子吃怎么判断对方断线是判负还是等重连基于 Python 实现这个项目等于一次性练习 socket、线程模型、状态同步和完整的象棋规则校验。它适合已经写过单机小游戏、想往网络方向迈一步的开发者也适合用来作为课程设计或简历里的独立项目。你不需要异步框架Python 标准库就能搭出一个可用版本先跑通一局再考虑要不要换更重的通信方案。这篇文章会按“架构选型 → 协议设计 → 服务器/客户端 → 规则细节 → 踩坑”的顺序把最小可运行方案和常见翻车点摊开讲。2. 拆解网络版中国象棋核心模块与最小可行架构2.1 棋盘与棋子建模坐标、编号与走法规则中国象棋棋盘是 9 列 10 行最直观的建模就是用二维数组board[row][col]空位为None棋子用两个字符表示第一个字符是阵营r/b第二个字符是类型k(帅/将)、a(士)、e(象/相)、h(马)、r(车)、c(炮)、p(兵/卒)。这里的e是为了避开关键字b避免和黑方阵营混淆。# chess_state.py def init_board(): board [[None for _ in range(9)] for _ in range(10)] # 黑方在上方row 0-4 for col, piece in enumerate([bc, bh, be, ba, bk, ba, be, bh, bc]): board[0][col] piece board[2][1] bb board[2][7] bb board[3][0] br board[3][8] br for col in (0, 2, 4, 6, 8): board[3][col] bp # 红方在下方row 5-9 for col, piece in enumerate([rc, rh, re, ra, rk, ra, re, rh, rc]): board[9][col] piece board[7][1] rb board[7][7] rb board[6][0] rr board[6][8] rr for col in (0, 2, 4, 6, 8): board[6][col] rp return board def side_of(piece): return piece[0] if piece else None这里把棋盘写成二维 list 而不是用对象封装是为了让网络广播时可以直接json.dumps(board)省掉序列化映射。代价是走法校验时要频繁做行列越界检查但 10x9 的规模完全没有性能压力。初始化时黑方row 0从左边到右边依次是车、马、象、士、将、士、象、马、车红方在row 9对称排列炮分别在row 2/7的 1 列和 7 列兵/卒在row 3/6的偶数列。这样摆放符合标准象棋开局。2.2 网络层选型socket vs 框架以及为什么 TCP 足够实现网络对战有很多路用socket自己控制用Flask-SocketIO做 WebSocket或者上asyncio事件循环。对于中国象棋这种低频、双向、需要强一致的状态同步我第一选择是标准库的socket threading原因有三个每步棋消息只有几十字节不需要高吞吐两个玩家之间是长期连接TCP 天然适合引入 WebSocket 或异步框架反而要把消息解析、心跳、线程模型都学一遍偏离了“象棋规则”这个核心难点。用框架的好处是连接管理省心但代价是调试时多一层黑匣子。我一般会先让普通 TCP 跑通之后再决定是否换框架。下表是选型对比方案并发模型心跳/断线适合场景标准库 socket threading每连接一线程可自己加小规模、教学、内部工具asyncio单线程事件循环需要自己实现超时想顺便练异步编程Flask-SocketIO事件驱动内置要做浏览器端、Web前端对于这个项目socket threading最直接。服务器为每个客户端开一个读线程主线程负责接受两个连接。两名玩家各自持有同一个棋盘状态的“服务器副本”所有修改都发生在服务器锁内客户端只负责发送走法和展示状态。2.3 最小架构与状态同步从本地逻辑到网络对局网络对战的核心问题是“服务器权威”还是“客户端各自为政”。常见做法是服务器持有棋盘状态接收走法后校验、更新再把完整局面广播给两个客户端客户端不自己改棋盘只负责渲染和发送用户输入。这样即使一台机器卡了另外一边也不会出现“我这边赢了、他那边没赢”的分裂局面。状态机可以简化为四个阶段状态含义触发事件WAITING_PLAYER等待第二个玩家加入第二个连接建立RED_TURN红方走棋收到合法红方走法BLACK_TURN黑方走棋收到合法黑方走法GAME_OVER结束将死、超时或一方退出服务器每次广播状态时都会附带当前轮到谁走。客户端收到state消息后如果自己不是当前回合就只刷新棋盘不允许本地输入。这个设计看起来保守但避免了两个人同时操作导致的状态冲突。在此基础上再考虑断线重连、日志回放这些进阶需求时服务器权威带来的统一状态会省很多事。3. 用 TCP 从零搭一个可下棋的服务器与客户端3.1 定义协议消息格式、回合控制与校验网络通信必须定义好协议否则两边各写各的最后根本对不上。我建议用“JSON 行协议”每条消息是一个 JSON 对象序列化后以换行符\n结尾。选 JSON 而不是 Python 原生pickle是因为 JSON 文本可读、便于抓包调试也方便以后把客户端改成其他语言。消息类型分成四类类型方向字段说明assigned服务器→客户端side告知玩家被分配为什么颜色state服务器→客户端boardturnwinner完整局面、当前回合、胜者move客户端→服务器from_rowfrom_colto_rowto_col玩家走法error服务器→客户端text非法操作提示客户端在本地输入坐标时行号从 0 到 9列号从 0 到 8。例如红方炮开局从(7,1)走到(7,4)就发送7 1 7 4。服务器收到move后会校验“是不是自己的回合”“起点有没有自己的棋子”“目标位置是否合法”这三步都通过才更新棋盘。3.2 服务器端接受连接、管理两方、转发棋步先写一个最小可运行的服务器。它监听0.0.0.0:9999接受两个连接分别分配给红黑双方之后进入消息循环。注意handle函数在独立线程里读消息主线程负责等待连接。# tcp_server.py import json, socket, threading import chess_state as cs class GameServer: def __init__(self, host, port): self.sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.sock.bind((host, port)) self.sock.listen(2) self.players {} self.board cs.init_board() self.turn r self.lock threading.Lock() self.over False def broadcast(self, msg): data (json.dumps(msg) \n).encode(utf-8) for conn in self.players.values(): try: conn.sendall(data) except Exception: pass def send_error(self, conn, text): conn.sendall((json.dumps({type: error, text: text}) \n).encode(utf-8)) def handle(self, conn, side): try: f conn.makefile(r, encodingutf-8) while True: line f.readline() if not line: break msg json.loads(line) if msg.get(type) move: with self.lock: self.process_move(conn, side, msg) else: self.send_error(conn, unknown type) except Exception: pass finally: with self.lock: if not self.over: self.over True self.broadcast({type: state, winner: b if side r else r, board: self.board, turn: self.turn}) def process_move(self, conn, side, msg): if self.over: self.send_error(conn, game over) return if side ! self.turn: self.send_error(conn, not your turn) return fr, fc msg[from_row], msg[from_col] tr, tc msg[to_row], msg[to_col] try: fr, fc, tr, tc int(fr), int(fc), int(tr), int(tc) except (TypeError, ValueError): self.send_error(conn, bad coordinate) return piece self.board[fr][fc] if piece is None or cs.side_of(piece) ! side: self.send_error(conn, invalid piece) return if not cs.is_valid_move(self.board, fr, fc, tr, tc): self.send_error(conn, illegal move) return self.board[tr][tc] piece self.board[fr][fc] None # 简单判定吃到帅/将即胜利后续可扩展将军死局 if k in piece: # 注意这里是我们自己的棋子实际是判断被吃的是帅/将 winner side self.over True else: winner None self.turn b if side r else r state_msg {type: state, board: self.board, turn: self.turn, winner: winner} self.broadcast(state_msg) def run(self): for side in (r, b): conn, _ self.sock.accept() self.players[side] conn conn.sendall((json.dumps({type: assigned, side: side}) \n).encode(utf-8)) threading.Thread(targetself.handle, args(conn, side), daemonTrue).start() self.broadcast({type: state, board: self.board, turn: self.turn, winner: None}) while not self.over: pass self.sock.close() if __name__ __main__: GameServer(0.0.0.0, 9999).run()这段代码有几个关键点。第一broadcast在锁内调用避免两个线程同时发消息导致交错第二handle里使用了makefile(r)会自动把字节流聚合成按行读取解决半包粘包问题第三process_move中判断if k in piece有问题因为piece是起点棋子如果红方走车去吃黑将piece是rck in piece为 False根本不会判胜。正确做法是判断目标位置self.board[tr][tc]在被赋值前是否是对方帅/将。这里故意留一个 bug后面避坑章会专门讲。大家先把这段当作一个能跑起来的骨架真实部署时要改成target_piece self.board[tr][tc] if target_piece and target_piece[1] k: winner side另外handle的finally里断线处理只是简单广播对手胜实际项目中还需要区分“掉线判负”和“断线重连”。3.3 客户端命令行交互与棋盘渲染客户端用命令行就够了重点是提供一个干净的棋盘渲染函数并让接收消息的线程和主线程的输入互不阻塞。这里先做一个能用的版本# tcp_client.py import json, socket, threading PIECE_NAMES { rk: 帅, ra: 仕, re: 相, rh: 马, rr: 车, rc: 炮, rp: 兵, bk: 将, ba: 士, be: 象, bh: 马, br: 车, bc: 炮, bp: 卒 } def print_board(board): print( 0 1 2 3 4 5 6 7 8) for r in range(10): line f{r} for c in range(9): p board[r][c] line (PIECE_NAMES.get(p, p) or ) print(line) def recv_loop(sock): f sock.makefile(r, encodingutf-8) while True: line f.readline() if not line: break msg json.loads(line) if msg[type] state: print(\n--- board ---) print_board(msg[board]) if msg.get(winner): print(winner:, msg[winner]) else: print(turn:, msg[turn]) elif msg[type] error: print(server error:, msg[text]) def main(): host input(host: ) or 127.0.0.1 sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((host, 9999)) f sock.makefile(r, encodingutf-8) assigned json.loads(f.readline()) print(your side:, assigned[side]) threading.Thread(targetrecv_loop, args(sock,), daemonTrue).start() while True: cmd input(move as fr fc tr tc ) if not cmd: break try: fr, fc, tr, tc map(int, cmd.split()) except ValueError: print(format: from_row from_col to_row to_col) continue sock.sendall((json.dumps({type: move, from_row: fr, from_col: fc, to_row: tr, to_col: tc}) \n).encode(utf-8)) if __name__ __main__: main()这里recv_loop单独跑一个线程避免readline()阻塞住主线程的输入print_board把英文字符映射成中文棋子方便人阅读。注意PIECE_NAMES里的p兵和卒用的是同一个键但红黑阵营区分了所以没冲突。玩游戏时红方输入的是棋盘上的绝对坐标不用关心朝向。客户端不校验合法性全部交给服务器判断这样服务器改了规则客户端不用同步改。4. 关键细节走法校验、将军判断与胜负判定4.1 走法校验规则表与边界检查走法校验是整个项目最容易写漏的部分。中国象棋每一类棋子的走法都不太一样但可以拆成几个辅助函数。第一是直线无障碍检查clear_path车、炮都用它判断中间有没有被挡。第二是九宫范围判断帅/将和士的活动范围不同。# chess_state.py 追加 def clear_path(board, r1, c1, r2, c2): if r1 r2: step 1 if c2 c1 else -1 for c in range(c1 step, c2, step): if board[r1][c]: return False return True if c1 c2: step 1 if r2 r1 else -1 for r in range(r1 step, r2, step): if board[r][c1]: return False return True return False def in_palace(side, r, c): if side r: return 7 r 9 and 3 c 5 return 0 r 2 and 3 c 5 def is_valid_move(board, r1, c1, r2, c2): piece board[r1][c1] if piece is None: return False side piece[0] target board[r2][c2] if target and target[0] side: return False if r1 r2 and c1 c2: return False if not (0 r2 10 and 0 c2 9): return False ptype piece[1] dr, dc r2 - r1, c2 - c1 if ptype k: return in_palace(side, r2, c2) and abs(dr) abs(dc) 1 if ptype a: return in_palace(side, r2, c2) and abs(dr) 1 and abs(dc) 1 if ptype e: # 象走田不能过河 if side r and r2 5: return False if side b and r2 4: return False if abs(dr) ! 2 or abs(dc) ! 2: return False mid_r, mid_c (r1 r2) // 2, (c1 c2) // 2 return board[mid_r][mid_c] is None if ptype h: # 马走日注意蹩马腿 if abs(dr) 1 and abs(dc) 2: leg_r, leg_c r1 if abs(dc) 2 else r1 dr // 1, c1 if abs(dr) 2 else c1 dc // 1 # 简化横走时马腿在左右竖走时马腿在上下 if abs(dc) 2: leg_r, leg_c r1, c1 dc // 2 else: leg_r, leg_c r1 dr // 2, c1 return board[leg_r][leg_c] is None return False if ptype r: return (r1 r2 or c1 c2) and clear_path(board, r1, c1, r2, c2) if ptype c: if r1 r2 or c1 c2: if target is None: return clear_path(board, r1, c1, r2, c2) # 吃子需要中间恰好隔一个子 count 0 if r1 r2: step 1 if c2 c1 else -1 for c in range(c1 step, c2, step): if board[r1][c]: count 1 else: step 1 if r2 r1 else -1 for r in range(r1 step, r2, step): if board[r][c1]: count 1 return count 1 return False if ptype p: # 红兵向前是-1黑卒向前是1 forward -1 if side r else 1 # 过河判定红兵row4黑卒row5 crossed (r1 4) if side r else (r1 5) if dc 0 and dr forward: return True if crossed and dr 0 and abs(dc) 1: return True return False return False象棋规则里有个隐藏陷阱帅/将不能“照面”。上面这段没有处理。照面的意思是红帅和黑将如果在同一条竖线上且中间没有任何棋子那么轮到谁都算被将军。规则要求这个局面不合法所以校验走法时需要额外检查。常见做法是在is_valid_move的第一步先判断如果走完这步后自己的将帅还在且和对方将帅照面就拒绝。这个检查要放在所有分支之后我会在 4.2 里给出方法。4.2 将军与将死从“被吃”到“无法解”的判定很多初学者只做到“吃到帅/将就赢”但这漏掉了“将军”和“将死”的规则。更合理的方式是每走一步后检查对方是否被将军如果被将军再检查对方是否还有合法走法可以解除将军如果没有才算将死。可以由两个函数配合实现。def find_king(board, side): k side k for r in range(10): for c in range(9): if board[r][c] k: return r, c return None def in_check(board, side): king find_king(board, side) if not king: return True kr, kc king # 检查敌方所有棋子是否能攻击到将/帅 for r in range(10): for c in range(9): p board[r][c] if p and p[0] ! side and is_valid_move(board, r, c, kr, kc): return True # 将帅照面 other find_king(board, b if side r else r) if other and kc other[1]: empty True lo, hi sorted([kr, other[0]]) for rr in range(lo 1, hi): if board[rr][kc]: empty False break if empty: return True return False def is_mate(board, side): if not in_check(board, side): return False for r1 in range(10): for c1 in range(9): p board[r1][c1] if p and p[0] side: for r2 in range(10): for c2 in range(9): if is_valid_move(board, r1, c1, r2, c2): tmp [row[:] for row in board] tmp[r2][c2] tmp[r1][c1] tmp[r1][c1] None # 走完后自己不能被将军且要能解除当前被将军状态 if not in_check(tmp, side): return False return True注意in_check循环里调用is_valid_move来判断“敌方能吃掉我的将/帅”但is_valid_move并不会考虑“走这步后敌人自己会不会被将军”这对攻击检测来说是对的。可是这里有一个循环依赖in_check调用is_valid_move而is_valid_move如果需要检查照面又可能调用in_check导致相互递归。解决方法是把“走法是否合法”和“走法是否导致我方仍在将军中”拆成两层。上面代码的is_valid_move还没有包含照面检查所以in_check单独处理了照面在is_mate的试走里in_check(tmp, side)已经检查了走完后自己是否处于将军状态因此最终判断是成立的。这个层次要分清楚。4.3 和棋与超时网络对局的特殊处理单机象棋可以不管超时网络版必须考虑。最基础的是自然限着连续 60 个回合双方各走一步算一回合都没有吃子判和棋。实现方法是在服务器维护一个no_capture_round计数器每次走法目标格子为空则加 1不为空则清零计数器到达 60 就在 state 里广播winner: draw。另外就是每步超时比如限制 120 秒。服务器需要为当前回合玩家启动一个倒计时线程def start_timer(self, side, timeout120): def _expire(): nonlocal expired time.sleep(timeout) with self.lock: if not self.over and self.turn side: self.over True self.broadcast({type: state, board: self.board, turn: self.turn, winner: b if side r else r}) expired False t threading.Thread(target_expire, daemonTrue) t.start() return t这个方案有一个坑上一手走完后要取消旧计时器否则两个计时器会同时存在。更干净的做法是在process_move里设置一个新的self.time_limit时间戳单独用一个监控线程检查当前时间是否超过time_limit。超时和掉线混在一起时容易把正常走子误判为超时所以状态机里最好有一个LAST_ACTIVITY字段每次收到合法move都更新。就个人经验而言这个项目里超时不是优先级最高的需求先把掉线和规则做稳定再加超时更顺手。5. 网络版象棋避坑排查连接、协议与并发问题记录5.1 玩家掉线导致整个对局卡死现象对局进行到一半某一方直接关闭终端服务器没有任何反应另一边的客户端也一直卡在“等待对方走棋”的状态。问题出在服务器主线程while not self.over: pass上。掉线后handle线程读到了 EOF它的finally里虽然设置了self.over True但如果掉线的玩家不是当前回合的玩家process_move永远不会被调用主循环就不会退出其实handle线程的finally确实会执行self.over变为 True主线程会退出。但为什么卡住原来是self.lock的问题如果掉线发生在process_move持锁期间handle线程在finally里尝试拿锁会被阻塞而主线程又在等over互相死锁。解决方法是把finally里的广播和状态设置放到锁外或者使用threading.Event来通知主循环退出。最简单的方式是不依赖over死循环而是直接让handle线程在断线时调用一个shutdown方法def shutdown(self): with self.lock: if self.over: return self.over True self.broadcast({type: state, board: self.board, turn: self.turn, winner: black}) # 对方获胜 self.sock.close()然后在handle的finally里直接调用self.shutdown()。这样锁只保护状态检查不保护网络发送不会死锁。5.2 两台机器显示局面不一致的复盘现象双方终端上的棋盘偶尔出现“红方已经吃过棋了但黑方界面还显示原棋子”多走几步后两边完全对不上。最常见原因是服务器广播的不是“完整局面”而是“增量步”。某开发者为了省流量只广播{from, to, piece}客户端各自更新本地棋盘结果一端发生一次丢消息或重复消息后续全部错位。我倾向于服务器每次广播完整board数组只有 90 个元素JSON 文本也就几 KB完全够用。另一个原因是并发打印。客户端recv_loop线程和主线程输入循环都可能往终端写input的提示符会被print_board的换行打断。解决方法是给客户端打印加一个锁print_lock threading.Lock() def safe_print(text): with print_lock: print(text)把recv_loop里的所有print都改成safe_print。这不影响协议但极大地改善体验。5.3 同步阻塞导致界面假死现象客户端输入一步棋后按回车没有任何反馈再按几次才有棋盘刷出来或者干脆卡死。原因是接收线程readline()在等待服务器消息时阻塞但你的主线程也在不断input()等待输入两个线程其实各自独立不会互相阻塞。但如果把接收线程也放在主线程里用input()等用户输入和socket.recv()混在一起就会卡死。真正的常见错误是服务器发送消息时客户端接收线程在等数据而主线程input()也在等用户输入两个都阻塞没问题可一旦服务器在等待当前回合玩家走棋而客户端主线程又input()等待用户输入好像正常。但为什么会出现“按几次回车才更新”因为客户端发送走法后没有立即刷新本地棋盘必须等服务器state回来。如果服务器处理慢或客户端recv_loop被print_lock卡住就会产生假死感。解决方案客户端发送move后不要阻塞等待而是依赖recv_loop的state消息驱动界面刷新。如果担心丢消息可以加一个发送队列和确认机制。真正需要修复的是服务器端的阻塞行为。服务器handle线程里如果某个连接不发送数据readline()会一直阻塞这没问题但broadcast使用conn.sendall是阻塞的如果某个客户端网卡慢了sendall会拖住整个线程。我一般会给sendall加超时conn.settimeout(5) try: conn.sendall(data) except socket.timeout: self.shutdown()这样即使一方网络异常也不会无限阻塞广播。5.4 半包粘包问题为什么我的棋步偶尔解析错现象客户端发送走法后服务器偶尔报json.JSONDecodeError有时是两个消息粘在一起被解析失败。TCP 是字节流消息边界是我们自己定的。如果发送方只发了 JSON 但没有加\n接收方用readline()时可能读到半个 JSON如果发送方连续快速发送两个消息接收方可能一次读到两行但readline()会按行分割所以粘包不是问题。真正的坑出现在服务器端process_move里用json.loads(line)如果客户端发来的行末带有\r\njson.loads也可能报错。解决方法是发送时统一加\n接收时先strip()。line f.readline() if not line: break try: msg json.loads(line.strip()) except json.JSONDecodeError: self.send_error(conn, bad json) continue另一个问题是makefile(r)和sendall的缓冲。客户端发送消息后如果不flush数据可能攒在缓冲里不出去。用sock.sendall不需要 flush因为它直接操作 socket。但如果用了sock.makefile(w)就必须flush()。标准做法是统一用sendall不经过makefile。上面第 3 章的代码已经这样做了所以你这边的半包粘包多数是\r\n或编码问题。6. 进阶断线重连、AI 推演与日志回放服务器权威状态的好处是断线重连变得可行。你可以在玩家assigned时生成一个 8 位随机match_id存入连接对象断线后服务器保留棋盘和turn状态等待 60 秒。当新连接发来带match_id的消息时服务器从字典里找回这个对局把新连接绑定到原side上重新广播一次state。注意重连的客户端需要知道自己的阵营所以协议里要加一个join请求。AI 推演不建议一上来就做完整 Minimax先做一个随机合法走法再到深度 2 的评估函数。评估函数可以只算双方棋子价值和位置分车 900、马 400、炮 450、象/士 200、兵 100帅 10000。每走一步对棋盘打分选择己方分数最高的后续局面。这样代码量不大但足够让玩家练习。日志回放是最容易被忽视的调试手段。给服务器加一个logging模块每收到一个move就记录一行 JSON包含时间戳、发送方、坐标和合法性结果。之后用 Python 脚本逐行重放可以把整个对局打印出来。我在做模拟项目X 时就是用回放日志发现了一个只在第 50 步后出现的规则 bug。那是一个“马”蹩腿边界判断写错的问题单步测试看不出来回放下完整棋局才定位。从此我养成了写项目先插日志的习惯。断线重连的坑在于恢复时不能简单把棋盘发出去还要恢复“当前轮到谁”。如果黑方掉线恢复后把turn设错红方会连续走两步。我一般会把整个状态机对象GameServer序列化成 JSON 存起来重连时反序列化。还有一个细节象棋的将帅照面判定在回放时也必须走同一套in_check逻辑否则回放数据会误导你。最后建议在做完基本网络对战后先写一个随机脚本来回走一千步用日志验证没有异常崩溃再让真人测试。这样的项目才算真正落地而不是只能演示第一回合。希望这篇笔记里的命令、参数和踩坑记录能帮到你少走几段我当年走过的弯路。本文还有配套的精品资源点击获取
