持續創作，加速成長！這是我參與「掘金日新計劃 · 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行）: http://github.com/HullQin/gobang

是一個非常值得學習的關於WebSocket的demo。

寫在最後

如果你像我一樣，追求極致使用者體驗，非常推薦你關注專欄：《極致使用者體驗》，我會分享更多文章，絕對是全網獨一無二的原創精華內容 😎 專欄裡面的內容，看完一定會有收穫！求贊，求關注，謝謝啦～

另外，如果你想學做小遊戲，歡迎關注我的專欄：《教你做小遊戲》，我會寫些文章，介紹製作遊戲過程中的難點及解決方案。