问题描述
我正在围绕Appium服务器编写Python包装器.Appium接受用于绑定到本地端口的命令行参数.不幸的是,Appium无法为其自身自动选择一个空闲端口,因此它要么绑定到明确指定的端口,要么失败,并显示 EADDRINUSE
.即使告诉它绑定到端口 0
,它也将成功启动,但不会显示其绑定到的端口.
I am writing a Python wrapper around Appium server. Appium accepts command-line parameter for a local port to bind to. Unfortunately, Appium cannot autoselect a free port for itself, so it either binds to explicitly specified port, or fails with EADDRINUSE
. Even when telling it to bind to port 0
, it will start successfully, but won't display what port it had bound to.
如果我自己在Python包装器中找到了一个免费端口,则无法保证在我将其传递给Appium的同时,其他某些进程也不会绑定到同一端口.而且,如果我自己不首先发布它,Appium将无法绑定到它,所以我必须这么做.
If I find a free port myself in the Python wrapper, there is no guarantee that some other process won't bind to the same port meanwhile I am passing it to Appium. And if I don't release it first myself, Appium won't be able to bind to it, so I have to.
我知道这在实践中几乎不可能发生,但是在以跨平台方式将其传递给另一个进程(Linux,macOS,Windows)之前,保留"本地端口号的正确方法"是什么??
I know this is unlikely to ever happen in practice, but what would be the "right way" to "reserve" a local port number before passing it to another process in a cross-platform way (Linux, macOS, Windows)?
推荐答案
感谢@rodrigo的建议,我最终得到了以下代码:
Thanks to @rodrigo suggestion in comments, I have ended up with this code:
import platform
import re
import subprocess
from typing import Set
if platform.system() == 'Windows':
def _get_ports(pid):
sp = subprocess.run(['netstat', '-anop', 'TCP'],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
check=True)
rx_socket = re.compile(br'''(?x) ^
\s* TCP
\s+ 127.0.0.1 : (?P<port>\d{1,5})
\s+ .*?
\s+ LISTENING
\s+ (?P<pid>\d+)
\s* $''')
for line in sp.stdout.splitlines():
rxm = rx_socket.match(line)
if rxm is None:
continue
sock_port, sock_pid = map(int, rxm.groups())
if sock_pid == pid:
yield sock_port
else:
def _get_ports(pid):
sp = subprocess.run(['lsof', '-anlPFn', '+w',
f'-p{pid}', '[email protected]', '-sTCP:LISTEN'],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
check=True)
for line in sp.stdout.splitlines():
if line.startswith(b'n'):
host, port = line.rsplit(b':', 1)
port = int(port)
yield port
def get_ports(pid: int) -> Set[int]:
"""Get set of local-bound listening TCPv4 ports for given process.
:param pid: process ID to inspect
:returns: set of ports
"""
return set(_get_ports(pid))
print(get_ports(12345))
它可在Linux,macOS和Windows上运行,并找出处于LISTEN状态的给定进程的所有本地绑定的TCPv4端口.它还跳过所有类型的主机/端口/用户名反向查找,以使其更快,并且不需要提升的特权.
It works on Linux, macOS and Windows, and finds out all locally-bound TCPv4 ports for given process that are in LISTEN state. It also skips all kinds of host/port/username reverse look-ups to make it faster, and does not require elevated privileges.
因此,最后的想法是让Appium(或其他任何东西)从 0.0.0.0:0
开始,它将自身绑定到OS提供的第一个可用端口,并且然后检查它现在正在侦听哪些端口.没有比赛条件.
So, finally, the idea is to just let Appium (or anything else) start on 0.0.0.0:0
, it will bind itself to the first available port, as provided by OS, and then inspect what ports is it now listening on. No race conditions.
这篇关于如何为子进程选择空闲端口?的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!