Python 自动化运维：Fabric、Ansible、Paramiko、Subprocess 完整对比

小飞兽 Python 5 次阅读 2026-07-24

Introduction

自动化运维是 DevOps 的核心技能。本文对比 Fabric（远程执行）、Ansible（配置管理）、Paramiko（SSH Python库）、Subprocess（本地命令）四种方案的适用场景和用法。

---

运维自动化方案对比

| 工具 | 类型 | 适用场景 | 难度 |
|------|------|----------|------|
| Subprocess | 本地命令 | 本地脚本自动化 | ⭐ |
| Paramiko | SSH 库 | Python 内嵌 SSH | ⭐⭐ |
| Fabric | 远程执行 | 应用部署/批量操作 | ⭐⭐ |
| Ansible | 配置管理 | 批量服务器管理 | ⭐⭐⭐ |

Subprocess（本地命令）

import subprocess, shlex

# 运行命令并获取输出
result = subprocess.run(['ls', '-la'], capture_output=True, text=True)
print(result.stdout)
print(result.stderr)

# Shell 命令（注意安全）
result = subprocess.run('find . -name "*.log" | head -5', shell=True, capture_output=True, text=True)

# 超时控制
result = subprocess.run(['sleep', '10'], timeout=5)  # 5秒超时

# Popen 流式输出（长时间任务）
proc = subprocess.Popen(['tail', '-f', '/var/log/syslog'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
for line in proc.stdout:
    print(line.decode(), end='')

Paramiko（SSH Python 库）

pip install paramiko

import paramiko, time

def ssh_command(host, port, user, password, command):
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(host, port=port, username=user, password=password)
    stdin, stdout, stderr = ssh.exec_command(command, get_pty=True)
    output = stdout.read().decode()
    ssh.close()
    return output

# 批量执行
def batch_ssh(hosts, command):
    results = {}
    for host in hosts:
        try:
            results[host] = ssh_command(host, 22, 'ubuntu', 'password', command)
        except Exception as e:
            results[host] = f'Error: {e}'
    return results

Fabric（远程执行）

pip install fabric

from fabric import Connection, task

@task
def deploy(c):
    with c.cd('/var/www/myapp'):
        c.run('git pull origin master')
        c.run('pip install -r requirements.txt')
        c.run('systemctl restart myapp')

@task
def health_check(c):
    result = c.run('curl -s http://localhost/health')
    if 'ok' not in result.stdout:
        c.run('systemctl restart myapp')

# 命令行执行
fab deploy -H server1,server2 -u ubuntu
fab health_check -H server1

Ansible（批量配置管理）

pip install ansible
# 或用 ansible-pull 拉取模式（不需要在管理机装 Ansible）

# playbook.yml
- hosts: webservers
  become: yes
  tasks:
    - name: 安装 Nginx
      apt:
        name: nginx
        state: present
        update_cache: yes

    - name: 复制配置文件
      template:
        src: nginx.conf.j2
        dest: /etc/nginx/nginx.conf
      notify: 重启 Nginx

    - name: 启动 Nginx
      service:
        name: nginx
        state: started
        enabled: yes

  handlers:
    - name: 重启 Nginx
      service:
        name: nginx
        state: restarted

ansible-playbook -i inventory playbook.yml
ansible all -i inventory -m shell -a "df -h"  # 快速批量命令

实用运维脚本

#!/usr/bin/env python3
import subprocess, json, smtplib
from email.mime.text import MIMEText

def check_service(name):
    result = subprocess.run(
        ['systemctl', 'is-active', name],
        capture_output=True, text=True
    )
    return result.stdout.strip() == 'active'

def check_disk():
    result = subprocess.run(
        ['df', '-h', '--output=source,size,used,avail,pcent', '-x tmpfs'],
        capture_output=True, text=True
    )
    return result.stdout

def send_alert(message):
    msg = MIMEText(message)
    msg['Subject'] = '[Alert] 服务器异常'
    # smtp.sendmail(...)

# 定时检查
if not check_service('nginx'):
    send_alert('Nginx 服务已停止')
    
if __name__ == '__main__':
    print(check_disk())

常见问题

Q1: Ansible 需要在被管理机器上装 Agent 吗？不需要，Ansible 通过 SSH 远程管理，默认只需要管理机装 Ansible，被管理机开放 SSH 即可。

Q2: Fabric 和 Paramiko 区别？Fabric 是基于 Paramiko 的高级封装，提供更方便的上下文管理器（cd/put/run）和命令行工具，适合应用部署。

Q3: Subprocess 的 shell=True 有安全风险吗？有，shell=True 会经过 shell 解析，恶意输入可能导致命令注入。尽量用列表形式 [cmd, arg1, arg2] 避免 shell=True。