python 中生產者與消費者模式

阿新 • • 發佈：2018-12-12

生產者與消費者模式

摘要: 最近一段時間,寫了一些生產者與消費者模型, 特此此文作為總結. 文中總結了不同的生產者與消費者的情況. 一個生產者, 多個消費者,與一個生產者,多個消費者的程式設計模式.

一. 生產者與消費者

在軟體開發的過程中，經常碰到這樣的場景：某些模組負責生產資料，這些資料由其他模組來負責處理（此處的模組可能是：函式、執行緒、程序等). 產生資料的模組稱為生產者,而處理資料的模組稱為消費者

二. 生產者與消費者模型的優點:

解耦耦合
併發
閒忙不均

一.多執行緒實現生產者與消費者模式例項

一個生產者與一個消費者

# !/usr/bin/env python3
# -*- coding: UTF-8 -*-
"""
@author: Frank
@contact:  
[email protected]
@file: 18.py
@time: 2018/7/10 下午8:59

一個簡單的demon   生產者與消費者

"""
import time
import queue

import threading
import random


class Producer(threading.Thread):
    """
    只負責產生資料
    """
    FINISHED = True

    def __init__(self, name, queue):
        # python3的寫法
        super().__init__(name=name)
        self.queue = queue

    def run(self):
        for i in range(10):
            print("%s is producing %d to the queue!" % (self.getName(), i))
            self.queue.put(i)
            time.sleep(random.randint(1, 10) * 0.1)

        # 設定完成的標誌位
        self.queue.put(self.FINISHED)
        print("%s finished!" % self.getName())


class Consumer(threading.Thread):
    """
    資料處理,
    寫入到資料庫裡面
    """
    FINISHED = True

    def __init__(self, name, queue):
        super().__init__(name=name)
        self.queue = queue

    def run(self):
        while True:
            value = self.queue.get()
            # 用來退出執行緒
            if value is self.FINISHED:
                break
            print("{} is consuming. {} in the queue is consumed!".format(self.getName(), value))

        print("%s finished!" % self.getName())


if __name__ == '__main__':
    queue = queue.Queue()
    producer = Producer('producer', queue)

    consumer = Consumer('consumer', queue)

    producer.start()
    consumer.start()

    consumer.join()
    producer.join()

    print('All threads  done')

結果如下:

producer is producing 0 to the queue!
consumer is consuming. 0 in the queue is consumed!
producer is producing 1 to the queue!
consumer is consuming. 1 in the queue is consumed!
producer is producing 2 to the queue!
consumer is consuming. 2 in the queue is consumed!
producer is producing 3 to the queue!
consumer is consuming. 3 in the queue is consumed!
producer is producing 4 to the queue!
consumer is consuming. 4 in the queue is consumed!
producer is producing 5 to the queue!
consumer is consuming. 5 in the queue is consumed!
producer is producing 6 to the queue!
consumer is consuming. 6 in the queue is consumed!
producer is producing 7 to the queue!
consumer is consuming. 7 in the queue is consumed!
producer is producing 8 to the queue!
consumer is consuming. 8 in the queue is consumed!
producer is producing 9 to the queue!
consumer is consuming. 9 in the queue is consumed!
producer finished!
consumer finished!
All threads  done

Process finished with exit code 0

FINISHED 通過這個屬性來通知消費者生產者已經完成生產. 相當於一個標誌,告訴消費者 生產者已經生產完成.

但是當有多個消費者就會出現問題. 看下面演示程式碼, 消費者就會阻塞在這裡 .

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
"""
@author: Frank 
@contact: [email protected]
@file: 18.py
@time: 2018/7/10 下午8:59

一個簡單的demon   生產者與消費者

"""
import time
from datetime import datetime, timedelta
import queue

import threading
import random

# 哨兵
_sentinel = object()


class Producer(threading.Thread):
    """
    只負責產生資料
    """

    def __init__(self, name, queue):
        # python3的寫法
        super().__init__(name=name)
        self.queue = queue

    def run(self):
        for i in range(5):
            print("%s is producing %d to the queue!" % (self.getName(), i))
            self.queue.put(i)
            time.sleep(random.randint(1, 10) * 0.1)

        # 設定完成的標誌位
        self.queue.put(_sentinel)
        print("%s finished!" % self.getName())


