在Python中分析来自连续输入流(记录)的数据,多处理?

简单的生产者/消费者实施。

虽然来回移动数据确实会导致开销并增加内存使用,但只要多个进程不需要相同的数据,那么开销是最小的。试试看:)可以通过更改队列和池大小数字来调整内存占用。

线程化是减少内存使用的另一种选择,但代价是在GIL上被阻塞,如果处理采用python字节码,则实际上是单线程的。

import multiprocessing
# Some fixed size to avoid run away memory use
recorded_data = multiprocessing.Queue(100)

def process(recorded_data):
    while True:
        data = recorded_data.get()
        <process data>

def record(recorded_data):
    for data in input_stream:
        recorded_data.put(data)

producer = multiprocessing.Process(target=record, 
                                   args=(recorded_data,))
producer.start()

# Pool of 10 processes
num_proc = 10
consumer_pool = multiprocessing.Pool(num_proc)
results = []
for _ in xrange(num_proc):
    results.append(
      consumer_pool.apply_async(process,
                                args=(recorded_data,)))

producer.join()

# If processing actually returns something
for result in results:
    print result
# Consumers wait for data from queue forever
#  so terminate them when done
consumer_pool.terminate()