查找可用端口(改进版) Args: host: 主机地址 start_port: 起始端口 max_port: 最大端口号,默认65535 randomize: 是否在端口范围内随机选择,默认True max_retries: 最大重试次数,默认50 Returns: 可用的端口号 Raises: RuntimeError: 在指定范围内没有找到可用端口
(host: str, start_port: int, max_port: int = 65535,
randomize: bool = True, max_retries: int = 50)
| 63 | |
| 64 | |
| 65 | def find_available_port(host: str, start_port: int, max_port: int = 65535, |
| 66 | randomize: bool = True, max_retries: int = 50) -> int: |
| 67 | """查找可用端口(改进版) |
| 68 | |
| 69 | Args: |
| 70 | host: 主机地址 |
| 71 | start_port: 起始端口 |
| 72 | max_port: 最大端口号,默认65535 |
| 73 | randomize: 是否在端口范围内随机选择,默认True |
| 74 | max_retries: 最大重试次数,默认50 |
| 75 | |
| 76 | Returns: |
| 77 | 可用的端口号 |
| 78 | |
| 79 | Raises: |
| 80 | RuntimeError: 在指定范围内没有找到可用端口 |
| 81 | """ |
| 82 | port_range = max_port - start_port + 1 |
| 83 | |
| 84 | for attempt in range(max_retries): |
| 85 | if randomize: |
| 86 | # 随机选择端口以减少冲突 |
| 87 | port = start_port + random.randint(0, min(port_range - 1, 1000)) |
| 88 | else: |
| 89 | # 顺序查找 |
| 90 | port = start_port + (attempt % port_range) |
| 91 | |
| 92 | while not is_port_available(host, port): |
| 93 | port = start_port + (attempt % port_range) |
| 94 | attempt += 1 |
| 95 | time.sleep(random.uniform(0.01, 0.1)) |
| 96 | if attempt >= max_retries: |
| 97 | raise RuntimeError(f"在端口范围 {start_port}-{max_port} 内尝试{max_retries}次后没有找到可用端口") |
| 98 | return port |
| 99 | |
| 100 | |
| 101 | def find_available_port_range(host: str, worker_id: str, base_port: int = 8000, |
no test coverage detected