class Consumer(threading.Thread):
    """
    資料處理,
    寫入到資料庫裡面
    """

    def __init__(self, name, queue):
        super().__init__(name=name)
        self.queue = queue

    def run(self):
        while True:
            # 預設是阻塞的等待.
            value = self.queue.get()
            # 用來退出執行緒
            if value is _sentinel:
                break
            print("{} is consuming. {} in the queue is consumed!".format(self.getName(), value))

        print("%s finished!" % self.getName())


if __name__ == '__main__':
    queue = queue.Queue()
    producer = Producer('producer', queue)

    consumer = Consumer('consumer', queue)
    consumer2 = Consumer('consumer2', queue)

    producer.start()

    consumer.start()
    consumer2.start()

    producer.join()
    consumer.join()
    consumer2.join()

    print('All threads  done')

producer is producing 0 to the queue!
consumer is consuming. 0 in the queue is consumed!
producer is producing 1 to the queue!
consumer is consuming. 1 in the queue is consumed!
producer is producing 2 to the queue!
consumer2 is consuming. 2 in the queue is consumed!
producer is producing 3 to the queue!
consumer is consuming. 3 in the queue is consumed!
producer is producing 4 to the queue!
consumer2 is consuming. 4 in the queue is consumed!
producer finished!
consumer finished!

上面就就阻塞在第二個消費者那裡了, 因為第一個消費者把標誌位拿走啦,並且退出了執行緒. 消費者2 發現佇列空了,就在那裡等待, 所以就這樣阻塞在這裡.

當出現一個生產者和多個消費者的時候.

解決方法,每個執行緒消費完成後, 在把標誌位放回佇列中, 這樣其他的消費者執行緒就可以退出來了. 來看下例子:

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
"""
@author: Frank 
@contact: [email protected]
@file: 18.py
@time: 2018/7/10 下午8:59

一個簡單的demon   生產者與消費者

"""
import time
import queue

import threading
import random

# 哨兵
_sentinel = object()


class Producer(threading.Thread):
    """
    只負責產生資料
    """

    def __init__(self, name, queue):
        # python3的寫法
        super().__init__(name=name)
        self.queue = queue

    def run(self):
        for i in range(15):
            print("%s is producing %d to the queue!" % (self.getName(), i))
            self.queue.put(i)
            time.sleep(random.randint(1, 20) * 0.1)

        # 設定完成的標誌位
        self.queue.put(_sentinel)
        print("%s finished!" % self.getName())


class Consumer(threading.Thread):
    """
    資料處理,對資料進行消費.
    """

    def __init__(self, name, queue):
        super().__init__(name=name)
        self.queue = queue

    def run(self):
        while True:
            value = self.queue.get()
            # 用來退出執行緒
            if value is _sentinel:
                # 新增哨兵,讓其他執行緒有機會退出來
                self.queue.put(value)
                break
            print("{} is consuming. {} in the queue is consumed!".format(self.getName(), value))

        print("%s finished!" % self.getName())


if __name__ == '__main__':
    queue = queue.Queue()
    producer = Producer('producer', queue)

    producer.start()

    consumer_threads = []
    for i in range(5):
        consumer = Consumer('consumer_' + str(i), queue)
        consumer_threads.append(consumer)
        consumer.start()

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

    producer.join()

    print('All threads  done')

結果如下:

