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

问题描述 投票:2回答:2

我正在围绕Appium服务器编写Python包装器。 Appium接受要绑定的本地端口的命令行参数。不幸的是,Appium无法为自己自动选择一个空闲端口,因此它要么绑定到明确指定的端口,要么使用EADDRINUSE失败。即使告诉它绑定到端口0,它也会成功启动,但不会显示它绑定的端口。

如果我自己在Python包装器中找到一个空闲端口,则无法保证某些其他进程不会绑定到同一个端口,同时我将它传递给Appium。如果我不首先发布它,Appium将无法绑定它,所以我必须这样做。

我知道这在实践中不太可能发生,但是在以跨平台方式(Linux,macOS,Windows)将其传递给另一个进程之前,“保留”本地端口号的“正确方法”是什么?

python linux macos networking network-programming
2个回答
1
投票

感谢评论中的@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}', '[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端口。它还会跳过所有类型的主机/端口/用户名反向查找以使其更快,并且不需要提升权限。

所以,最后,我们的想法是让Appium(或其他任何东西)从0.0.0.0:0开始,它将自己绑定到操作系统提供的第一个可用端口,然后检查它正在侦听的端口。没有比赛条件。


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,如果端口指定为零,则服务提供者从动态客户端端口范围为应用程序分配唯一端口。

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

在ip_local_port_range中,您可以阅读以下内容:

在以下情况下,临时端口被分配给套接字:

  • 调用bind(2)时,套接字地址中的端口号被指定为0;

getsockname()用于知道已选择哪个端口。

© www.soinside.com 2019 - 2024. All rights reserved.