在 python 中,这是我的多处理设置。我对 Process 方法进行了子类化,并为它提供了一个队列和一些其他字段,用于酸洗/数据目的。
这种策略大约有 95% 的时间有效,另外 5% 的原因不明,队列只是挂起并且它永远不会完成(通常 4 个核心中的 3 个完成他们的工作,最后一个需要永远,所以我必须杀死工作)。
我知道队列在 python 中的大小是固定的,否则它们会挂起。我的队列只存储一个字符串……处理器的 id,所以不能这样。
这是我的代码停止的确切行:
res = self._recv()
有人有想法吗?正式代码如下。谢谢你。
from multiprocessing import Process, Queue
from multiprocessing import cpu_count as num_cores
import codecs, cPickle
class Processor(Process):
def __init__(self, queue, elements, process_num):
super(Processor, self).__init__()
self.queue = queue
self.elements = elements
self.id = process_num
def job(self):
ddd = []
for l in self.elements:
obj = ... heavy computation ...
dd = {}
dd['data'] = obj.data
dd['meta'] = obj.meta
ddd.append(dd)
cPickle.dump(ddd, codecs.open(
urljoin(TOPDIR, self.id+'.txt'), 'w'))
return self.id
def run(self):
self.queue.put(self.job())
if __name__=='__main__':
processes = []
for i in range(0, num_cores()):
q = Queue()
p = Processor(q, divided_work(), process_num=str(i))
processes.append((p, q))
p.start()
for val in processes:
val[0].join()
key = val[1].get()
storage = urljoin(TOPDIR, key+'.txt')
ddd = cPickle.load(codecs.open(storage , 'r'))
.. unpack ddd process data ...