如何为子进程选择一个空闲端口？

Question

我正在编写一个Python封装的Appium服务器。Appium接受用于本地端口绑定的命令行参数。不幸的是，Appium无法自动选择一个空闲端口，所以它要么绑定到明确指定的端口，要么失败并显示EADDRINUSE错误。即使告诉它绑定到端口0，它也会成功启动，但无法显示它绑定的端口。

如果在Python封装中自己找到一个空闲端口，那么就没有保证其他进程在此期间不会绑定到同一端口。如果我不先释放它，Appium将无法绑定到该端口，因此我必须先释放它。

我知道这在实践中不太可能发生，但在Linux、macOS和Windows上，在向其他进程传递之前“预留”本地端口号的“正确方式”是什么？

- toriningen

---

你可以随机选择一个端口，将其传递给Appium，并检查是否显示了正确的错误信息。 - Alex Hall

你能不能尝试使用任意端口，如果返回EADDRINUSE错误，就递增端口号并循环直到找到一个可用的？ - rodrigo

@toriningen：当我说“pick”时，我的意思是“选择”，而不是“绑定”。如果端口在服务器启动时已被使用，则递增并重复。除非你没有可用的空闲端口，否则最终会成功。 - rodrigo

@rodrigo，但这意味着每次失败尝试都需要启动一个重量级的服务器进程，而我正试图避免这种情况，并且这也是我目前正在做的事情。 - toriningen

@toriningen：啊，一个重型服务器...我其实对Appium一无所知，但我还是会给你建议。如果以0成功启动，我可以想到两个解决方案：1. 修补Appium以报告使用的端口（毕竟它是开源的）；2. 使用类似lsof -p <pid> -i4 -P -n | grep LISTEN这样的命令来发现它正在使用的端口。 - rodrigo

Answer 1

Selenium库使用这个技巧：

https://github.com/SeleniumHQ/selenium/blob/master/py/selenium/webdriver/common/utils.py#L31

import socket

def free_port():
    """
    Determines a free port using sockets.
    """
    free_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    free_socket.bind(('0.0.0.0', 0))
    free_socket.listen(5)
    port = free_socket.getsockname()[1]
    free_socket.close()
    return port

如果将套接字绑定到端口0，内核会为其分配一个空闲端口。这适用于Windows和Linux操作系统。

https://msdn.microsoft.com/en-us/library/windows/desktop/ms737550.aspx

对于TCP/IP协议，如果端口号被指定为0，则服务提供者会从动态客户端端口范围中为应用程序分配一个唯一的端口。

http://man7.org/linux/man-pages/man7/ip.7.html

在ip_local_port_range中，您可以阅读以下内容：

在以下情况下，将为套接字分配临时端口：

• 在调用bind（2）时，将套接字地址中的端口号指定为0;

使用getsockname（）可知道选择了哪个端口。

- miguel martinez

---

虽然这可能回答了问题，但需要一些解释。请更新问题并说明此解决方案如何以及为什么有效。 - Brett DeWoody

你的代码片段只是选择了第一个可用端口，这与问题无关。关键在于将其传递给第三方子进程时不引入竞态条件，而不仅仅是选择端口。 - toriningen

Answer 2

感谢@rodrigo在评论中的建议，我最终使用了这段代码：

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}', '-i4TCP@127.0.0.1', '-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端口。为了加快速度，它还跳过了所有种类的主机/端口/用户名反向查找，并且不需要提升权限。因此，最后的想法就是让Appium（或其他任何东西）只在0.0.0.0:0上启动，它会绑定到操作系统提供的第一个可用端口，然后检查它现在正在侦听哪些端口。没有竞争条件。

- toriningen