Python多进程中的子进程与父进程通信

Question

我正在写一个Python脚本，这个脚本可以快速处理一个文件，通过把每一行发送给不同的进程来处理。最后，我希望父进程能收到每个子进程的结果，然后对这些结果进行处理。以下是我的代码：

#!/usr/bin/env python
import os
import re

from datetime import datetime
from multiprocessing import Process, JoinableQueue

class LineConsumer(Process):

    def __init__(self, queue):
        self.queue = queue
        self.lines = 0
        super(LineConsumer, self).__init__( )

    def run(self):
        print "My PID is %d" % self.pid

        while True:
            line = self.queue.get( )
            print self.lines
            if ':' in line:
                self.lines += 1
            self.queue.task_done( )

class Parser(object):

    def __init__(self, filename, processes=4):
        self.filename = filename
        self.processes = processes

    def parse(self):
        queue = JoinableQueue(100)
        consumers = [ ]
        parents = [ ]
        for i in range(0, self.processes):
            lc = LineConsumer(queue)
            lc.start( )
            consumers.append(lc)

        starttime = datetime.now( )
        problem = False
        numlines = 0

        with open(self.filename, 'r') as data:
            for line in data:
                numlines += 1

                def checkAlive(p):
                    if not p.is_alive( ):
                        return False
                    return True

                alive = map(checkAlive, consumers)
                if False in alive:
                    problem = True
                    print "A process died!!!"
                    break
                queue.put(line)

        if not problem:
            queue.join( )

        for p in consumers:
            print p.lines( )
            p.terminate( )
            p.join( )

        endtime = datetime.now( )
        timedelta = endtime - starttime 
        lps = numlines / timedelta.total_seconds( )
        print "Processed packets at %f lps" % lps

if __name__ == "__main__":
    import sys

    if len(sys.argv) != 2:
        print "Supply a file to read"
        sys.exit(1)

    parser = Parser(sys.argv[1])
    parser.parse( )

这是结果：

My PID is 11578
My PID is 11579
My PID is 11580
My PID is 11581
0
1
0
2
1
3
2
1
...
555
627
564
556
628
0
0
0
0
Processed packets at 27189.771341 lps

你可以看到，每个子进程都能保存它处理的行数，但当我试图从父进程访问这个行数时，总是得到0。我该怎么把行数发送给父进程呢？

Answer 1

我想问题出在你试图访问一个对象的副本，而这个副本实际上在另一个进程（子进程）中被改变了。你需要使用明确的进程间通信方式，把响应发送回父进程。也就是说，你可以用类似于 JoinableQueue 的东西来传递信息给子进程。

顺便说一下，我看不出你是怎么把数据发送给消费者的。

Answer 2

你可以通过一个结果队列把值传回去。

在 LineConsumer 里：

def __init__(self, queue, result_queue):
   self.result_queue = result_queue
   # ...

def terminate(self):
   self.results_queue.put(self.lines)
   super(LineConsumer, self).terminate()

在 Parser 里：

queue = JoinableQueue(100)
result_queue = Queue()
# ...
  lc = LineConsumer(queue, result_queue)
# ...
for p in consumers:
  p.terminate()
  p.join()

while True:
  try:
    print results.queue.get(False)
  except Queue.Empty: # need to import Queue
    break