作者：李曉飛

來源：Python

Facebook 改名為 meta，一下子點燃了 元宇宙 這個概念。

今天我就用 Python 實現一個簡單的迷你元宇宙。

代碼簡潔易懂，不僅可以學習 Python 知識，還能用實踐理解元宇宙的概念。

還等什么，現在就開始吧！

迷你元宇宙

什么是元宇宙？

不同的人有不同的理解和認識，最能達成共識的是：

元宇宙是個接入點，每個人都可以成為其中的一個元素，彼此互動。

那么我們的元宇宙有哪些功能呢？

首先必須有可以接入的功能。

然后彼此之間可以交流信息。比如 a 發消息給 b，b 可以發消息給 a，同時可以將消息廣播出去，也就是成員之間，可以私信 和 群聊。

另外，在元宇宙的成員可以收到元宇宙的動態，比如新人加入，或者有人離開等，如果玩膩了，可以離開元宇宙。

最終的效果像這樣：

效果

設計

如何構建接入點

直接思考可能比較困難，換個角度想，接入點其實就是 —— 服務器。

只要是上網，每時每刻，我們都是同服務器打交的。

那就選擇最簡單的 TCP 服務器，TCP 服務器的核心是維護套接字（socket）的狀態，向其中發送或者獲取信息。

python 的 socket 庫，提供了很多有關便捷方法，可以幫助我們構建。

核心代碼如下：

import socket socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) socket.bind((ip, port)) socket.listen() data = socket.recv(1024) ... 

創建一個 socket，讓其監聽機器所擁有的一個 ip 和 端口，然后從 socket 中讀取發送過來的數據。

如何構建客戶端

客戶端是為了方便用戶鏈接到元宇宙的工具，這里，就是能鏈接到服務器的工具，服務器是 TCP 服務器，客戶端自然需要用可以鏈接 TCP 服務器的方式。

python 也已為我們備好，幾行代碼就可以搞定，核心代碼如下：

import socket client = socket.socket(socket.AF_INET,socket.SOCK_STREAM) client.connect((ip, port)) data = client.recv(1024) ... 

代碼與服務器很像，不過去鏈接一個服務器的 ip 和 端口。

如何構建業務邏輯

首先需要讓服務器將接入的用戶管理起來。

然后當接收到用戶消息時做出判斷，是轉發給其他用戶，廣播還是做出回應。

這樣就需要構造一種消息格式，用來表示用戶消息的類型或者目的。

我們就用 @username 的格式來區分，消息發給特殊用戶還是群發。

另外，為了完成注冊功能，需要再定義一種命令格式，用于設置 username，我們可以用 name:username 的格式作為設置用戶名的命令。

構建

有了初步設計，就可以進一步構建我們的代碼了。

服務端

服務器需要同時響應多個鏈接，其中包括新鏈接創建，消息 和 鏈接斷開 等。

為了不讓服務器阻塞，我們采用非阻塞的鏈接，當鏈接接入時，將鏈接存儲起來，然后用 select 工具，等待有了消息的鏈接。

這個功能，已經有人實現好了 simpletcp[1]，只要稍作改動就好。

將其中的收到消息，建立鏈接，關閉鏈接做成回調方法，以便再外部編寫業務邏輯。

核心業務

這里說明一下核心代碼：

# 創建一個服務器鏈接 self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self._socket.setblocking(0) self._socket.bind((self.ip, self.port)) self._socket.listen(self._max_connections) # 存放已建立的鏈接 readers = [self._socket] # 存放客戶端 ip和端口 IPs = dict() # 退出標記 用于關閉服務器 self._stop = False # 服務器主循環 while readers and not self._stop: # 利用 select 從 建立的鏈接中選取一些有新消息的 read, _, err = select.select(readers, [], readers) for sock in read: if sock is self._socket: # 建立了新鏈接 # 獲取新鏈接的 socket 以及 ip和端口 client_socket, client_ip = self._socket.accept() # 將鏈接設置為非阻塞的 client_socket.setblocking(0) # 添加到監聽隊列 readers.append(client_socket) # 存儲ip信息 IPs[client_socket] = client_ip # 調用建立鏈接回調函數 self.onCreateConn(self, client_socket, client_ip) else: # 收到了新消息 try: # 獲取消息 data = sock.recv(self.recv_bytes)
            except socket.error as e: if e.errno == errno.ECONNRESET: # 表明鏈接已退出 data = None else:
                    raise e if data: # 調用收到消息回調函數 self.onReceiveMsg(self, sock, IPs[sock], data) else: # 鏈接退出時，移除監聽隊列 readers.remove(sock)
                sock.close() # 調用鏈接關閉回調函數 self.onCloseConn(self, sock, IPs[sock]) # 處理存在錯誤的鏈接 for sock in err: # 移除監聽隊列 readers.remove(sock)
        sock.close() # 調用鏈接關閉回調函數 self.onCloseConn(self, sock, IPs[sock])

