首页 > 程序开发 > 软件开发 > 其他 >

python生产者/消费者示例

2017-01-25

python生产者 消费者示例:标准线程多进程,生产者 消费者示例。一个生产者,一个消费者,实现如下:

python生产者/消费者示例:标准线程多进程,生产者/消费者示例。一个生产者,一个消费者,实现如下:

# coding: utf-8
import threading
import time
import Queue


class Consumer(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        self._queue = queue

    def run(self):
        while True:
            msg = self._queue.get()

            if isinstance(msg, str) and msg == 'quit':
                break

            time.sleep(1)
            print 'consumer receive msg: %s' % msg

        print 'consumer finished'


def Producer():
    q = Queue.Queue()
    c = Consumer(q)
    c.start()

    i = 0
    while i < 10:
        print &#39;producer put msg: %s&#39; % i
        q.put(str(i))
        time.sleep(0.5)

        i += 1

    q.put(&#39;quit&#39;)
    c.join()


if __name__ == &#39;__main__&#39;:
    Producer()

为了提高处理速度,我们会许要多个消费者。这是可以把消费者放到一个线程池中进行管理。一个生产者,多个消费者时,实现如下:

# coding: utf-8
import threading
import time
import Queue


class Consumer(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        self._queue = queue

    def run(self):
        while True:
            msg = self._queue.get()

            if isinstance(msg, str) and msg == &#39;quit&#39;:
                break

            time.sleep(1)
            print &#39;consumer receive msg: %s&#39; % msg

        print &#39;consumer finished&#39;


def build_consumer_pool(size, queue):
    consumers = []

    for i in range(size):
        c = Consumer(queue=queue)
        c.start()
        consumers.append(c)

    return consumers


def Producer():
    q = Queue.Queue()
    consumers = build_consumer_pool(3, q)

    i = 0
    while i < 12:
        print &#39;producer put msg: %s&#39; % i
        q.put(str(i))

        i += 1

    for c in consumers:
        q.put(&#39;quit&#39;)

    for c in consumers:
        c.join()


if __name__ == &#39;__main__&#39;:
    Producer()

上面的代码都是些基础功能,而且很容易出错。Python提供了更加优雅的线程池实现,multiprocessing模块中的Pool非常好用。

# coding: utf-8
from multiprocessing.dummy import Pool as ThreadPool
import time


def consumer(msg):
    print &#39;consumer receive msg: %s&#39; % msg
    time.sleep(1)
    return msg


def producer():
    items = []
    pool = ThreadPool(4)

    i = 0
    while i < 12:
        print &#39;producer put msg: %s&#39; % i
        items.append(str(i))
        i += 1

    results = pool.map(consumer, items)
    pool.close()
    pool.join()
    print results


if __name__ == &#39;__main__&#39;:
    producer()

这里map方法极大简化了对多线程的处理,代码也更优美、可靠。

相关文章
最新文章
热点推荐