前一篇文章：https://blog.csdn.net/zhang_jiamin/article/details/149199519?spm=1011.2415.3001.5331
介绍了在一台服务器上并发启动master，另一台服务器上并发启动viewer。

这次介绍一下使用python实现分布式在多台服务器上并发启动master和viewer，用以实现更高的并发和负载。

远程 SSH 连接 EC2的python工具

Paramiko

安装

pip install paramiko

使用

import paramiko

hosts = ['70.xxx.250.xxx', '55.xx.222.xx']

for host in hosts:
    with paramiko.SSHClient() as ssh:
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(hostname=host, username='ec2-user', key_filename='qa.pem')
        ssh.exec_command("touch 222.txt")

注：访问EC2的证书放在脚本的同目录。

Fabric

安装

pip install fabric

使用

from fabric import Connection

ec2_hosts = ['70.xxx.250.xxx', '55.xx.222.xx']


def connect_ec2(host):
    # 登录EC2
    conn = Connection(host=host, user='ec2-user', connect_kwargs={"key_filename": "qa.pem"})

    # 执行shell脚本
    conn.run("cd kvs-webrtc-sdk/ && bash multi_master_v2.sh")

• cd kvs-webrtc-sdk/ && bash multi_master_v2.sh 是在 同一个 shell session 中连续执行的
• 如果写成两条 conn.run()，每条命令会启动一个新的 shell，因此 cd 不会“保留路径”；
• 可以执行任意复杂命令，比如：

conn.run("cd /data/scripts && chmod +x *.sh && ./deploy.sh")

Paramiko 与 Fabric 对比

工具	本质	底层依赖
Paramiko	一个低层级的 SSH 库	原生实现 SSH 协议
Fabric	一个高级 SSH 操作库（自动化部署工具）	使用了 Paramiko

Fabric 是基于 Paramiko 封装的更易用工具，提供了更简洁的 API 和自动化功能，适合执行多台服务器上的批量操作。

对比项	Paramiko	Fabric
使用难度	较低层级，需要手动管理连接、执行命令等	较高层级，命令式风格更接近 Shell 脚本
灵活性	高，可以精细控制每一步	中，适合批量任务但不适合复杂交互
并发执行	需要自己用 threading 或 multiprocessing	Fabric 2.x+ 原生支持并发执行（使用 Group 等）
自动化批处理	要自己写逻辑	很适合，比如部署、更新、批量执行命令
安装体积	小	大（因为依赖更多，如 invoke）
学习成本	稍高（偏底层）	较低，语法接近 Shell，适合部署场景

需求类型	推荐工具	理由
只需连接一台机器，执行简单命令	Paramiko	更轻量、无额外依赖
批量连接多台服务器	Fabric	API 更高级，支持并发、结果收集、任务复用等
自定义复杂 SSH 协议行为	Paramiko	Fabric 无法精细操作底层，如端口转发、SFTP 文件校验等
熟悉 Shell + Python	Fabric	类似写 Shell 自动化脚本，集成方便

结论

基于以上分析，我选择的是用fabric。

python多线程工具选择

并发（Concurrency）指的是程序在同一个时间段内处理多个任务。

在 Python 中，实现并发的主要方式有：

并发模型	核心模块	适合场景
线程 Threading	threading / concurrent.futures.ThreadPoolExecutor	IO 密集型任务，如网络请求、文件读写
进程 Multiprocessing	multiprocessing / ProcessPoolExecutor	CPU 密集型任务，如图像处理、加密计算
协程 Coroutine	asyncio, trio, curio 等	高并发、轻量级协程任务，如高并发服务器、爬虫

并发 vs 并行（多核）
并发：多个任务看起来在同时进行，实际是在一个线程中快速切换（适合 IO 密集）
并行：多个任务真正同时在多个 CPU 核心上执行（适合 CPU 密集）
在 Python 中，由于 GIL（全局解释器锁） 的存在，线程不能真正并行跑 Python 字节码，但对于 IO 密集型任务非常有效。

