Python多处理将命令传递到进程/池/队列？

Question

不幸的是，我试图一次与14个raspberry pi进行通信（艺术项目、burning man、don't ask…）并使用Paramiko通过SSH连接到RPi，然后发出其他各种命令：同步文件、启动服务器等等。。。在

为此，我一直在使用python中的多处理模块，但是遇到了错误。在连接到各种rpi之后，我希望python脚本挂起并等待输入，例如：start server（传递服务器名称、位置等），它将通过Paramiko发送ssh命令，在每个rpi上开始运行python脚本。在

我的问题是：如何确保将适当的命令发送到正确的进程/池？例如，如果我实例化连接到各种rpi的类，然后发出start server命令，我希望： 要用名称A初始化的RPi_A上的服务器， 要用名称B初始化的RPi_B上的服务器， 而不是名为B的RPi_，等等。。。在

我需要为此使用process命令吗？或者说游泳池有用吗？如果是这样的话，那么apply\u async，map，map_async。不幸的是，文件有点模糊。在

样本代码：

import sys
import time
import paramiko
import multiprocessing as mp

login = 'pi'
password = 'raspberry'
serverIp = '192.168.1.143'

config = [
    {'hostname': 'pi1.local', 'name': 'carousel'},
    {'hostname': 'pi2.local', 'name': 'bench'}
]



class Dreamlandia():
    def __init__(self):
        pool = mp.Pool(processes=12)
        results = [pool.apply_async(self.connectToServer, args=(dreamlandObject,)) for dreamlandObject in config]
        output = [p.get() for p in results]

    def connectToServer(self, dreamlandObject):
        host = dreamlandObject['hostname']
        structureName = dreamlandObject['name']
        i = 1
        while True:
            try:
                ssh = paramiko.SSHClient()
                ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
                ssh.connect(host, username=login, password=password)
                print ("Connected to " + structureName)
                break
            except paramiko.AuthenticationException:
                print ("Authentication failed when connecting to " + structureName)
                sys.exit(1)
            except:
                print ("Could not SSH to " + structureName + ", waiting for it to start")
                i += 1
                time.sleep(1)

            # If we could not connect within time limit
            if i == 30:
                print ("Could not connect to " + structureName + ". Giving up")
                sys.exit(1)

Answer 1

主流程将负责创建、管理和终止子流程。在

from threading import Thread
from Queue import Queue


def worker(name, queue):
    # connect() #connect to raspPI & stuff ...
    # while True: # in real example this loop ...
        cmd = queue.get()
        print "MSG: thread_%s %s\n" % (name, cmd)
        # execute(cmd) # send command to raspPI
        queue.task_done()

# spawn threads
queues = []
num_threads=4
for i in range(num_threads):
    q = Queue()
    queues.append(q)
    t = Thread(target=worker, args=(i,q))
    t.start()

# send message to every threads
for q in queues:
    q.put("hello", False)

由于我的工人结束了，我没有终止我的子进程。您可能需要为进程添加一个终止条件，否则您将不得不手动终止它。在