我是HullQin，公众号线下聚会游戏的作者（欢迎关注公众号，发送加微信，交个朋友），转发本文前需获得作者HullQin授权。我独立开发了《联机桌游合集》，是个网页，可以很方便的跟朋友联机玩斗地主、五子棋等游戏，不收费没广告。还开发了《Dice Crush》参加Game Jam 2022。喜欢可以关注我 HullQin 噢～我有空了会分享做游戏的相关技术。
背景
上篇文章《用177行代码写个体验超好的五子棋》，我们一起用177行代码实现了一个本地对战的五子棋游戏。
现在，如果我们要做一个联机五子棋，怎么办呢？
需求分析
首先，我们需要一个后端服务。2个不同的玩家，一起连接这个后端服务，把要下的棋告诉后端，后端再转发给另一个玩家即可。当然，如果有观战的，也要把当前期局转发给观战者。
此外，为了让2个玩家联机，还需要有「房间号」的概念，只有同一个房间的人才能联机对战。不同房间的人互不影响，允许同时有多个房间的人同时玩游戏。
流程
整个通信流程是这样的：
- 玩家A请求进入房间1。玩家A会执黑棋。
- 玩家B请求进入房间1。玩家B会执白棋。此时人已满，其他人进入将观战。
- 玩家C请求进入房间1。玩家C是观战者。
- 玩家A请求下棋，告诉坐标给服务器。
- 服务器通知玩家B、玩家C，告诉大家A下棋的坐标。
- 玩家B请求下棋，告诉坐标给服务器。
- 服务器通知玩家A、玩家C，告诉大家B下棋的坐标。
之后循环4-7步骤。
为了简化后端逻辑，把逻辑判断都放在前端。例如在前端判断是否游戏结束（五联珠），如果游戏结束，前端不允许再发任何请求。
技术选型
协议与方案
因为涉及到服务器主动给用户发送数据，所以有几种可选方案：
- Http轮询：若在等待对方下棋，则前端每隔1s就发送一条请求，看看对方是否下棋。
- Http长轮询：若在等待对方下棋，则前端每隔1s就发送一条请求，看看对方是否下棋。但是后台不会立即返回结果，要等到接口超过某个时间才返回结果。
- WebSocket：建立好浏览器、服务器的连接，可随时主动向浏览器推送数据。
这里我们选择WebSocket，因为这种场景下Http协议确实有很大的资源浪费。而WebSocket虽然实现起来有点难度，但是节约了资源。
具体实现方案
只要某个编程语言/框架可以支持WebSocket就可以。
因为我以前经常用Django，用过Channels，对它的底层依赖daphne有所了解，所以我直接选择了daphne。它是ASGI标准的一种实现。
daphne是一个非常轻量的选择，不像Django+Channels这套框架提供了很重的解决方案。daphne只提供了基础的ASGI实现，没有其它冗余的功能。就好比：我开发五子棋前端时，使用了SVG + Dom API，没有用React框架一样。
开发
基础知识
daphne要求我们以这样的格式定义一个服务：
# server.py
async def application(scope, receive, send):
# 处理websocket协议
if scope['type'] == 'websocket':
# 先接收第一个包，必须是建立连接的包(connect)，否则拒绝服务
event = await receive()
if event['type'] != 'websocket.connect':
return
# 校验通过，发送accept，表明建立ws连接成功
await send({'type': 'websocket.accept'})
# 此后双方可以互相随时发消息。开启个无限循环
while True:
# 接收一个包
event = await receive()
# 如果是断开连接的请求，就结束循环
if event['type'] == 'websocket.disconnect':
break
# 这种方式可以读取包的文本内容
data = event['text']
# 这种方式可以发送一个包给浏览器，这里是把浏览器发来的包原封不动传回去
await send({'type': 'websocket.send', 'text': data})运行方法：
pip install daphne
daphne -b 0.0.0.0 -p 8001 server:application业务开发
我们需要定义一个房间集合，称之为house
house = {}编写玩家初次连接（进入房间）的逻辑：
import json
async def application(scope, receive, send):
if scope['type'] == 'websocket':
event = await receive()
if event['type'] != 'websocket.connect':
return
await send({'type': 'websocket.accept'})
# 建立连接后，要求前端发送一个EnterRoom事件，以json格式提供用户id和房间号room
event = await receive()
data = json.loads(event['text'])
if data['type'] != 'EnterRoom' or not data['id'] or not data['room']:
# 若前端发送的第一个事件不是这个，就报错，断开连接
await send({'type': 'websocket.close', 'code': 403})
return
room_id = data['room']
user_id = data['id']
# 看看房间号是否在house内，不在则创建一个room
if room_id not in house:
house[room_id] = {
'black': None,
'white': None,
'pieces': [],
'sends': [],
'users': [],
}
room = house[room_id]
old = False # 看玩家是不是老玩家（断线重连进来的）
if room['black'] == user_id or room['white'] == user_id:
old = True
if user_id in room['users']:
old_send = room['sends'][room['users'].index(user_id)]
room['sends'].remove(old_send)
room['users'].remove(user_id)
await old_send({'type': 'websocket.close', 'code': 4000})
else: # 说明玩家是第一次进，给他拿黑棋或白棋
if room['black'] is None:
room['black'] = user_id
elif room['white'] is None:
room['white'] = user_id
# 如果玩家没拿到黑棋也没拿到白旗，就是观战者
visiting = room['black'] != user_id and room['white'] != user_id
# 把玩家的send函数存到room里，方便其他玩家下棋时调用，从而广播下棋事件
room['sends'].append(send)
# 把玩家ID存进去
room['users'].append(user_id)玩家进入房间后，我们需要给他通知一下这个房间的基本信息，例如是否已经开始了？当前场上的期局是怎样的？
await send({'type': 'websocket.send', 'text': json.dumps({
'type': 'InitializeRoomState',
'pieces': room['pieces'], # 场上棋子情况
'visiting': visiting, # 你是否是观战者
'black': room['black'] == user_id if not visiting else bool(len(room['pieces']) % 2), # 如果你在下棋：黑棋是你吗？如果你是观战者：黑棋是谁？
'ready': bool(room['black'] and room['white']), # 房间是否准备好开局了？只要有2个人同时在，就可以开了
})})
# 因为有人进入了房间，所以需要广播一下这个消息。
if not old and (room['black'] == user_id or room['white'] == user_id):
for _send in room['sends']:
if _send == send:
continue
await _send({'type': 'websocket.send', 'text': json.dumps({
'type': 'AddPlayer',
'ready': bool(room['black'] and room['white']),
})})
while True:
event = await receive()
# 有人断线了，处理一下。若房间空了，还要删掉房间，以防内存占用无限增大
if event['type'] == 'websocket.disconnect':
if send in room['sends']:
room['sends'].remove(send)
room['users'].remove(user_id)
if len(room['pieces']) == 0 and len(room['sends']) == 0:
del house[room_id]
break
# 有人发送了事件，接收一下
data = json.loads(event['text'])
# 如果是下棋事件，就改一下room的pieces数据，并广播给大家
if data['type'] == 'DropPiece':
room['pieces'].append((data['x'], data['y']))
for _send in room['sends']:
if _send == send: # 不需要给自己通知，所以跳过自己
continue
await _send({'type': 'websocket.send', 'text': json.dumps({
'type': 'DropPiece',
'x': data['x'],
'y': data['y'],
})})当然，写好这些后，还需要测试，最好直接写好前端一起联调。我们下篇文章把前端的WebSocket逻辑补充一下。
完整源码
包含了前后端源码（总共不到400行）: https://github.com/HullQin/gobang
是一个非常值得学习的关于WebSocket的demo。
写在最后
我是HullQin，公众号线下聚会游戏的作者（欢迎关注公众号，发送加微信，交个朋友），转发本文前需获得作者HullQin授权。我独立开发了《联机桌游合集》，是个网页，可以很方便的跟朋友联机玩斗地主、五子棋等游戏，不收费没广告。还开发了《Dice Crush》参加Game Jam 2022。喜欢可以关注我 HullQin 噢～我有空了会分享做游戏的相关技术。
相关文章
