我是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':returnawait send({'type': 'websocket.accept'})# 建立連接后，要求前端發送一個EnterRoom事件，以json格式提供用戶id和房間號roomevent = 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})returnroom_id = data['room']user_id = data['id']# 看看房間號是否在house內，不在則創建一個roomif 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 = Trueif 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_idelif 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:continueawait _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: # 不需要給自己通知，所以跳過自己continueawait _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 噢～我有空了會分享做游戲的相關技術。

總結

以上是生活随笔為你收集整理的[教你做小游戏] 用86行代码写一个联机五子棋WebSocket后端的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。