• 首先利用 socket 建立一個服務器鏈接，這個和最初的 socket 核心代碼一樣
• 不同的是設置鏈接為非阻塞的，這樣就可以通過 select 同時監控多個鏈接，也不至于阻塞服務器了。關于 select 可以看這里[2]
• 在主循環中，篩選出有了消息的鏈接，判斷是建立鏈接還是消息發送，調用不同的回調函數
• 最后處理一下異常

事件處理

現在通過回調函數，就可以編寫業務了，之間看代碼。

這段是建立鏈接時的處理：

def onCreateConn(server, sock, ip): cid = f'{ip[0]}_{ip[1]}' clients[cid] = {'cid': cid, 'sock': sock, 'name': None}
    sock.send("你已經接入元宇宙，告訴我你的代號,輸入格式為 name:lily.".encode('utf-8'))

• 首先計算出客戶端 id，即 cid，通過 ip 和 端口 組成
• clients 是個詞典，用 cid 為 key，存儲了 cid、鏈接、和名稱
• 一旦建立起來鏈接，向鏈接發送一段問候語，并要求其設置自己的名稱

然后是接收消息的回調函數，這個相對復雜一些，主要是處理的情況更多：

def onReceiveMsg(server, sock, ip, data): cid = f'{ip[0]}_{ip[1]}' data = data.decode('utf-8')
    print(f"收到數據: {data}")
    _from = clients[cid] if data.startswith('name:'): # 設置名稱 name = data[5:].strip() if not name:
            sock.send(f"不能設置空名稱，否則其他人找不見你".encode('utf-8')) elif not checkname(name, cid):
            sock.send(f"這個名字{name}已經被使用，請換一個試試".encode('utf-8')) else: if not _from['name']:
                sock.send(f"{name} 很高興見到你，現在可以暢游元宇宙了".encode('utf-8'))
                msg = f"新成員{name} 加入了元宇宙，和TA聊聊吧".encode('utf-8')
                sendMsg(msg, _from) else:
                sock.send(f"更換名稱完成".encode('utf-8'))
                msg = f"{_from['name']} 更換名稱為 {name}，和TA聊聊吧".encode('utf-8')
                sendMsg(msg, _from)
            _from['name'] = name elif '@' in data: # 私信 targets = re.findall(r'@(.+?) ', data)
        print(targets)
        msg = f"{_from['name']}: {data}".encode('utf-8')
        sendMsg(msg, _from, targets) else: # 群信 msg = f"{_from['name']}：{data}".encode('utf-8')
        sendMsg(msg, _from)

• 代碼分為兩大部分，if 前面是處理收到的消息，將 bytes 轉化為 字符串；if 開始處理具體的消息
• 如果收到 name: 開頭的消息，表示需要設置用戶名，其中包括判重，以及給其他成員發送消息
• 如果收到的消息里有 @，表示在發私信，先提取出需要發出的用戶們，然后將消息發送給對應的用戶
• 如果沒有特殊標記，就表示群發
• 其中 sendMsg 用于發送消息，接收三個參數，第一個是消息，第二是發送者，第三個是接收者名稱數組

當鏈接關閉時，需要處理一下關閉的回調函數：

def onCloseConn(server, sock, ip): cid = f'{ip[0]}_{ip[1]}' name = clients[cid]['name'] if name:
        msg = f"{name} 從元宇宙中消失了".encode('utf-8')
        sendMsg(msg, clients[cid]) del clients[cid]

