关于python:最近兴起的元宇宙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:

而后再关上一个命令行，执行 python metaClient.py，输出指令 :start，就能够接入到元宇宙：

设置本人的名字：

如果有新的成员退出时，就会失去音讯揭示，还能够玩点互动：

怎么样好玩吧，一个元宇宙就这样造成了，连忙让其余搭档退出试试吧。

元宇宙当初是个很热的概念，但还是基于现有的技术打造的，元宇宙给人们提供了一个生存在虚构的神奇世界里的设想空间，其实自从有了互联网，咱们就曾经逐渐生存在元宇宙之中了。

明天咱们用根底的 TCP 技术，构建了一个本人的元宇宙聊天室，尽管性能上和设想中的元宇宙相去甚远，不过其中的次要性能曾经成形了。

如果有趣味还能够在这个根底上退出更好玩的性能，比方好友，群组，音讯记录等等，在深刻理解的同时，让这个元宇宙更好玩。

关于python:最近兴起的元宇宙Python几行代码就能撸

迷你元宇宙

设计

如何构建接入点

如何构建客户端

如何构建业务逻辑

构建

服务端

外围业务

事件处理

客户端

接管音讯

输出解决

启动

总结