producer is producing 0 to the queue!
consumer_0 is consuming. 0 in the queue is consumed!
producer is producing 1 to the queue!
consumer_0 is consuming. 1 in the queue is consumed!
producer is producing 2 to the queue!
consumer_1 is consuming. 2 in the queue is consumed!
producer is producing 3 to the queue!
consumer_2 is consuming. 3 in the queue is consumed!
producer is producing 4 to the queue!
consumer_3 is consuming. 4 in the queue is consumed!
producer is producing 5 to the queue!
consumer_4 is consuming. 5 in the queue is consumed!
producer is producing 6 to the queue!
consumer_0 is consuming. 6 in the queue is consumed!
producer is producing 7 to the queue!
consumer_1 is consuming. 7 in the queue is consumed!
producer is producing 8 to the queue!
consumer_2 is consuming. 8 in the queue is consumed!
producer is producing 9 to the queue!
consumer_3 is consuming. 9 in the queue is consumed!
producer is producing 10 to the queue!
consumer_4 is consuming. 10 in the queue is consumed!
producer is producing 11 to the queue!
consumer_0 is consuming. 11 in the queue is consumed!
producer is producing 12 to the queue!
consumer_1 is consuming. 12 in the queue is consumed!
producer is producing 13 to the queue!
consumer_2 is consuming. 13 in the queue is consumed!
producer is producing 14 to the queue!
consumer_3 is consuming. 14 in the queue is consumed!
producer finished!
consumer_4 finished!
consumer_0 finished!
consumer_1 finished!
consumer_2 finished!
consumer_3 finished!
All threads  done

Process finished with exit code 0

當出現多個生產和多個消費者模型

二. 多個生產者與多個消費者模型

第一種實現方式

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
"""
@author: Frank
@contact: [email protected]
@file: test_producer_consumer.py
@time: 2018/8/31 上午9:55

一個簡單的demon   生產者與消費者模型
多個生產者 與多個消費者

2個生產者 與 多個消費者 模型.

"""

import time
import queue

import threading
import random

from collections import deque

# 哨兵
_sentinel = object()
_sentinel2 = object()


class Producer(threading.Thread):
    """
    只負責產生資料
    """

    def __init__(self, name, queue):
        # python3的寫法
        super().__init__(name=name)
        self._queue = queue

    def run(self):
        for i in range(5):
            print("{} is producing {} to the queue!".format(self.getName(), i))

            self._queue.put(i)
            time.sleep(random.randint(1, 20) * 0.1)

        # 設定完成的標誌位
        self._queue.put(_sentinel)
        print("%s finished!" % self.getName())


class Producer2(Producer):

    def run(self):
        for i in range(65, 70):
            item = chr(i)

            print("{} is producing {} to the queue!".format(self.getName(), item))
            self._queue.put(item)
            time.sleep(random.randint(1, 20) * 0.8)

        # 設定完成的標誌位
        self._queue.put(_sentinel2)
        print("%s finished!" % self.getName())


class Consumer(threading.Thread):
    """
    資料處理
    """

    _deque = deque()

    def __init__(self, name, queue, lock):
        super().__init__(name=name)
        self._queue = queue
        self._lock = lock

    def run(self):
        while True:
            value = self._queue.get(block=True, timeout=10)
            # 用來退出執行緒
            if value in (_sentinel, _sentinel2):
                with self._lock:
                    if value not in Consumer._deque:
                        Consumer._deque.append(value)

                self._queue.put(value)
                if len(Consumer._deque) == 2:
                    print('Consumer._deque  ==2  break')
                    break
            else:
                print("{} is consuming. {} in the queue is consumed!".format(self.getName(), value))

        print("{} finished!".format(self.getName()))


if __name__ == '__main__':

    q = queue.Queue()

    lock = threading.Lock()

    sentienl_queue = queue.Queue()

    producer = Producer('producer111', q)
    producer2 = Producer2('producer222', q)

    producer2.start()
    producer.start()

    consumer_threads = []
    for i in range(5):
        consumer = Consumer('consumer_' + str(i), q, lock)
        consumer_threads.append(consumer)
        consumer.start()

    for consumer in consumer_threads:
        consumer.join()

    producer.join()
    producer2.join()

    print('All threads  done')

結果如下:

