[教你做小遊戲] 用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。

寫在最後

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

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

「其他文章」