我是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 噢～我有空了会分享做游戏的相干技术。