producer222 is producing A to the queue!
producer111 is producing 0 to the queue!
consumer_0 is consuming. A in the queue is consumed!
consumer_0 is consuming. 0 in the queue is consumed!
producer111 is producing 1 to the queue!
consumer_0 is consuming. 1 in the queue is consumed!
producer111 is producing 2 to the queue!
consumer_1 is consuming. 2 in the queue is consumed!
producer111 is producing 3 to the queue!
consumer_2 is consuming. 3 in the queue is consumed!
producer111 is producing 4 to the queue!
consumer_3 is consuming. 4 in the queue is consumed!
producer111 finished!
producer222 is producing B to the queue!
consumer_4 is consuming. B in the queue is consumed!
producer222 is producing C to the queue!
consumer_4 is consuming. C in the queue is consumed!
producer222 is producing D to the queue!
consumer_1 is consuming. D in the queue is consumed!
producer222 is producing E to the queue!
consumer_0 is consuming. E in the queue is consumed!
producer222 finished!
Consumer._deque  ==2  break
consumer_1 finished!
Consumer._deque  ==2  break
consumer_2 finished!
Consumer._deque  ==2  break
consumer_3 finished!
Consumer._deque  ==2  break
consumer_4 finished!
Consumer._deque  ==2  break
consumer_0 finished!
All threads  done

Process finished with exit code 0

第二種實現方式

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
"""
@author: Frank
@contact: [email protected]
@file: test_producer_consumer.py
@time: 2018/8/31 上午9:55

一個簡單的demon   生產者與消費者模型
多個生產者 與多個消費者


2個生產者 與 多個消費者 模型.

"""
import time
import queue

import threading
import random
from collections import deque

# 兩個哨兵  作為結束標誌位
_sentinel = object()
_sentinel2 = object()


class Producer(threading.Thread):
    """
    只負責產生資料
    """

    def __init__(self, name, queue):
        # python3的寫法
        super().__init__(name=name)
        self._queue = queue

    def run(self):
        for i in range(5):
            print("{} is producing {} to the queue!".format(self.getName(), i))

            self._queue.put(i)
            time.sleep(0.1)

        # 設定完成的標誌位
        self._queue.put(_sentinel)
        print("%s finished!" % self.getName())


class Producer2(Producer):

    def run(self):
        for i in range(65, 70):
            item = chr(i)

            print("{} is producing {} to the queue!".format(self.getName(), item))
            self._queue.put(item)
            time.sleep(random.randint(1, 20) * 0.8)

        # 設定完成的標誌位
        self._queue.put(_sentinel2)
        print("%s finished!" % self.getName())


class Consumer(threading.Thread):
    """
    資料處理
    """

    _deque = deque()

    def __init__(self, name, queue):
        super().__init__(name=name)
        self._queue = queue

    @staticmethod
    def consume_data(datas):
        """
        消費資料 的方法 ,具體 怎麼消費資料, 看如何實現. 比如寫入檔案, 寫入資料庫等.
        :param datas:
        :return: int
        """

        return 0

    def run(self):
        while True:
            try:
                datas = self._queue.get(block=True, timeout=10)
            except queue.Empty:
                # print(f"queue  is  empty. ")
                datas = None

            # 用來退出執行緒
            if datas in (_sentinel, _sentinel2):
                if datas not in Consumer._deque:
                    print(f'put {datas} into the  Consumer._deque')
                    Consumer._deque.append(datas)
                if len(Consumer._deque) == 2:
                    print('Consumer._deque length == 2 ,break, current_thread:{} has finished'.format(self.getName()))
                    break
            else:
                # 消費資料
                if datas:
                    # 消費資料
                    sucess_number = self.consume_data(datas)
                    print("{} is consuming.  datas: {} in the queue is consumed! sucess_number:{}".format(self.getName(), datas, sucess_number))
                else:
                    print('datas is  None...')
                    if len(Consumer._deque) == 2:
                        print('Consumer._deque length == 2 ,break, current_thread:{} has finished'.format(self.getName()))
                        break


if __name__ == '__main__':

    q = queue.Queue()

    producer = Producer('producer111', q)
    producer2 = Producer2('producer222', q)

    producer.start()
    producer2.start()

    consumer_threads = []
    for i in range(5):
        consumer = Consumer('consumer_' + str(i), q)
        consumer_threads.append(consumer)
        consumer.start()

    for consumer in consumer_threads:
        consumer.join()

    producer2.join()
    producer.join()

    print('All threads  done')

執行結果的一部分如下圖:


