我正在使用pythonmultiprocessing派生一些子进程来运行我的工作。有两个要求：
但我得到：
by multiprocessing.Process()具有属性“ pid”以获取其pid。但是我不能添加异步回调，当然我也不能同步等待。multiprocessing.Pool()提供回调接口。但是我无法确定池中哪个进程与我的工作匹配，因为我可能需要根据特定的工作来终止该进程。任务很便宜，这里显示代码：
import random, time
import multiprocessing
import os
class Job(object):
def __init__(self, jobid, jobname, command):
self.jobid, self.jobname, self.command = jobid, jobname, command
def __str__(self):
return "Job <{0:05d}>".format(self.jobid)
def __repr__(self):
return self.__str__()
def _run_job(job):
time.sleep(1)
print "{} done".format(job)
return job, random.choice([True, False]) # the second argument indicates whether job has finished successfully
class Test(object):
def __init__(self):
self._loc = multiprocessing.Lock()
self._process_pool = multiprocessing.Pool()
def submit_job(self, job):
with self._loc:
self._process_pool.apply_async(_run_job, (job,), callback=self.job_done)
print "submitting {} successfully".format(job)
def job_done(self, result):
with self._loc:
# stuffs after job has finished is related to some cleanning work, so it needs the lock of the parent process
job, success = result
if success:
print "{} success".format(job)
else:
print "{} failure".format(job)
j1 = Job(1, "test1", "command1")
j2 = Job(2, "test2", "command2")
t = Test()
t.submit_job(j1)
t.submit_job(j2)
time.sleep(3.1) # wait for all jobs finishing
但是现在我无法获得与每个工作相对应的pid。例如，我需要终止作业<1>，但是我找不到进程池中与该作业<1>相关的进程，因此我可能不想随时终止该作业。
如果我使用multiprocessing.Process其他方法，则可以记录每个进程的pid及其对应的jobid。但是我现在不能添加回调方法。
那么，有没有一种方法既可以获取子进程的pid，又可以添加回调方法？
最后，我找到了一个解决方案：multiprocessing.Event改为使用。
由于multiprocessing.Pool无法告诉我分配了哪个进程，因此无法记录它，因此我可以随时根据作业ID杀死它。
幸运的是，multiprocessing提供了Eventobject作为回调方法的替代方法。回顾回调方法的作用：它提供对子进程的异步响应。子进程完成后，父进程可以检测到它并调用回调方法。因此，核心问题是父进程如何检测子进程是否已完成。这是Event目的。
因此解决方案很简单：将Event对象传递给子进程。子进程完成后，它将设置Event对象。在父进程中，它启动守护程序线程以监视是否设置了事件。如果是这样，它可以调用执行这些回调操作的方法。此外，由于我使用multiprocessing.Process而不是创建了进程multiprocessing.Pool，因此我可以轻松获取其PID，从而可以将其杀死。
解决方案代码：
import time
import multiprocessing
import threading
class Job(object):
def __init__(self, jobid, jobname, command):
self.jobid, self.jobname, self.command = jobid, jobname, command
self.lifetime = 0
def __str__(self):
return "Job <{0:05d}>".format(self.jobid)
def __repr__(self):
return self.__str__()
def _run_job(job, done_event):
time.sleep(1)
print "{} done".format(job)
done_event.set()
class Test(object):
def __init__(self):
self._loc = multiprocessing.Lock()
self._process_pool = {}
t = threading.Thread(target=self.scan_jobs)
t.daemon = True
t.start()
def scan_jobs(self):
while True:
with self._loc:
done_jobid = []
for jobid in self._process_pool:
process, event = self._process_pool[jobid]
if event.is_set():
print "Job<{}> is done in process <{}>".format(jobid, process.pid)
done_jobid.append(jobid)
map(self._process_pool.pop, done_jobid)
time.sleep(1)
def submit_job(self, job):
with self._loc:
done_event = multiprocessing.Event()
new_process = multiprocessing.Process(target=_run_host_job, args=(job, done_event))
new_process.daemon = True
self._process_pool[job.jobid] = (new_process, done_event)
new_process.start()
print "submitting {} successfully".format(job)
j1 = Job(1, "test1", "command1")
j2 = Job(2, "test2", "command2")
t = Test()
t.submit_job(j1)
t.submit_job(j2)
time.sleep(5) # wait for job to finish
本文收集自互联网，转载请注明来源。
如有侵权，请联系[email protected] 删除。
我来说两句