[教你做小游戏] 用86行代码写一个联机五子棋WebSocket后端

语言: CN / TW / HK

持续创作,加速成长!这是我参与「掘金日新计划 · 6 月更文挑战」的第7天,点击查看活动详情

我是公众号「线下聚会游戏」的作者,开发了一些联机桌游网页(UNO、斗地主、五子棋等),总结了一些小游戏开发经验,汇总在专栏《教你做小游戏》,分享给大家,欢迎关注。

背景

上篇文章《177行代码手撸一个体验超好的五子棋》,我们一起用177行代码实现了一个本地对战的五子棋游戏。

现在,如果我们要做一个联机五子棋,怎么办呢?

需求分析

首先,我们需要一个后端服务。2个不同的玩家,一起连接这个后端服务,把要下的棋告诉后端,后端再转发给另一个玩家即可。当然,如果有观战的,也要把当前期局转发给观战者。

此外,为了让2个玩家联机,还需要有「房间号」的概念,只有同一个房间的人才能联机对战。不同房间的人互不影响,允许同时有多个房间的人同时玩游戏。

流程

整个通信流程是这样的:

  1. 玩家A请求进入房间1。玩家A会执黑棋。
  2. 玩家B请求进入房间1。玩家B会执白棋。此时人已满,其他人进入将观战。
  3. 玩家C请求进入房间1。玩家C是观战者。
  4. 玩家A请求下棋,告诉坐标给服务器。
  5. 服务器通知玩家B、玩家C,告诉大家A下棋的坐标。
  6. 玩家B请求下棋,告诉坐标给服务器。
  7. 服务器通知玩家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要求我们以这样的格式定义一个服务:

```python

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}) ```

运行方法:

shell pip install daphne daphne -b 0.0.0.0 -p 8001 server:application

业务开发

我们需要定义一个房间集合,称之为house

python house = {}

编写玩家初次连接(进入房间)的逻辑:

python 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)

玩家进入房间后,我们需要给他通知一下这个房间的基本信息,例如是否已经开始了?当前场上的期局是怎样的?

python 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。

写在最后

如果你像我一样,追求极致用户体验,非常推荐你关注专栏:《极致用户体验》,我会分享更多文章,绝对是全网独一无二的原创精华内容 😎 专栏里面的内容,看完一定会有收获!求赞,求关注,谢谢啦~

另外,如果你想学做小游戏,欢迎关注我的专栏:《教你做小游戏》,我会写些文章,介绍制作游戏过程中的难点及解决方案。

「其他文章」