共计 5905 个字符,预计需要花费 15 分钟才能阅读完成。
大家好,我是查理~
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 技术,构建了一个本人的元宇宙聊天室,尽管性能上和设想中的元宇宙相去甚远,不过其中的次要性能曾经成形了。
如果有趣味还能够在这个根底上退出更好玩的性能,比方好友,群组,音讯记录等等,在深刻理解的同时,让这个元宇宙更好玩。