producer222 is producing D to the queue!
consumer_0 is consuming.  datas: D in the queue is consumed! sucess_number:0
datas is  None...
datas is  None...
datas is  None...
datas is  None...
datas is  None...
datas is  None...
datas is  None...
datas is  None...
datas is  None...
producer222 is producing E to the queue!
consumer_0 is consuming.  datas: E in the queue is consumed! sucess_number:0
producer222 finished!
put <object object at 0x102720110> into the  Consumer._deque
Consumer._deque length == 2 ,break, current_thread:consumer_1 has finished
datas is  None...
Consumer._deque length == 2 ,break, current_thread:consumer_2 has finished
datas is  None...
Consumer._deque length == 2 ,break, current_thread:consumer_3 has finished
datas is  None...
Consumer._deque length == 2 ,break, current_thread:consumer_4 has finished
datas is  None...
Consumer._deque length == 2 ,break, current_thread:consumer_0 has finished
All threads  done

總結: 本文簡單總結的 Python中消費者與生產者模式這種程式設計模式的寫法,通過多執行緒來實現生產者與消費者模型.

分享快樂,留住感動.2018-09-30 22:46:42 --frank

python 中生產者與消費者模式

生產者與消費者模式摘要: 最近一段時間,寫了一些生產者與消費者模型, 特此此文作為總結. 文中總結了不同的生產者與消費者的情況. 一個生產者, 多個消費者,與一個生產者,多個消費者的程式設計模式. 一. 生產者與消費者在軟體開發的過程中，經常碰

java生產者與消費者模式

食物會有實現條件 3.2 釋放 tac lee trace 前言：生產者和消費者模式是我們在學習多線程中很經典的一個模式，它主要分為生產者和消費者，分別是兩個線程，目錄一：生產者和消費者模式簡介二：生產者和消費者模式的實現聲明：本例來源於java經典著作：《

python：生產者與消費者模型

target 代碼所有 name body style pos 方便 tar 1，生產者與消費者模型的矛盾在於數據供需的不平衡 import time import random from multiprocessing import Queue from multip

Java並發編程(4)--生產者與消費者模式介紹

定義一點 ava empty 釋放 nal cond 場景 inter 一、前言　　這種模式在生活是最常見的，那麽它的場景是什麽樣的呢？下面是我假象的，假設有一個倉庫，倉庫有一個生產者和一個消費者，消費者過來消費的時候會檢測倉庫中是否有庫存，如果沒有了則等待生產，如

Java併發程式設計(4)--生產者與消費者模式介紹

一、前言　　這種模式在生活是最常見的，那麼它的場景是什麼樣的呢？下面是我假象的，假設有一個倉庫，倉庫有一個生產者和一個消費者，消費者過來消費的時候會檢測倉庫中是否有庫存，如果沒有了則等待生產，如果有就先消費直至消費完成；而生產者每天的工作就是先檢測倉庫是否有庫存，如果沒有就開始生產，滿倉了就停止生產等待

程式碼編寫生產者與消費者模式思路

## **目前有個任務是建立大量的資料同時向kafka裡寫入，於是之前開了大量的執行緒建立資料並寫入，發現kafka並不能連線那麼多執行緒，後來就用到生產者與消費者模式，大量的執行緒生產資料放入佇列中，然後只開幾個執行緒從佇列中獲取並寫入kafka.**

Java生產者與消費者模式的簡單寫法

生產者消費者模式是研究多執行緒程式的經典問題之一，它描述是有一塊緩衝區作為快取佇列/倉庫，生產者可以將產品放入佇列，消費者則可以從佇列中取走產品。大多數的後臺服務程式併發控制的基本原理都可以歸納為生產者消費者模式。 1、使用Synchronized（）、wait() 、 n

JAVA生產者與消費者模式 BlockingQueue與ExecutorService例項

首先建立生產者： package ProducerAndConsumer; import java.util.Random; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Ti

學習筆記- 自己寫的關於生產者與消費者模式，還有定時任務的demo。

為了加深對生產者消費者模式的理解，特意寫了這個demo，裡面還包含了一個自己寫的定時任務。程式碼下載地址：http://download.csdn.net/detail/li_yan_fei/9811572 是個maven專案，只用了spring框架。學到的內容有3個

實現生產者與消費者模式

產品 style 等待 pytho ask 緩沖制造目錄 tex 實現生產者與消費者模式目錄生產者與消費者模式實現生產者與消費者模式什麽是生產者消費者模式生產者消費者模式是通過一個容器來解決生產者和消費者的強耦合問題。生產者和消費者彼

