用 paramiko 对几十上百台机器批量跑命令，最容易踩的两个坑是：远端命令卡住导致线程永久阻塞，以及stderr 没读干净导致 exit_status 拿到 None。前者会让整个批量任务的时间不可控，后者会让你明明看到失败却拿不到错误原因。paramiko 本身没有内置的命令级超时，exec_command 的 timeout 参数只作用于 channel 建立阶段，不作用于命令执行阶段，这一点被大量教程误读。
下面这套实现解决三件事：给每条命令一个可中断的执行上限、把 stdout 与 stderr 完整分流收集、在被中断时给出可诊断的返回结构而不是抛出一个没有上下文的异常。代码按可直接复制的粒度给出，线程池、信号量、单机连接复用、批量调度全部包含。
为什么 exec_command 的 timeout 参数不够用
先看一个常见的错误写法：
import paramiko
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect("10.0.0.11", port=22, username="ops", password="secret", timeout=10)
# 这里的 timeout 只管 connect，不管命令
stdin, stdout, stderr = client.exec_command("sleep 300", timeout=10)
print(stdout.read()) # 这里会阻塞接近 300 秒
阅读 paramiko 源码（client.py 中的 exec_command）可以确认：传入的 timeout 被赋给 self._transport.open_session(timeout=timeout)，也就是打开 session channel 的超时，命令自身的执行不受它约束。stdout.read() 内部调用 chan.recv()，默认以阻塞模式等待 EOF，只要远端进程不结束，这个调用就不会返回。
更隐蔽的问题是：即使你在别的线程里给 stdout.read() 加了超时，stderr 也可能卡住。paramiko 的 channel 有窗口流控（window size 默认 2097152 字节，max packet size 32768），当远端往 stderr 写的数据超过窗口而本地没有读走时，远端进程会阻塞在 write 上，此时 stdout 也等不到 EOF。所以必须同时、持续地消费两个流。
第三点：exec_command 返回后立刻调用 stdout.channel.recv_exit_status() 是另一个经典陷阱。如果 stdout/stderr 还没读完就调它，paramiko 内部会因为 channel 未关闭而一直等待，表现为「命令明明执行完了，程序却卡住」。正确顺序永远是：先读干净两个流，再取 exit_status。
核心实现：基于 select 的非阻塞读取 + 命令级超时
不依赖第三方库，用 channel.recv_ready() / recv_stderr_ready() 配合 select.select 做轮询，是 paramiko 场景下最稳的做法。它的好处是单线程内同时照看两个流，且能精确判断超时时刻已经收到了多少数据。
import select
import socket
import time
from dataclasses import dataclass, field
import paramiko
@dataclass
class CmdResult:
host: str
command: str
exit_code: int | None
stdout: str
stderr: str
timed_out: bool = False
error: str = ""
duration: float = 0.0
@property
def ok(self) -> bool:
return self.exit_code == 0 and not self.timed_out and not self.error
def run_command(
client: paramiko.SSHClient,
host: str,
command: str,
timeout: float = 30.0,
idle_timeout: float = 0.0,
encoding: str = "utf-8",
max_bytes: int = 4 * 1024 * 1024,
) -> CmdResult:
"""在已连接的 client 上执行一条命令，硬超时 timeout 秒。"""
start = time.monotonic()
out_buf: list[bytes] = []
err_buf: list[bytes] = []
out_len = err_len = 0
exit_code: int | None = None
try:
chan = client.get_transport().open_session(timeout=10)
chan.settimeout(0.0) # 关键：切到非阻塞，避免 recv 卡死
chan.exec_command(command)
deadline = start + timeout
last_data_at = start
while True:
now = time.monotonic()
if now >= deadline:
chan.close()
return CmdResult(
host, command, None,
b"".join(out_buf).decode(encoding, "replace"),
b"".join(err_buf).decode(encoding, "replace"),
timed_out=True,
error=f"hard timeout after {timeout}s",
duration=now - start,
)
if idle_timeout and now - last_data_at > idle_timeout:
chan.close()
return CmdResult(
host, command, None,
b"".join(out_buf).decode(encoding, "replace"),
b"".join(err_buf).decode(encoding, "replace"),
timed_out=True,
error=f"idle timeout after {idle_timeout}s",
duration=now - start,
)
got = False
# select 的第三个参数只接受带 fileno() 的对象
rlist = []
if chan.recv_ready():
rlist.append(chan)
if chan.recv_stderr_ready():
rlist.append(chan)
if rlist:
select.select(rlist, [], [], 0.05)
else:
select.select([chan], [], [], 0.05)
if chan.recv_ready():
chunk = chan.recv(32768)
if chunk:
out_len += len(chunk)
if out_len <= max_bytes:
out_buf.append(chunk)
got = True
if chan.recv_stderr_ready():
chunk = chan.recv_stderr(32768)
if chunk:
err_len += len(chunk)
if err_len <= max_bytes:
err_buf.append(chunk)
got = True
if got:
last_data_at = time.monotonic()
if chan.exit_status_ready() and not chan.recv_ready() \
and not chan.recv_stderr_ready():
exit_code = chan.recv_exit_status()
break
return CmdResult(
host, command, exit_code,
b"".join(out_buf).decode(encoding, "replace"),
b"".join(err_buf).decode(encoding, "replace"),
duration=time.monotonic() - start,
)
except (paramiko.SSHException, socket.error, EOFError) as exc:
return CmdResult(
host, command, None,
b"".join(out_buf).decode(encoding, "replace"),
b"".join(err_buf).decode(encoding, "replace"),
error=f"{type(exc).__name__}: {exc}",
duration=time.monotonic() - start,
)
finally:
try:
chan.close()
except Exception:
pass
几个必须解释的设计点：
chan.settimeout(0.0)：把 channel 切到非阻塞。paramiko 的Channel.settimeout直接影响底层 socket，设为 0 后recv没数据时抛socket.timeout；但由于我们每次调用前都先判断recv_ready()，实际上不会触发异常。exit_status_ready()与两次 ready 检查的组合：只有「退出状态已就绪」且「两个缓冲区都空了」才算真正结束，避免漏掉尾部输出。idle_timeout：可选参数，用于对付「不退出但长期无输出」的进程，比如某些交互式脚本。默认 0 表示关闭。max_bytes：防止某条命令cat了一个大文件把内存打爆。超限部分直接丢弃但继续排空 channel，否则远端仍会被窗口流控卡住。- 退出码用
None表示「没拿到」，与 `0`（成功）区分开，下游判断逻辑才不会出错。
如果你在 Windows 上运行，
select.select只支持 socket，paramiko 的 Channel 对象恰好满足这个条件，所以上述代码跨平台可用；但不要试图把sys.stdin之类的对象塞进 select 的列表里，Windows 会直接报OSError: [WinError 10038]。
连接复用与批量调度：线程池的正确打开方式
批量执行时，每台机器建一个连接、在连接上串行跑多条命令是性价比最高的方案。不要为每条命令建一次连接——SSH 握手加认证在高并发下会成为瓶颈，而且很多堡垒机有连接频率限制。
from concurrent.futures import ThreadPoolExecutor, as_completed
import paramiko
def make_client(host: str, username: str, password: str = None,
key_path: str = None, port: int = 22,
connect_timeout: float = 10.0) -> paramiko.SSHClient:
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
kwargs = dict(
hostname=host, port=port, username=username,
timeout=connect_timeout,
banner_timeout=connect_timeout,
auth_timeout=connect_timeout,
allow_agent=False, look_for_keys=False,
)
if key_path:
kwargs["key_filename"] = key_path
else:
kwargs["password"] = password
client.connect(**kwargs)
# 保活：防止长时间批量任务中被中间的防火墙掐断
client.get_transport().set_keepalive(30)
return client
def run_batch(hosts, commands, username, password=None, key_path=None,
workers: int = 16, cmd_timeout: float = 30.0):
results = []
def work(host):
local = []
try:
client = make_client(host, username, password, key_path)
except Exception as exc:
for cmd in commands:
local.append(CmdResult(host, cmd, None, "", "",
error=f"connect failed: {exc}"))
return local
try:
for cmd in commands:
local.append(run_command(client, host, cmd, timeout=cmd_timeout))
finally:
client.close()
return local
with ThreadPoolExecutor(max_workers=workers) as pool:
futures = {pool.submit(work, h): h for h in hosts}
for fut in as_completed(futures):
results.extend(fut.result())
return results
参数取值建议，来自实际跑过几百台机器的经验：
| 参数 | 建议值 | 说明 |
|---|---|---|
workers |
8 ~ 32 | 超过 64 时本机文件描述符和远端 sshd 的 MaxStartups 都可能成为瓶颈 |
connect_timeout |
10s | 同时设 timeout/banner_timeout/auth_timeout，只设 timeout 挡不住认证阶段卡住 |
cmd_timeout |
30 ~ 120s | 按命令类型分档，别给所有命令一个统一的大值 |
set_keepalive |
30s | 批量任务超过 5 分钟必设，否则中间设备会静默断链 |
max_bytes |
4MB | 需要完整日志时改为写临时文件再 sftp 拉回 |
注意 allow_agent=False, look_for_keys=False：批量场景下如果你显式传了密码，却因为本机有 ssh-agent 或 ~/.ssh/id_rsa 而导致 paramiko 优先尝试密钥，会出现「明明密码对却认证失败」或者反复尝试造成远端 MaxAuthTries（默认 6）超限。显式关掉能避免这类干扰。
真实报错与排查对照表
下面这些都是实际会遇到、且网上资料经常答非所问的错误，逐一给出成因和处置。
1. paramiko.ssh_exception.SSHException: Server '10.0.0.11' not found in known_hosts
用了 RejectPolicy（paramiko 2.x 之后的默认策略在部分版本中表现为拒绝未知主机）。批量内网环境常用 AutoAddPolicy；安全要求高的环境应预先分发 known_hosts，用 client.load_host_keys('/path/known_hosts') 配合 RejectPolicy。不要用 WarningPolicy 敷衍，它只打印警告不阻断。
2. socket.timeout: timed out（发生在 connect 阶段）
多数不是网络问题，而是 banner_timeout 太小。某些设备（老版本 OpenSSH、网络设备）在 TCP 建连后要好几秒才发 banner，默认 15 秒有时仍不够，尤其是高并发时。把 banner_timeout 提到 30 秒再试。如果是 Authentication timeout，则是 auth_timeout 的问题。
3. paramiko.ssh_exception.AuthenticationException: Authentication failed.
三类原因：密码错；用户名错（很多镜像默认禁用 root 密码登录，要用普通用户 + sudo）；远端 sshd 配置了 PubkeyAuthentication yes 且 PasswordAuthentication no。第三种情况日志里通常只有上面这一行，需要在远端看 /var/log/auth.log 或 journalctl -u sshd 才能确认。
4. exit_status 拿到 None
就是我们开头说的顺序问题。还有一种情况：远端命令是被 kill -9 掉的，或者 shell 在执行 exec 之前就崩了，也可能拿不到状态。判断逻辑应写成 exit_code == 0 才算成功，而不是 != 1。
5. 命令返回码 127 / 126
127 是 command not found，126 是权限不足。批量场景下这两个码极高频，根因通常是 非交互式 SSH 会话的 PATH 与登录 shell 不同。paramiko 执行命令时用的是非登录、非交互 shell，不会读 ~/.bash_profile 里的 PATH 追加。解决方式：命令里写绝对路径，或在命令前加 source /etc/profile; source ~/.bashrc;。注意 ~/.bashrc 顶部常有 [ -z "$PS1" ] && return，非交互下会直接返回，所以更稳的是显式 export PATH=/usr/local/bin:$PATH。
6. ChannelException: Administratively prohibited
远端 sshd 的 MaxSessions（默认 10）被打满，或者中间有跳板机限制。表现为同一个连接上开第 11 个 session 失败。批量工具里要确保 chan.close() 一定执行，finally 块不能省。
7. 读到的中文是乱码
远端 locale 是 C 或 POSIX 时，输出的中文按字节透传，本地用 utf-8 解码就会出现 UnicodeDecodeError。用 decode(encoding, "replace") 兜底，并在命令前加 export LANG=en_US.UTF-8;（如果远端装了该 locale）。也可以用 errors="backslashreplace" 保留原始字节信息便于排查。
超时之后的清理：别让僵尸 channel 拖垮连接
超时分支里 chan.close() 只是关闭 channel，远端进程未必会死。如果远端跑的是 tail -f、ping 这类不响应 SIGHUP 的进程，它会继续占着资源。稳妥做法是在命令外层包一层 timeout 命令，做双保险：
# 远端侧超时：命令最多跑 25 秒，比本地 30 秒硬超时略小，让远端先自己收尾
REMOTE_WRAP = "timeout -k 5 {secs} bash -lc {cmd!r}"
def wrap(command: str, secs: int = 25) -> str:
return f"timeout -k 5 {secs} bash -lc {command!r}"
timeout 是 coreutils 自带，主流 Linux 发行版都有（Alpine 需要装 coreutils）。-k 5 表示发出 TERM 后 5 秒仍未退出就发 KILL。远端超时后返回码是 124（TERM 生效）或 137（被 KILL），在结果里可以据此区分「远端主动超时」和「本地硬超时」。
另外，超时关闭 channel 后，不建议复用这个 SSHClient 继续跑下一条命令。虽然 Transport 层通常还活着，但已被 kill 的远端进程可能还在往 channel 写数据，残留的窗口流控状态会干扰后续会话。实践中的处理是：一条命令超时就把这台机器标记为 degraded，剩余命令跳过，连接直接关闭，交给下一轮重试。
小结
paramiko 做批量命令执行，核心就三句话：exec_command 的 timeout 不管命令执行，命令级超时必须自己在 channel 层用非阻塞读实现；stdout 和 stderr 必须同时排空，否则窗口流控会让远端 write 卡死、本地永远等不到 EOF；先读完流再取 exit_status，顺序错了就拿到 None 或者直接挂起。把远端 timeout 命令作为第二道防线，配合清晰的 CmdResult 结构把「超时 / 连接失败 / 命令失败」三种情况区分开，批量任务的耗时才真正可控。
完整代码去掉注释大约 120 行，不依赖 paramiko 以外的第三方库。如果你需要更复杂的场景（交互式提权、SFTP 与命令混合执行），可以在此基础上扩展，但上面这套骨架足以覆盖 90% 的运维批量任务。
FAQ
Q：能不能直接用 stdout.channel.settimeout(30) 然后 stdout.read()？
能跑通，但有两个问题。一是 read() 会读满一整块或等到 EOF，遇到超时抛 socket.timeout 时，已经读到的数据会随异常一起丢失，你拿不到超时前那部分输出。二是 stderr 仍然没被消费，流控卡死的问题依旧存在。所以宁可用上面的循环写法。
Q：为什么不用 stdout.read() + stderr.read() 顺序读？
因为这是串行阻塞的。当远端先往 stderr 写满 2MB 窗口，而你在等 stdout 的 EOF 时，双方互等，就是死锁。小输出场景碰巧不出问题，一旦命令有大量错误日志就会复现，非常难查。
Q：chan.settimeout(0.0) 之后 recv 会不会抛异常？
没数据时会抛 socket.timeout（paramiko 里也可能表现为返回空字节或抛 BlockingIOError 的变体，取决于版本）。所以代码里每次 recv 之前都先判断 recv_ready() / recv_stderr_ready()，这是避免异常的唯一可靠方式。
Q：批量执行时提示 OSError: [Errno 24] Too many open files
线程数 × 每连接的文件描述符超过了 ulimit -n。先 ulimit -n 看当前限制，临时调到 65535（ulimit -n 65535），同时把 workers 降到 16 以下。根本办法是控制并发，而不是无限调大限制。
Q：远端返回码 255 是什么意思？
255 是 ssh 客户端自身的错误码，通常表示连接被中断、主机密钥校验失败或者远端 sshd 拒绝。如果命令本身返回 255，在 paramiko 里往往伴随 SSHException 或者连接直接断开。看到 255 优先查网络和认证，不要查命令本身。
Q：需要区分 stdout 和 stderr 的先后顺序怎么办？
paramiko 的 channel 把两个流分开缓冲，不保留时序。如果需要严格时序（比如构建日志），最省事的办法是在远端命令后追加 2>&1 合并到 stdout，一条流读出来即天然有序。这时就只用 recv_ready() 一个判断，循环还能再简化。