幾行代碼，擼了個(gè) 元宇宙？！

作者：派森醬 2021-12-12 18:18:15

Facebook 改名為 meta，一下子點(diǎn)燃了元宇宙這個(gè)概念。今天我就用 Python 實(shí)現(xiàn)一個(gè)簡單的迷你元宇宙。代碼簡潔易懂，不僅可以學(xué)習(xí) Python 知識(shí)，還能用實(shí)踐理解元宇宙的概念。

[[439939]]

Facebook 改名為 meta，一下子點(diǎn)燃了元宇宙這個(gè)概念。

今天我就用 Python 實(shí)現(xiàn)一個(gè)簡單的迷你元宇宙。

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

還等什么，現(xiàn)在就開始吧!

迷你元宇宙

什么是元宇宙?

不同的人有不同的理解和認(rèn)識(shí)，最能達(dá)成共識(shí)的是：

元宇宙是個(gè)接入點(diǎn)，每個(gè)人都可以成為其中的一個(gè)元素，彼此互動(dòng)。

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

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

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

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

最終的效果像這樣：

效果

設(shè)計(jì)

如何構(gòu)建接入點(diǎn)

直接思考可能比較困難，換個(gè)角度想，接入點(diǎn)其實(shí)就是 —— 服務(wù)器。

只要是上網(wǎng)，每時(shí)每刻，我們都是同服務(wù)器打交的。

那就選擇最簡單的 TCP 服務(wù)器，TCP 服務(wù)器的核心是維護(hù)套接字(socket)的狀態(tài)，向其中發(fā)送或者獲取信息。

python 的 socket 庫，提供了很多有關(guān)便捷方法，可以幫助我們構(gòu)建。

核心代碼如下：

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

創(chuàng)建一個(gè) socket，讓其監(jiān)聽機(jī)器所擁有的一個(gè) ip 和端口，然后從 socket 中讀取發(fā)送過來的數(shù)據(jù)。

如何構(gòu)建客戶端

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

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

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

代碼與服務(wù)器很像，不過去鏈接一個(gè)服務(wù)器的 ip 和端口。

如何構(gòu)建業(yè)務(wù)邏輯

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

然后當(dāng)接收到用戶消息時(shí)做出判斷，是轉(zhuǎn)發(fā)給其他用戶，廣播還是做出回應(yīng)。

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

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

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

構(gòu)建

有了初步設(shè)計(jì)，就可以進(jìn)一步構(gòu)建我們的代碼了。

服務(wù)端

服務(wù)器需要同時(shí)響應(yīng)多個(gè)鏈接，其中包括新鏈接創(chuàng)建，消息和鏈接斷開等。

為了不讓服務(wù)器阻塞，我們采用非阻塞的鏈接，當(dāng)鏈接接入時(shí)，將鏈接存儲(chǔ)起來，然后用 select 工具，等待有了消息的鏈接。

這個(gè)功能，已經(jīng)有人實(shí)現(xiàn)好了 simpletcp[1]，只要稍作改動(dòng)就好。

將其中的收到消息，建立鏈接，關(guān)閉鏈接做成回調(diào)方法，以便再外部編寫業(yè)務(wù)邏輯。

核心業(yè)務(wù)

這里說明一下核心代碼：

# 創(chuàng)建一個(gè)服務(wù)器鏈接 
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() 
 
# 退出標(biāo)記 用于關(guān)閉服務(wù)器 
self._stop = False 
 