java多執行緒--ReentrantLock實現生產者與消費者模式

一.本例實現：一對一交替列印，一.生產者邏輯：每次只允許一個生產者來進行生產操作（生產者之間互斥訪問倉庫），必須等消費者取走資料之後，才能進行下一次的生產二.消費者邏輯：每次只允許一個消費者來進行生產操作（消費者之間互斥訪問倉庫），必須等生產者生產資料之後，才能

生產者與消費者模式（執行緒的同步與互斥）

條件變數條件變數的提出首先要涉及一個概念，就是生產者消費者模型：生產者消費者，是在多執行緒同步的一個問題，兩個固定大小緩衝區的執行緒，在實際執行是會發生問題，生產者是生成資料放入緩衝區，重複過程，消費者在緩衝區取走資料。生產者消費者的模型提出了三種關係，兩種角色，

多執行緒實現生產者與消費者模式

生產者-消費者模式的簡介：在實際的軟體開發過程中，我們將產生資料的模組稱為生產者，處理資料的模組成為消費者。但僅有這兩者還不夠成為一個生產者-消費者模式，還需要有一個緩衝區（一段記憶體區域）作為中介，生產者產生的資料放入緩衝區，消費者從緩衝區讀取資料並處理。

IOS多執行緒使用GCD與訊號量實現生產者與消費者模式

一、原理的簡述在生產者消費者模式當中，首先需要分清在這個模式當中有哪些角色？各角色分別擔任什麼職責與它們之間的關係如何？角色之間是在保證資料的準確性的情況下如何通訊（同步資料）的？假設現在有一個這樣的情形：有兩個人共同訪問一個容量有限的倉庫，這2個人，

九、生產者與消費者模式

# 生產者消費者模式 - 生產者消費者模式是程式設計中非常常見的一種設計模式，被廣泛運用在解耦、訊息佇列等場景。 - - 使用生產者消費者模式通常需要在兩者之間增加一個阻塞佇列作為媒介，有了媒介之後就相當於有了一個緩衝，平衡了兩者的能力。 - 整體如上圖所示，最上面是阻塞佇列，右側的 1 是生產者執行緒

在並發編程中使用生產者和消費者模式能夠解決絕大多數並發問題。該模式通過平衡生產線程和消費線程的工作能力來提高程序的整體處理數據的速度。

容器模板類負責速度線程並發進行並發耦合所有在並發編程中使用生產者和消費者模式能夠解決絕大多數並發問題。該模式通過平衡生產線程和消費線程的工作能力來提高程序的整體處理數據的速度。在並發編程中使用生產者和消費者模式能夠解決絕大多數並發問題。該模式通過平衡

廣播模式下的生產者與消費者fanout模式

生成 ack word 需要 bin 隊列 highlight time host 生產者 #coding=utf-8 import pika import sys connection = pika.BlockingConnection(pika.ConnectionP

使用Python多線程實現生產者與消費者模型

watermark vpd 51cto 實現 this read sleep get DG 1，我所使用到的python版本 2，下面編寫具體的實現過程 import threadingimport time import Queue #首先生成一個隊列q =Queue.

python 生產者和消費者模式

簡單的程式碼理解 python中生產者和消費者模式 #encoding: utf-8 from Queue import Queue import time import threading #生產者消費者模式是執行緒間通訊的一種應用 #在使用資料結構的

一看就懂系列之 php中的生產者and消費者模式

前言在工作中常常聽到某某大牛之間的交談會涉及到，xx消費者啊啥的，到底什麼大牛之間講的是什麼？這篇文章主要解決三個問題： 1.到底什麼是生產者和消費者，以及它們之間的故事 2.它們之間靠什麼交流 3.應用場景正文一、什麼是生產者和消費者，以

python 中生產者與消費者 模式

生產者與消費者 模式

一.多執行緒實現生產者與消費者模式 例項

二. 多個生產者與多個消費者 模型

相關推薦

python 中生產者與消費者模式

生產者與消費者模式

一.多執行緒實現生產者與消費者模式例項

二. 多個生產者與多個消費者模型