为什么进程不会在多进程中终止

Question

昨天我在python中使用多重处理处理了大约2000万行的日志文件。

1. 启动名为“producer”的进程，逐行读取文件并将其放入队列

代码在下面

from multiprocessing import Process, Queue
from Queue import Empty
import os
import time    


def put_ip(src, q, number):
    """ 
    read file line by line, and put it to queue
    """
    print "start put_ip: %d" % os.getpid()
    with open(src) as f:
        for line in f:
            q.put(line)
        for i in range(number):
            q.put(EOFError)
    print "stop put_ip"


def get_ip(lock, src, result, index):
    """ 
    fetch line, and extract ip from it
    """
    print "start get_ip %d: %d" % (index, os.getpid())
    ips = []
    while True:
        line = src.get()
        if line == EOFError:
            print "%d get EOFError" % index
            break
        else:
            res = json.loads(line.strip())
            # process res, get ip
            ips.append(ip)
    print "get_ip %d get %d ips" % (os.getpid(), len(ips))
    result.put('\n'.join(ips))
    ips = []
    print "stop get_ip %d" % os.getpid()
    return


def test_get_ip(src, dest, number):
    """ 
    test with single process
    """
    srcq = Queue()
    result = Queue()

    with open(src) as f:
        for line in f:
            # if 'error' not in line:
            srcq.put(line)
        for i in range(number):
            srcq.put(EOFError)

    get_ip(srcq, result, 0)


def main(src, dest, number):
    """ 
    with multiprocess
    """
    srcq = Queue()
    result = Queue()

    producer = Process(target=put_ip, args=(src, srcq, number))

    consumers = [Process(target=get_ip, args=(srcq, result, i)) for i in xrange(number)]

    print 'start at %s' % time.asctime()
    starttime = time.time()

    producer.start()
    for consumer in consumers:
        consumer.start()

    producer.join()
    for consumer in consumers:
        consumer.join()

    with open(dest, 'w') as w:
        while True:
            try:
                res = result.get_nowait()
                w.write(res +'\n')
            except Empty:
                print 'Empty'
                break

    print "time: %f" % (time.time()-starttime)

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('-i', dest='src', required=True)
    parser.add_argument('-o', dest='dest', required=True)
    parser.add_argument('-n', dest='number', type=int, default=2)

    args = parser.parse_args()

    main(args.src, args.dest, args.number)
    # test_get_ip(args.src, args.dest, args.number)

结果很奇怪，工作完成后，消费者进程不会终止，并且主函数在连接（）处被阻塞。

使用以下不同的套装和代码进行测试：

• 使用test_get_ip（）而不进行多处理来处理大小日志文件，效果很好

那么，有什么问题？列表中有限制吗？有什么我错过的吗？

我的机器环境是：

Linux 3.19.0-32-generic #37~14.04.1-Ubuntu SMP Thu Oct 22 09:41:40 UTC 2015 x86_64 x86_64 x86_64 GNU/Linux 
Python 2.7.6 (default, Jun 22 2015, 17:58:13) 
[GCC 4.8.2] on linux2

谢谢你抽出时间！

Answer 1

我认为你的问题是，你在缓冲所有的数据，只有当消费者完成时才发送结果。

假设运行get_ip（）的进程在一个列表中收集了1M个ip地址。现在，在终止之前，它需要序列化所有这些数据并将其通过队列，然后main（）函数将接收并反序列化所有这些数据。

我的建议是，您直接将IP地址放入结果队列，然后让main（）进程获取它们并在它们出现时写入它们。