• 當收到鏈接斷開的消息時，合成消息，發送給其他用戶
• 然后從客戶端緩存中刪除

客戶端

客戶端需要解決兩個問題，第一個是處理接收到的消息，第二個是允許用戶的輸入。

我們將接收消息作為一個線程，將用戶輸入作為主循環。

接收消息

先看接收消息的代碼：

def receive(client): while True: try:
            s_info = client.recv(1024) # 接受服務端的消息并解碼 if not s_info:
                print(f"{bcolors.WARNING}服務器鏈接斷開{bcolors.ENDC}") break print(f"{bcolors.OKCYAN}新的消息：{bcolors.ENDC}n", bcolors.OKGREEN + s_info.decode('utf-8')+ bcolors.ENDC) except Exception:
            print(f"{bcolors.WARNING}服務器鏈接斷開{bcolors.ENDC}") break if close: break 

• 這是線程中用的代碼，接收一個客戶端鏈接作為參數
• 在循環中不斷地從鏈接中獲取信息，如果沒有消息時 recv 方法會阻塞，直到有新的消息過來
• 收到消息后，將消息寫出到控制臺上
• bcolors 提供了一些顏色標記，將消息顯示為不同的顏色
• close 是一個全局標記，如果客戶端需要退出時，會設置為 True，可以讓線程結束

輸入處理

下面再看一下輸入控制程序：

while True: pass value = input("")
    value = value.strip() if value == ':start': if thread:
            print(f"{bcolors.OKBLUE}您已經在元宇宙中了{bcolors.ENDC}") else:
            client = createClient(ip, 5000)
            thread = Thread(target=receive, args=(client,))
            thread.start()
            print(f"{bcolors.OKBLUE}您進入元宇宙了{bcolors.ENDC}") elif value == ':quit' or value == ':stop': if thread:
            client.close()
            close = True print(f"{bcolors.OKBLUE}正在退出中…{bcolors.ENDC}")
            thread.join()
            print(f"{bcolors.OKBLUE}元宇宙已退出{bcolors.ENDC}")
            thread = None if value == ':quit':
            print(f"{bcolors.OKBLUE}退出程序{bcolors.ENDC}") break pass elif value == ':help':
        help() else: if client: # 聊天模式 client.send(value.encode('utf-8')) else:
            print(f'{bcolors.WARNING}還沒接入元宇宙，請先輸入 :start 接入{bcolors.ENDC}')
    client.close()

• 主要是對不同的命令做出的相應，比如 :start 表示需要建立鏈接，:quit 表示退出等
• 命令前加 : 是為了和一般的消息做區分，如果不帶 : 就認為是在發送消息

啟動

完成了整體編碼之后，就可以啟動了，最終的代碼由三部分組成。

第一部分是服務器端核心代碼，存放在 simpletcp.py 中。

第二部分是服務器端業務代碼，存放在 metaServer.py 中。

第三部分是客戶端代碼，存放在 metaClient.py 中。

另外需要一些輔助的處理，比如發送消息的 sendMsg 方法，顏色處理方法等，具體可以下載本文源碼了解。

進入代碼目錄，啟動命令行，執行 python metaServer.py，輸入指令 start:

server

然后再打開一個命令行，執行 python metaClient.py，輸入指令 :start，就可以接入到元宇宙：

client

設置自己的名字：

如果有新的成員加入時，就會得到消息提醒， 還可以玩點互動：

怎么樣好玩吧，一個元宇宙就這樣形成了，趕緊讓其他伙伴加入試試吧。

總結

元宇宙現在是個很熱的概念，但還是基于現有的技術打造的，元宇宙給人們提供了一個生活在虛擬的神奇世界里的想象空間，其實自從有了互聯網，我們就已經逐步生活在元宇宙之中了。

今天我們用基礎的 TCP 技術，構建了一個自己的元宇宙聊天室，雖然功能上和想象中的元宇宙相去甚遠，不過其中的主要功能已經成形了。

如果有興趣還可以在這個基礎上加入更好玩的功能，比如好友，群組，消息記錄等等，在深入了解的同時，讓這個元宇宙更好玩。