# 服務(wù)器主循環(huán) 
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() 
             
            # 將鏈接設(shè)置為非阻塞的 
            client_socket.setblocking(0) 
            # 添加到監(jiān)聽隊(duì)列 
            readers.append(client_socket) 
            # 存儲(chǔ)ip信息 
            IPs[client_socket] = client_ip 
 
            # 調(diào)用建立鏈接回調(diào)函數(shù) 
            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: 
                # 調(diào)用收到消息回調(diào)函數(shù) 
                self.onReceiveMsg(self, sock, IPs[sock], data) 
            else: 
                # 鏈接退出時(shí)，移除監(jiān)聽隊(duì)列 
                readers.remove(sock) 
                sock.close() 
 
                # 調(diào)用鏈接關(guān)閉回調(diào)函數(shù) 
                self.onCloseConn(self, sock, IPs[sock])          
    # 處理存在錯(cuò)誤的鏈接 
    for sock in err: 
        # 移除監(jiān)聽隊(duì)列 
        readers.remove(sock) 
        sock.close() 
 
        # 調(diào)用鏈接關(guān)閉回調(diào)函數(shù) 
        self.onCloseConn(self, sock, IPs[sock])

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

事件處理

現(xiàn)在通過回調(diào)函數(shù)，就可以編寫業(yè)務(wù)了，之間看代碼。

這段是建立鏈接時(shí)的處理：

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

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

然后是接收消息的回調(diào)函數(shù)，這個(gè)相對(duì)復(fù)雜一些，主要是處理的情況更多：

def onReceiveMsg(server, sock, ip, data): 
    cid = f'{ip[0]}_{ip[1]}' 
    data = data.decode('utf-8') 
    print(f"收到數(shù)據(jù): {data}") 
    _from = clients[cid] 
    if data.startswith('name:'): 
        # 設(shè)置名稱 
        name = data[5:].strip() 
        if not name: 
            sock.send(f"不能設(shè)置空名稱，否則其他人找不見你".encode('utf-8')) 
        elif not checkname(name, cid): 
            sock.send(f"這個(gè)名字{name}已經(jīng)被使用，請(qǐng)換一個(gè)試試".encode('utf-8')) 
        else: 
            if not _from['name']: 
                sock.send(f"{name} 很高興見到你，現(xiàn)在可以暢游元宇宙了".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 轉(zhuǎn)化為字符串;if 開始處理具體的消息
如果收到 name: 開頭的消息，表示需要設(shè)置用戶名，其中包括判重，以及給其他成員發(fā)送消息
如果收到的消息里有 @，表示在發(fā)私信，先提取出需要發(fā)出的用戶們，然后將消息發(fā)送給對(duì)應(yīng)的用戶
如果沒有特殊標(biāo)記，就表示群發(fā)
其中 sendMsg 用于發(fā)送消息，接收三個(gè)參數(shù)，第一個(gè)是消息，第二是發(fā)送者，第三個(gè)是接收者名稱數(shù)組

當(dāng)鏈接關(guān)閉時(shí)，需要處理一下關(guān)閉的回調(diào)函數(shù)：

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]

當(dāng)收到鏈接斷開的消息時(shí)，合成消息，發(fā)送給其他用戶
然后從客戶端緩存中刪除

客戶端

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

我們將接收消息作為一個(gè)線程，將用戶輸入作為主循環(huán)。

接收消息

先看接收消息的代碼：

def receive(client): 
    while True: 
        try: 
            s_info = client.recv(1024)  # 接受服務(wù)端的消息并解碼 
            if not s_info: 
                print(f"{bcolors.WARNING}服務(wù)器鏈接斷開{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}服務(wù)器鏈接斷開{bcolors.ENDC}") 
            break 
        if close: 
            break

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

輸入處理

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

while True: 
    pass 
    value = input("") 
    value = value.strip() 
     
    if value == ':start': 
        if thread: 
            print(f"{bcolors.OKBLUE}您已經(jīng)在元宇宙中了{(lán)bcolors.ENDC}") 
        else: 
            client = createClient(ip, 5000) 
            thread = Thread(target=receive, args=(client,)) 
            thread.start() 
            print(f"{bcolors.OKBLUE}您進(jìn)入元宇宙了{(lán)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}還沒接入元宇宙，請(qǐng)先輸入 :start 接入{bcolors.ENDC}') 
    client.close()

主要是對(duì)不同的命令做出的相應(yīng)，比如 :start 表示需要建立鏈接，:quit 表示退出等
命令前加 : 是為了和一般的消息做區(qū)分，如果不帶 : 就認(rèn)為是在發(fā)送消息