SSH和Ping与Python异步并发主机？

Question

我试图同时向主机发送SSH/Ping，但没有看到任何结果，因此，可能我的实现是不正确的。这就是我到目前为止所拥有的。任何想法都很感激。

import paramiko
import time
import asyncio
import subprocess

async def sshTest(ipaddress,deviceUsername,devicePassword,sshPort): #finalDict
    try:
            print("Performing SSH Connection to the device")

            client = paramiko.SSHClient()
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

            client.connect(ipaddress, username=deviceUsername, password=devicePassword, port=sshPort, look_for_keys=False, allow_agent=False)
            print("Channel established")     

    except Exception as e:
        print(e)       

async def pingf(ip):
    p1 = subprocess.Popen(['ping', '-c','5', ip], stdout=subprocess.PIPE)
    output = p1.communicate()[0]
    print(output)

async def main():
    taskA = loop.create_task(sshTest('192.168.255.68','admin','admin','22'))
    taskB = loop.create_task(sshTest('192.168.254.108','admin','admin','22'))
    taskC = loop.create_task(sshTest('192.168.249.134','admin','admin','22'))
    taskD = loop.create_task(sshTest('192.168.254.108','admin','admin','22'))
    task1 = loop.create_task(pingf('192.168.255.68'))
    task2 = loop.create_task(pingf('192.168.254.108'))
    task3 = loop.create_task(pingf('192.168.249.134'))
    task4 = loop.create_task(pingf('192.168.254.108'))
    await asyncio.wait([taskA,taskB,taskC,taskD,task1,task2,task3,task4]) 
    
if __name__ == "__main__":
    start = time.time()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    end = time.time()
    print("The time of execution of above program is :", end-start)

Answer 1

Asyncio是合作多任务的一种形式。这意味着，为了使任务并发运行，任务必须显式地将控制权返回调度程序，这在Python中意味着“您的任务需要在某件事情上进行await”。

您的任务都没有调用await，因此它们不会同时运行。你现在拥有的将是连续运行的。

如果您想同时运行ssh连接，则必须这样做：

• 将paramiko替换为类似于AsyncSSH的东西，后者是为使用asyncio而编写的，或者
• 使用线程或多处理将任务并行化，而不是使用asyncio。

此外，如果使用asyncio，任何涉及运行外部命令(例如pingf任务)的操作都需要使用asyncio的run_in_executor方法。

对于这里展示的示例，我建议使用concurrent.futures模块。您的代码最终可能看起来像这样(我已经修改了代码，使其在我的测试环境中运行，并给了sshTest任务一些不只是简单连接的任务)：

import concurrent.futures

import paramiko
import asyncio
import subprocess


def sshTest(ipaddress, deviceUsername, devicePassword, sshPort):  # finalDict
    try:
        print("Performing SSH Connection to the device")

        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        client.connect(
            ipaddress,
            username=deviceUsername,
            password=devicePassword,
            port=sshPort,
            look_for_keys=True,
            allow_agent=True,
        )
        stdin, stdout, stderr = client.exec_command("sh -c 'sleep 2; uptime'")
        output = stdout.read()
        return output
    except Exception:
        return "failed to connect"


def pingf(ip):
    output = subprocess.check_output(["ping", "-c", "5", ip])
    return output


def main():
    futures = []
    with concurrent.futures.ThreadPoolExecutor() as pool:
        futures.append(pool.submit(sshTest, "localhost", "root", "admin", "2200"))
        futures.append(pool.submit(sshTest, "localhost", "root", "admin", "2201"))
        futures.append(pool.submit(sshTest, "localhost", "root", "admin", "2202"))
        futures.append(pool.submit(pingf, "192.168.1.1"))
        futures.append(pool.submit(pingf, "192.168.1.5"))
        futures.append(pool.submit(pingf, "192.168.1.254"))

    for future in concurrent.futures.as_completed(futures):
        print("return value from task:", future.result())


if __name__ == "__main__":
    main()