miropython urequests和gnet的兼容性问题
gnet 是一个高性能的golang 网络io,性能和xlang java c++ 的几个框架一起占在网络框架的天花板上。并多次位居榜首。
但是gnet的文档不太全，有一些小坑需要一点时间排查。
按照gnet的例子创建的http serv 在api调试工具 curl 浏览器 都没啥问题，但是在miropython的urequests库下却有问题。看了gnet部分，不想轻易去修改，因为性能在目前项目场景下真的太重要了。
看了一下urequests.py感觉不好直接修改，但是参考了一下自己实现一个客户端还是没问题的。miropython的代码嘛，在设备里面运行，只要没有bug能跑起来就可以，不用太操心性能问题，毕竟设备都卖给用户了，占用的性能也不是咱们的。
python
import usocket
# 目标服务器信息
#host = "10.1.1.75"
#port = 8084
#url = "/devs/?cid=xxx"
# timeout =1
def HttpGet(host,port,url,timeout):
try:
# 创建 socket 连接
s = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)
addr = usocket.getaddrinfo(host, port)[0][-1]
s.connect(addr)
# 构建 GET 请求
request = "GET {url} HTTP/1.1\r\nHost: {host}\r\nUser-Agent: myUrequests\r\n\r\n".format(url=url, host=host)
s.send(request) # 发送请求
response = b"" # 接收响应
s.settimeout(timeout) # 设置超时时间为x秒
while True:
try:
data = s.recv(4096)
if not data:
break
response += data
except OSError:
break
# 关闭 socket 连接
s.close()
# 解析响应数据
headers, body = response.split(b'\r\n\r\n', 1)
content_length = 0
for header in headers.split(b'\r\n'):
if header.startswith(b'Content-Length:'):
content_length = int(header.split(b':')[1])
break
return body[:content_length].decode()
except:
print("http get错误")
return ""
def HttpsGet(host, port, url, timeout):
import ussl
import select
import time
# 创建 socket 连接
s = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)
addr = usocket.getaddrinfo(host, port)[0][-1]
s.connect(addr)
# 使用 SSL 加密连接
s = ussl.wrap_socket(s, server_hostname=host)
# 设置超时时间
s.setblocking(False)
# 构建 GET 请求
request = "GET {url} HTTP/1.1\r\nHost: {host}\r\nUser-Agent: myUrequests\r\n\r\n".format(url=url, host=host)
s.write(request.encode()) # 发送请求
# 等待响应数据
response = b""
start_time = time.time()
while True:
ready = select.select([s], [], [], 0.1) # 每次等待 0.1 秒钟
if ready[0]:
try:
data = s.read(4096)
if data:
response += data
else:
break
except OSError:
break
# 检查超时
elapsed_time = time.time() - start_time
if elapsed_time >= timeout:
break
# 关闭 socket 连接
s.close()
# 解析响应数据
headers, body = response.split(b'\r\n\r\n', 1)
content_length = 0
for header in headers.split(b'\r\n'):
if header.startswith(b'Content-Length:'):
content_length = int(header.split(b':')[1])
break
return body[:content_length].decode()
def AutoHttpGet(url, timeout):
try:
# 解析URL
parts = url.split('://')
scheme = parts[0]
remaining = parts[1].split('/', 1)
host_port = remaining[0].split(':')
if len(host_port) == 1: #默认端口
host = host_port[0]
port = 443 if scheme == 'https' else 80
else:#自定义端口
host = host_port[0]
port = int(host_port[1])
path = '/' + remaining[1] if len(remaining) > 1 else '/'
query = ''
if '?' in path:
path, query = path.split('?', 1)
query = '?' + query
print(scheme,host, port, path + query)
if scheme == 'https':
return HttpsGet(host, port, path + query, timeout)
else:
return HttpGet(host, port, path + query, timeout)
except:
print("自动http/s get错误")
return ""