网络通信基础（Socket）

运维里「网络是否正常」常落地为：端口是否通、TCP 能否建立、简单协议探测。标准库 socket 无需额外依赖，适合写轻量探活脚本；复杂 HTTP 见下一章。

---

TCP 端口探测

#!/usr/bin/env python3
"""检测目标主机 TCP 端口是否可达（类似 nc -zv）。"""
import socket
import sys

def check_port(host: str, port: int, timeout: float = 3.0) -> bool:
    """
    返回 True 表示能在 timeout 内完成 TCP 三次握手。
    不发送应用层数据，适用于 MySQL/Redis/SSH 等端口探活。
    """
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except (socket.timeout, ConnectionRefusedError, OSError):
        return False

def scan_ports(host: str, ports: list[int]) -> dict[int, bool]:
    return {p: check_port(host, p) for p in ports}

if __name__ == "__main__":
    target = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1"
    checks = scan_ports(target, [22, 80, 443, 6379, 8080])
    for port, ok in sorted(checks.items()):
        status = "open" if ok else "closed"
        print(f"{target}:{port} {status}")

---

批量探活与并发（简单版）

from concurrent.futures import ThreadPoolExecutor, as_completed

def probe_many(hosts: list[str], port: int, timeout: float = 2.0) -> list[tuple[str, bool]]:
    results = []
    with ThreadPoolExecutor(max_workers=20) as pool:
        futures = {pool.submit(check_port, h, port, timeout): h for h in hosts}
        for fut in as_completed(futures):
            host = futures[fut]
            results.append((host, fut.result()))
    return sorted(results)

备注

大规模扫描注意频率与授权；内网资产表探活应限流，避免被安全设备判定为攻击。

---

DNS 解析

import socket

def resolve_host(hostname: str) -> list[str]:
    """返回 A 记录 IPv4 列表。"""
    try:
        infos = socket.getaddrinfo(hostname, None, family=socket.AF_INET)
        return list({item[4][0] for item in infos})
    except socket.gaierror as e:
        raise RuntimeError(f"DNS 解析失败: {hostname}: {e}") from e

# 示例
print(resolve_host("kubernetes.default.svc.cluster.local"))  # 在集群 Pod 内有效

---

简单 TCP 客户端（发收一行）

def send_line(host: str, port: int, message: str, timeout: float = 5.0) -> str:
    """向服务发送一行文本并读取一行响应（如某些自定义 agent）。"""
    with socket.create_connection((host, port), timeout=timeout) as sock:
        sock.settimeout(timeout)
        sock.sendall((message + "\n").encode())
        data = sock.recv(4096)
    return data.decode(errors="replace").strip()

# Redis PING 示例（无 TLS）
# send_line("127.0.0.1", 6379, "PING")  # 可能返回 +PONG

---

UDP 探测（可选）

def udp_probe(host: str, port: int, payload: bytes = b"ping", timeout: float = 2.0) -> bool:
    """UDP 无连接，仅表示能否发出；是否收到回复取决于服务。"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.settimeout(timeout)
    try:
        sock.sendto(payload, (host, port))
        sock.recvfrom(1024)
        return True
    except socket.timeout:
        return False
    finally:
        sock.close()

---

与运维场景对照

场景	做法
LB 后端是否存活	探活 host:port
K8s Service	从集群内 Pod 探 svc:port
防火墙变更验证	变更前后脚本对比端口结果
依赖就绪	部署前循环 check_port + sleep

提示

HTTP/HTTPS 健康检查优先用下一章 httpx/requests（状态码、Header、Body）；纯端口层用 socket 即可。