使用Python进程池消费流数据

import multiprocessing as mp  
import time

def worker(msg, out_queue):
    print('Processing %s (MP: %s) ' % (msg, mp.current_process().name))
    time.sleep(0.01)
    out_queue.put('Job finished - %s'%(msg))

def sender(out_queue):
    while True:
        if not out_queue.empty():
            msg = out_queue.get()
            print('Kafka sending %s' % msg)
        else:
            time.sleep(0.01)

def writer(q):
    # Write to the in-queue
    count = 30
    for ii in range(0, count):
        print('Writing %s' % (ii))
        # Write 'count' numbers into the queue
        q.put(ii)
        time.sleep(0.001)

if __name__=='__main__':  
    # initialize queues
    data_manager = mp.Manager()
    # this is where we are going to store input data
    in_queue = data_manager.Queue(10)
    # this where we are gonna push them out
    out_queue = data_manager.Queue(10)

    # initialize a message writer
    # the writer would generate data for data stream
    mp.Process(target=writer, args=(in_queue,)).start()

    # initialize a message sender
    # the sender would get the processed results out
    mp.Process(target=sender, args=(out_queue,)).start()

    # create processes pool to consume the data
    # the process in the pool would take input data and
    # out-queue as arguments
    # the data will get out through the out-queue
    pool = mp.Pool(3)

    # daemonize it
    while True:
        if not in_queue.empty():
            # consume the input data
            msg = in_queue.get()
            pool.apply_async(worker, (msg, out_queue))
        else:
            time.sleep(0.01)