Closed raininboat closed 1 year ago
通过将端口分配函数包装至上下文中,在分配的过程中不释放已分配端口,避免端口冲突。
具体测试由于很难复现该issue(本地默认顺序分配端口),在本地只能通过强制设定为固定端口进行一定程度的模拟:
class free_port_selector: "对先前的端口获取函数进行二次包装,使得在上下文范围内不会重复生成套接字" def __init__(self) -> None: self._socket_list = [] def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() def get_free_port(self): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.bind(('', 12345)) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self._socket_list.append(s) # 通过在分配端口阶段维持所有套接字,理论上可以杜绝 Issue #83 return s.getsockname()[1] def close(self): for s in self._socket_list: s.close() def get_free_port(): with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s: s.bind(('', 12345)) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) return s.getsockname()[1] print("original get_free_port") for i in range(10): print(i, get_free_port()) print("changed get_free_port") with free_port_selector() as f: for i in range(10): print(i, f.get_free_port())
具体结果如下:
original get_free_port 0 12345 1 12345 2 12345 3 12345 4 12345 5 12345 6 12345 7 12345 8 12345 9 12345 changed get_free_port 0 12345 Traceback (most recent call last): File "d:\DICE\230606_dhu_web\1.py", line 38, in <module> print(i, f.get_free_port()) ^^^^^^^^^^^^^^^^^ File "d:\DICE\230606_dhu_web\1.py", line 16, in get_free_port s.bind(('', 12345)) OSError: [WinError 10048] 通常每个套接字地址(协议/网络地址/端口)只允许使用一次。
通过将端口分配函数包装至上下文中,在分配的过程中不释放已分配端口,避免端口冲突。
具体测试由于很难复现该issue(本地默认顺序分配端口),在本地只能通过强制设定为固定端口进行一定程度的模拟:
具体结果如下: