使用Python websockets搭建互联网服务器

article/2025/10/16 0:52:38

在上一篇文章中实现局域网的通信，使用websockets,python作服务端，unity作客服端完成网络通信
这次准备实现互联网上的通信。

1.整个云服务器

要实现互联网通信，简单的方法就是先准备一个云服务器，这里我直接使用了阿里云服务器。学习用，就整了个最便宜的，38块一年。
阿里云网站
在这里插入图片描述
然后熟悉一下阿里云服务器的使用，这里参考了这篇文章：
阿里云ECS服务器入门使用流程新手必看教程

登陆后，进入右上角控制台
在这里插入图片描述在页面里点击“云服务器ESC”

然后点击管理控制台
进入到自己服务器页面：

注意这个公网ip，远程连接时要用。

主要需要设置两个地方，一个是登陆服务器用的密码，一个是设置服务器允许的端口。
首先设置密码：
点击左侧列表中“实例”，随后在右侧“更多”中点击重置实例密码。
在这里插入图片描述
修改好后记着，这就是之后远程登陆服务器用的密码。

然后设置端口：
点击“安全组配置”
在这里插入图片描述
点击配置规则：

然后点快速添加，加入配置端口。我这里怕麻烦，直选了所有端口。
云服务器准备完毕。

2.测试云服务器

远程连接：自己电脑上打开一个终端，输入

ssh root@你服务器的公网ip

root是它默认的用户名，然后输入之前设置的密码。
在这里插入图片描述

感觉就像连接一般的远程，然后就跟用终端控制电脑一样了。
不过这时候服务器里面没有东西，ls出来都是空的。

毕竟用终端写代码很不舒服，那直接用vscode远程连接也是可以的。
在这里插入图片描述那个webServer是我新建的文件夹，其它就是服务器里面默认的东西了。

3.测试代码

服务器代码，这个放在云服务器上运行
注意服务器的ip在云服务器上的终端中使用ifconfig查看，不是公网ip

#https://www.cnblogs.com/lsdb/p/10949766.html
import asyncio
import websockets# 接收客户端消息并处理，把客户端发来的返回回去
async def recv_msg(websocket):while True:recv_text = await websocket.recv()response_text = f"context: {recv_text}"print(recv_text)await websocket.send(response_text)# 服务器端主逻辑
# websocket和path是该函数被回调时自动传过来的，不需要自己传
async def main_logic(websocket, path):await recv_msg(websocket)# 把ip换成自己本地的ip
start_server = websockets.serve(main_logic, '172.1.1.1', 8080)  #这个ip用ifconfig查看
# 如果要给被回调的main_logic传递自定义参数，可使用以下形式
# 一、修改回调形式
# import functools
# start_server = websockets.serve(functools.partial(main_logic, other_param="test_value"), '10.10.6.91', 5678)
# 修改被回调函数定义，增加相应参数
# async def main_logic(websocket, path, other_param)asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

客户端代码，这个放在自己电脑上运行
注意客户端的连接的ip是云服务器的公网ip

import asyncio
import websockets
import json# 向服务器端发送认证后的消息
async def send_msg(websocket):while True:_text = input("please enter your context: ")if _text == "exit":print(f'you have enter "exit", goodbye')await websocket.close(reason="user exit")return Falseawait websocket.send(_text)recv_text = await websocket.recv()print(f"{recv_text}")async def send_msg_json(websocket):while True:data = {"health":8000,"cards":5,}         #测试数据new_data = json.dumps(data,ensure_ascii=False)     #将data转化为字符串await websocket.send(new_data)                                #发送请求response_str = await websocket.recv()print("rec:",response_str)return True# 客户端主逻辑
async def main_logic():async with websockets.connect('ws://公网ip:8080') as websocket:await send_msg_json(websocket)asyncio.get_event_loop().run_until_complete(main_logic())

在云服务器上运行服务端代码，在自己电脑上运行客户端代码，客户端发送消息到服务端，并成功收到服务端返回的消息。