Python+Paramiko多线程SSH批量巡检与Excel可视化报表实战

简介这是一款面向运维工程师与Python自动化学习者的服务器巡检工具基于Python与Paramiko实现SSH远程连接支持多线程并发执行远程命令脚本可同时对多台服务器完成CPU、内存、磁盘、网络等状态检测解决传统人工巡检重复耗时的问题。资源包共10个文件约40KB包含py主程序脚本、cmd批处理、host与hosts主机配置、xls巡检报表、log运行日志、md与docx说明文档及txt附赠资源覆盖从任务配置到结果输出的完整链路。工具可将巡检数据自动导出为Excel报表并借助Pyecharts生成可视化图表便于直观掌握服务器运行状况。目前已有149人学习下载适合希望快速搭建批量巡检流程、参考多线程远程执行与报表生成实现思路的读者也可作为运维自动化入门与二次开发的实践素材。1. 运维巡检这件事为什么手工 SSH 迟早会把你拖垮凌晨两点被告警叫醒登上去一看是某台机器的磁盘满了。你揉着眼睛打开终端一台一台 SSH 上去敲df -h、free -m、uptime敲到第七台的时候已经忘了第三台的输出是什么。这不是段子这是很多中小团队运维的日常。服务器规模一旦超过十台纯手工巡检就会从「顺手的事」变成「迟早出事的事」——漏检、误判、记录丢失全是血泪经验。这个标题讲的就是把这套动作自动化用 Python 加 Paramiko 建立 SSH 连接多线程并发把命令脚本推到几十上百台机器上执行回收结果后自动生成 Excel 明细表和 Pyecharts 可视化报表。它解决的核心痛点是「批量」和「留痕」——批量让你从重复劳动里脱身留痕让你在复盘和汇报时有据可查。适合谁适合手里管着十几到几百台 Linux 服务器、还没上重型监控平台、又想快速拿到巡检结果的运维和开发同学。整套东西不依赖 Agent纯 SSH落地门槛低一台能跑 Python 的机器就能当控制端。2. Paramiko 建连与命令执行把 SSH 这件事拆开看2.1 为什么选 Paramiko 而不是 subprocess 调 ssh很多人第一反应是subprocess直接调系统ssh命令简单粗暴。但这条路在批量场景下会翻车每台机器起一个子进程密码交互要靠sshpass之类的工具跨平台兼容性差而且你拿不到结构化的返回码和错误流。Paramiko 是纯 Python 实现的 SSH 协议库连接、执行、读流全在进程内完成异常可以精确捕获天然适合并发。选型上还有一个关键点Paramiko 支持密钥和密码两种认证。生产环境我一般强制走密钥密码只作为兜底。密钥认证下connect不需要传password用key_filename指定私钥路径即可。下面是最小可运行的单机执行示例。import paramiko def run_once(host, user, key_path, cmd, timeout10): # 创建 SSH 客户端实例 client paramiko.SSHClient() # 自动接受未知主机指纹生产环境建议换成 load_system_host_keys client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect( hostnamehost, usernameuser, key_filenamekey_path, timeouttimeout, # 建连超时别设太大 banner_timeouttimeout, # 握手 banner 超时 auth_timeouttimeout # 认证超时 ) # exec_command 返回三个类文件对象 stdin, stdout, stderr client.exec_command(cmd, timeouttimeout) out stdout.read().decode(utf-8, errorsignore) err stderr.read().decode(utf-8, errorsignore) code stdout.channel.recv_exit_status() # 必须读否则可能阻塞 return {host: host, code: code, out: out, err: err} except Exception as e: return {host: host, code: -1, out: , err: str(e)} finally: client.close() # 无论成败都要关否则连接泄漏逻辑说明exec_command之后必须调用recv_exit_status()否则通道可能一直挂着不返回这是新手最容易踩的坑。timeout参数分三个层次——建连、banner、认证任何一个卡住都会拖垮整个批次。参数建议内网环境timeout设 5 到 10 秒足够跨机房可以放宽到 15 秒但不要超过 30 秒否则单台故障会拖慢整轮巡检。2.2 命令脚本怎么组织才不混乱巡检命令不要硬编码在 Python 里抽成一个独立的命令清单文件比如commands.yaml或纯文本。每条命令带一个key和desc方便后面映射到 Excel 列。常见做法是这样# commands.yaml - key: disk desc: 磁盘使用率 cmd: df -h | grep -v tmpfs - key: mem desc: 内存使用 cmd: free -m - key: load desc: 系统负载 cmd: uptime - key: top_cpu desc: CPU占用前五进程 cmd: ps aux --sort-%cpu | head -6读取时用yaml.safe_load遍历执行。这样加一条巡检项只改配置不改代码团队里谁都能维护。注意命令里尽量用grep -v tmpfs这类过滤把无关输出剔掉否则 Excel 里全是噪音。命令本身要幂等、只读绝对不要在巡检脚本里放rm、kill这类写操作——巡检就是巡检别让它有副作用。3. 多线程并发执行线程池怎么设、结果怎么收3.1 ThreadPoolExecutor 的并发模型Paramiko 的执行是 IO 密集型等网络返回的时间远大于 CPU 计算所以多线程是合适的不需要上多进程。用concurrent.futures.ThreadPoolExecutor最省心。核心问题是并发数设多少设太小跑得慢设太大目标机的 sshd 可能拒绝连接甚至触发防护。我的经验值是并发数控制在 20 到 50 之间具体看目标机数量和网络质量。50 台以内可以开到 20上百台建议 30 到 50并且配合重试。下面是把单机执行和线程池拼起来的骨架。from concurrent.futures import ThreadPoolExecutor, as_completed import yaml def load_commands(pathcommands.yaml): with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def inspect_host(host, user, key_path, commands): 对单台机器执行全部巡检命令返回结构化结果 result {host: host, items: {}, error: } client paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect(hostnamehost, usernameuser, key_filenamekey_path, timeout10, banner_timeout10, auth_timeout10) for item in commands: _, stdout, stderr client.exec_command(item[cmd], timeout15) out stdout.read().decode(utf-8, errorsignore) stdout.channel.recv_exit_status() result[items][item[key]] out.strip() except Exception as e: result[error] str(e) finally: client.close() return result def run_all(hosts, user, key_path, max_workers30): commands load_commands() results [] with ThreadPoolExecutor(max_workersmax_workers) as pool: futures {pool.submit(inspect_host, h, user, key_path, commands): h for h in hosts} for fut in as_completed(futures): host futures[fut] try: results.append(fut.result(timeout60)) except Exception as e: results.append({host: host, items: {}, error: str(e)}) return results逻辑说明as_completed让结果按完成顺序回收不用等最慢的那台。fut.result(timeout60)是第二道保险防止某台机器在inspect_host内部卡死导致整个批次挂起。参数上max_workers是并发上限timeout60是单台总耗时上限两者要配合——如果单台要跑 10 条命令、每条 15 秒理论上限就是 150 秒那 60 秒就不够得按命令条数动态算。3.2 连接复用与失败重试每台机器只建一次连接、跑完所有命令再关这是上面代码的做法比每条命令建一次连接快得多。但要注意如果某条命令执行时间过长连接可能被服务端断开所以命令级timeout要设合理。失败重试我一般做两层连接失败重试 2 次命令执行失败不重试因为多半是命令本身的问题。重试要加退避别立刻重连否则目标机 sshd 压力更大。import time def connect_with_retry(client, host, user, key_path, retries2): for i in range(retries 1): try: client.connect(hostnamehost, usernameuser, key_filenamekey_path, timeout10) return True except Exception: if i retries: time.sleep(2 ** i) # 1s, 2s 退避 else: return False参数说明retries2表示最多试 3 次退避用2**i实现 1 秒、2 秒递增。别把重试次数设太大否则一台坏机器会占用线程很久拖慢整体。对于明确不可达的机器快速失败比反复重试更有价值。4. 巡检报表生成Excel 明细与 Pyecharts 可视化4.1 用 openpyxl 生成结构化 Excel原始输出是一堆文本直接丢进 Excel 没法看。要做的是把每台机器、每个巡检项的结果整理成「一行一台机器、一列一个巡检项」的宽表。用openpyxl写顺便把异常机器标红。from openpyxl import Workbook from openpyxl.styles import PatternFill, Font def build_excel(results, commands, out_pathinspect_report.xlsx): wb Workbook() ws wb.active ws.title 巡检明细 # 表头主机 每个巡检项的描述 headers [主机] [c[desc] for c in commands] [状态] ws.append(headers) for cell in ws[1]: cell.font Font(boldTrue) red PatternFill(start_colorFFC7CE, end_colorFFC7CE, fill_typesolid) for r in results: row [r[host]] for c in commands: row.append(r[items].get(c[key], )) row.append(异常 if r[error] else 正常) ws.append(row) if r[error]: # 整行标红一眼看出问题机器 for cell in ws[ws.max_row]: cell.fill red wb.save(out_path)逻辑说明表头用命令的desc而不是key可读性更好。异常行整行标红汇报时一眼能定位。参数上PatternFill的颜色值用十六进制FFC7CE是常见的浅红。注意单元格内容如果太长比如ps aux输出Excel 会撑得很宽建议在写入前对超长文本做截断比如保留前 200 字符。4.2 Pyecharts 出图把「正常率」和「资源水位」画出来Excel 适合查明细但汇报要看趋势和分布。Pyecharts 生成的是 HTML可以直接嵌到内部看板。最实用的两张图一张是各主机巡检状态饼图一张是磁盘使用率柱状图。from pyecharts.charts import Bar, Pie from pyecharts import options as opts def build_charts(results, out_htmlinspect_chart.html): # 状态分布饼图 ok sum(1 for r in results if not r[error]) bad len(results) - ok pie (Pie() .add(, [(正常, ok), (异常, bad)]) .set_global_opts(title_optsopts.TitleOpts(title巡检状态分布))) # 磁盘使用率柱状图从 df 输出里解析百分比 hosts, usage [], [] for r in results: df_out r[items].get(disk, ) pct 0 for line in df_out.splitlines(): parts line.split() if len(parts) 5 and parts[4].endswith(%): pct max(pct, int(parts[4].rstrip(%))) hosts.append(r[host]) usage.append(pct) bar (Bar() .add_xaxis(hosts) .add_yaxis(最高磁盘使用率(%), usage) .set_global_opts( title_optsopts.TitleOpts(title各主机磁盘水位), yaxis_optsopts.AxisOpts(max_100))) # 两张图渲染到同一个 HTML from pyecharts.charts import Page page Page() page.add(pie, bar) page.render(out_html)逻辑说明磁盘解析那段遍历df -h的每一行取第 5 列使用率用max拿到该机最高分区水位这样一台机器只出一个数。参数上yaxis_opts的max_100固定 Y 轴上限避免个别机器 99% 时柱子比例失真。Pyecharts 默认输出依赖在线 JS内网环境要提前把资源本地化否则图打不开——这是很常见的翻车点。5. 避坑与排查这些坑我替你踩过了5.1 连接超时但 ping 得通现象ping目标机正常但 Paramiko 报TimeoutError或NoValidConnectionsError。原因通常是目标机 sshd 的MaxStartups限制了并发未认证连接数你并发一开大就被拒。解决把max_workers降到 20 以下或者在目标机 sshd 配置里调大MaxStartups。排查时先用单线程跑一遍确认不是网络问题。5.2 命令执行完拿不到输出现象stdout.read()返回空字符串但命令明明有输出。原因多半是没调recv_exit_status()或者命令本身把输出写到了 stderr。解决先读 stdout 再读 stderr最后调recv_exit_status()如果命令用了管道注意 shell 的退出码是管道最后一个命令的必要时加set -o pipefail。5.3 中文乱码现象Excel 里中文变成问号或方块。原因是目标机 locale 不是 UTF-8输出编码是 GBK 之类。解决解码时用errorsignore兜底更稳妥的做法是在命令前加export LANGen_US.UTF-8或者用chardet探测编码后再解码。我一般直接在命令模板里统一加LANGC输出纯 ASCII省心。5.4 线程池跑完进程不退出现象脚本执行完结果都拿到了但进程卡住不结束。原因是 Paramiko 的连接没关干净或者某个exec_command的通道还挂着。解决确保每个client.close()都在finally里执行如果用了Transport还要显式transport.close()。排查时用threading.enumerate()看还有哪些线程活着。5.5 Pyecharts 图表在内网打不开现象生成的 HTML 打开是空白控制台报 JS 加载失败。原因是 Pyecharts 默认从 CDN 拉 echarts.min.js。解决用pyecharts的CurrentConfig.ONLINE_HOST指向本地静态资源目录或者直接把 echarts 的 JS 文件下载到本地渲染时引用本地路径。这一步不做报表在内网就是废的。6. 进阶把巡检做成可调度、可对比的长期资产单次巡检只是起点真正有价值的是把它变成每天自动跑、结果可对比的资产。我的做法是用schedule或系统自带的定时任务每天凌晨低峰期跑一次输出带日期的 Excel 和 HTML存到按日期分目录的归档里。这样一周后你就能做趋势对比——某台机器磁盘是不是在持续涨某台机器负载是不是每天同一时段飙高。对比逻辑很简单读最近两天的 Excel按主机和巡检项做差。下面是个最小实现。from openpyxl import load_workbook def diff_reports(yesterday, today, key_col磁盘使用率): 对比两天报表找出指标变化超过阈值的机器 def read(path): wb load_workbook(path) ws wb.active headers [c.value for c in ws[1]] idx headers.index(key_col) data {} for row in ws.iter_rows(min_row2, values_onlyTrue): data[row[0]] row[idx] return data old, new read(yesterday), read(today) alerts [] for host, val in new.items(): if host in old and old[host] and val: # 简单解析百分比数字做差 try: delta float(str(val).replace(%, )) - float(str(old[host]).replace(%, )) if abs(delta) 10: # 变化超过 10 个百分点就告警 alerts.append((host, old[host], val, delta)) except ValueError: continue return alerts参数说明key_col指定要对比的列abs(delta) 10是告警阈值按你的业务调。这个函数返回的列表可以直接塞进邮件正文或企业微信机器人。注意解析百分比时要做异常兜底因为有些机器的df输出格式可能不一致。再往上走一层可以把巡检结果写进时序库比如用 SQLite 先顶着这样趋势图就不用靠对比 Excel 了直接查库出图。但别一上来就上重型方案先用 Excel 加 HTML 跑通闭环等确实有需求了再演进。我见过太多人一开始就搭一套复杂的采集加存储加告警结果维护成本比手工巡检还高最后不了了之。一个具体技巧把命令清单和主机清单都做成配置文件代码里不出现任何硬编码的 IP 和命令。这样换环境、加机器、改巡检项都不用动代码团队里其他人也能接手。我自己维护的巡检脚本三年里代码几乎没改过改的全是配置——这才是自动化该有的样子。希望帮到你。本文还有配套的精品资源点击获取