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

Question

昨天，我在python中使用多处理处理处理了一个约有2000万行的日志文件。

1. 启动一个名为"生产者"的进程，逐行读取文件并将其放入队列
2. 启动三个名为"consumeri"的进程，从队列中提取一行，并对其进行分析以获得ip
3. 在主函数中，我启动这些进程并使用join（）等待

代码低于

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)

结果很奇怪，使用者进程在工作完成后不会终止，主函数在join（）处被阻塞

使用以下不同的套件和记录进行测试：

• 使用test_get_ip（）而不进行多处理来处理大小日志文件，效果很好
• 将main（）与multiprocessing一起使用来处理大型日志文件，它将在join（）处阻塞。每个get_ip进程都将打印"停止get_ip XXXX"，但不会终止
• 使用main（）处理2000行的较小日志文件，它也能很好地工作。get_ip将终止
• 如果我不将ip存储在get_ip（）中的列表ips中，那么无论日志文件大小，它都可以正常工作

那么，问题出在哪里呢？它在列表中有限制吗？我错过了什么吗？

我的机器环境是：

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地址。现在，在终止之前，它需要序列化所有这些数据并将其传递给Queue，然后main()函数将接收并反序列化所有这些数据。

我的建议是，将IP地址直接放入结果队列中，然后让main()进程获取它们并在它们到来时写入。