适合使用 ThreadPoolExecutor 的场景

• 网络请求（如批量请求 API、ping 多个 EC2）
• 文件操作（如并发读写多个日志）
• 数据库访问（如查询多个数据源）

如果要跑大量 CPU 密集任务（如图像压缩、加密运算），应改用 ProcessPoolExecutor：

from concurrent.futures import ProcessPoolExecutor

我选择的是多线程并发。

最后代码

from fabric import Connection
from concurrent.futures import ThreadPoolExecutor

ec2_hosts = ['70.xxx.250.xxx', '55.xx.222.xx']


def connect_ec2(host):
    # 登录EC2
    conn = Connection(host=host, user='ec2-user', connect_kwargs={"key_filename": "qa.pem"})

    # 执行shell脚本
    conn.run("cd kvs-webrtc-sdk/ && bash multi_master_v2.sh")


with ThreadPoolExecutor() as pool:
    futures = [pool.submit(connect_ec2, host) for host in ec2_hosts]
    for future in futures:
        future.result()

• 如果机器很多（10+），可以设置线程池最大并发数max_workers：

with ThreadPoolExecutor(max_workers=10) as pool:

表示最多允许10个线程并发。如果没传 max_workers，Python 会默认用 min(32, os.cpu_count() + 4)。

• ThreadPoolExecutor()

这是 Python 提供的线程池执行器。它管理一个线程池来运行函数。
等价于手动用 threading.Thread(target=func).start() 启动多个线程，但它更安全、结构更清晰、具备异常处理能力。

• pool.submit(connect_ec2, host)

submit 表示提交任务到线程池异步执行：

future = pool.submit(func, *args, **kwargs)

它会立即返回一个 Future 对象，任务会在后台某个线程中执行。

所以：

futures = [pool.submit(connect_ec2, host) for host in ec2_hosts]

表示：对 ec2_hosts 中每个主机，启动一个线程去运行 connect_ec2(host)。

• future.result()

future.result() 表示阻塞等待某个任务执行完，并返回执行结果。

for future in futures:
    future.result()

1、表示要等所有线程都跑完，确保全部执行完成后再往下执行。
2、这段代码是“并发执行任务，顺序获取结果”，它不是串行运行函数，而是：
（1）任务已经同时提交、并发执行（线程池里跑着）
（2）for future in futures: 只是按顺序取回每个任务的结果
（3）取结果的时候，如果该任务还没执行完，就会阻塞等它完成
（4）实际任务早就并发地在跑了。
3、这个等待是必要的，因为线程执行是异步的，如果不等，有可能主线程就结束了。

总结

元素	含义
ThreadPoolExecutor	线程池执行器，用于 IO 并发任务
submit(func, arg)	提交任务，返回 Future
future.result()	阻塞等待任务完成，返回执行结果
max_workers=N	控制最多并发线程数量

在pycharm中执行结果如下：

/Users/testmanzhang/PycharmProjects/multi_process_launch_viewer.py 
🔄 当前并发数：1
🔄 当前并发数：2
🔄 当前并发数：3
🔄 当前并发数：4
🔄 当前并发数：5
>>> 正在处理 Channel: xxx, SN: xxx
>>> 正在处理 Channel: xxx, SN: xxx
🔄 当前并发数：1
>>> 正在处理 Channel: xxx, SN: xxx
🔄 当前并发数：2
>>> 正在处理 Channel: xxx, SN: xxx
🔄 当前并发数：3
🔄 当前并发数：4
🔄 当前并发数：5
>>> 正在处理 Channel: xxx, SN: xxx
>>> 正在处理 Channel: xxx, SN: xxx
>>> 正在处理 Channel: xxx, SN: xxx
>>> 正在处理 Channel: xxx, SN: xxx
>>> 正在处理 Channel: xxx, SN: xxx
>>> 正在处理 Channel: xxx, SN: xxx
...

这些打印实际上是服务器上面shell脚本的执行打印，其实是通过fabric同步到了当前代码执行的IDE中。