写点什么

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

作者:HullQin
  • 2022 年 8 月 23 日
    广东
  • 本文字数:3921 字

    阅读完需:约 13 分钟

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

我是 HullQin,公众号线下聚会游戏的作者(欢迎关注公众号,发送加微信,交个朋友),转发本文前需获得作者 HullQin 授权。我独立开发了《联机桌游合集》,是个网页,可以很方便的跟朋友联机玩斗地主、五子棋等游戏,不收费没广告。还独立开发了《合成大西瓜重制版》。还开发了《Dice Crush》参加 Game Jam 2022。喜欢可以关注我 HullQin 噢~我有空了会分享做游戏的相关技术。

背景

上篇文章《用 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要求我们以这样的格式定义一个服务:


# server.pyasync 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 daphnedaphne -b 0.0.0.0 -p 8001 server:application
复制代码

业务开发

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


house = {}
复制代码


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


import jsonasync 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 噢~我有空了会分享做游戏的相关技术。

发布于: 2022 年 08 月 23 日阅读数: 4
用户头像

HullQin

关注

公众号【线下聚会游戏】 2020.10.07 加入

game.hullqin.cn 我做了一些联机桌游网页:支持2-10人联机的UNO、2-4人联机的斗地主、2人联机的五子棋。无需下载,点开即玩!叫上朋友,即刻开局!不看广告,不做任务,享受「纯粹」的游戏!

评论

发布
暂无评论
[教你做小游戏] 用86行代码写一个联机五子棋WebSocket后端_CSS_HullQin_InfoQ写作社区