{"action":"closed","issue":{"url":"https://api.github.com/repos/cundi/blog/issues/28","repository_url":"https://api.github.com/repos/cundi/blog","labels_url":"https://api.github.com/repos/cundi/blog/issues/28/labels{/name}","comments_url":"https://api.github.com/repos/cundi/blog/issues/28/comments","events_url":"https://api.github.com/repos/cundi/blog/issues/28/events","html_url":"https://github.com/cundi/blog/issues/28","id":133415552,"number":28,"title":"Python中的生产与消费问题［翻译］","user":{"login":"cundi","id":10941075,"avatar_url":"https://avatars.githubusercontent.com/u/10941075?v=3","gravatar_id":"","url":"https://api.github.com/users/cundi","html_url":"https://github.com/cundi","followers_url":"https://api.github.com/users/cundi/followers","following_url":"https://api.github.com/users/cundi/following{/other_user}","gists_url":"https://api.github.com/users/cundi/gists{/gist_id}","starred_url":"https://api.github.com/users/cundi/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cundi/subscriptions","organizations_url":"https://api.github.com/users/cundi/orgs","repos_url":"https://api.github.com/users/cundi/repos","events_url":"https://api.github.com/users/cundi/events{/privacy}","received_events_url":"https://api.github.com/users/cundi/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"milestone":null,"comments":0,"created_at":"2016-02-13T08:56:40Z","updated_at":"2016-02-15T07:50:57Z","closed_at":"2016-02-15T07:50:57Z","body":"原始链接：\r\n\r\n# Producer-consumer problem in Python\r\n`By : Akshar Raaj`\r\n\r\nWe will solve Producer Consumer problem in Python using Python threads. This problem is nowhere as hard as they make it sound in colleges.  \r\n\r\nThis blog will make more sense if you have some idea about [Producer Consumer problem](http://en.wikipedia.org/wiki/Producer–consumer_problem).  \r\n\r\nWhy care about Producer Consumer problem:  \r\n\r\n- Will help you understand more about concurrency and different concepts of concurrency.\r\n- The concept of Producer Consumer problem is used to some extent in implementing a message queue. - And you will surely need message queue at some point of time.\r\n\r\nWhile we use threads, you will learn about the following thread topics:  \r\n\r\n- Condition in threads.\r\n- wait() method available on Condition instances.\r\n- notify() method available on Condition instances.\r\n\r\nI will assume you are comfortable with basics of Threads, race condition and how to prevent race condition i.e using locks. If not, my last post on basics of Threads should be able to help.  \r\n\r\nQuoting Wikipedia:  \r\n\r\n```\r\nThe producer's job is to generate a piece of data, put it into the buffer and start again.\r\nAt the same time, the consumer is consuming the data (i.e., removing it from the buffer) one piece at a time\r\n```\r\n\r\nThe catch here is \"At the same time\". So, producer and consumer need to run concurrently. Hence we need separate threads for Producer and Consumer.  \r\n\r\n```python\r\nfrom threading import Thread\r\n\r\nclass ProducerThread(Thread):\r\n    def run(self):\r\n        pass\r\n\r\nclass ConsumerThread(Thread):\r\n    def run(self):\r\n        pass\r\n```\r\n\r\nQuoting Wikipedia again:  \r\n\r\n```\r\nThe problem describes two processes, the producer and the consumer, who share a common,\r\nfixed-size buffer used as a queue.\r\n```\r\n\r\nSo we keep one variable which will be global and will be modified by both Producer and Consumer threads. Producer produces data and adds it to the queue. Consumer consumes data from the queue i.e removes it from the queue.  \r\n\r\n```python\r\nqueue = []\r\n```\r\n\r\nIn first iteration, we will not put fixed-size constraint on queue. We will make it fixed-size once our basic program works.  \r\n\r\nInitial buggy program:  \r\n\r\n```python\r\nfrom threading import Thread, Lock\r\nimport time\r\nimport random\r\n\r\nqueue = []\r\nlock = Lock()\r\n\r\nclass ProducerThread(Thread):\r\n    def run(self):\r\n        nums = range(5) #Will create the list [0, 1, 2, 3, 4]\r\n        global queue\r\n        while True:\r\n            num = random.choice(nums) #Selects a random number from list [0, 1, 2, 3, 4]\r\n            lock.acquire()\r\n            queue.append(num)\r\n            print \"Produced\", num \r\n            lock.release()\r\n            time.sleep(random.random())\r\n\r\n\r\nclass ConsumerThread(Thread):\r\n    def run(self):\r\n        global queue\r\n        while True:\r\n            lock.acquire()\r\n            if not queue:\r\n                print \"Nothing in queue, but consumer will try to consume\"\r\n            num = queue.pop(0)\r\n            print \"Consumed\", num \r\n            lock.release()\r\n            time.sleep(random.random())\r\n\r\n\r\nProducerThread().start()\r\nConsumerThread().start()\r\n```\r\n\r\nRun it few times and notice the result. Your program might not end after raising IndexError. Use Ctrl+Z to terminate.  \r\n\r\nSample output:  \r\n\r\n```bash\r\nProduced 3\r\nConsumed 3\r\nProduced 4\r\nConsumed 4\r\nProduced 1\r\nConsumed 1\r\nNothing in queue, but consumer will try to consume\r\nException in thread Thread-2:\r\nTraceback (most recent call last):\r\n  File \"/usr/lib/python2.7/threading.py\", line 551, in __bootstrap_inner\r\n    self.run()\r\n  File \"producer_consumer.py\", line 31, in run\r\n    num = queue.pop(0)\r\nIndexError: pop from empty list\r\n```\r\n\r\nExplanation:  \r\n\r\n- We started one producer thread(hereafter referred as producer) and one consumer thread(hereafter referred as consumer).\r\n- Producer keeps on adding to the queue and consumer keeps on removing from the queue.\r\n- Since queue is a shared variable, we keep it inside lock to avoid race condition.\r\n- At some point, consumer has consumed everything and producer is still sleeping. Consumer tries to consume more but since queue is empty, an IndexError is raised.\r\n- But on every execution, before IndexError is raised you will see the print statement telling \"Nothing in queue, but consumer will try to consume\", which explains why you are getting the error.\r\n\r\nWe found this implementaion as the wrong behaviour.  \r\n\r\n### What is the correct behaviour?\r\nWhen there was nothing in the queue, consumer should have stopped running and waited instead of trying to consume from the queue. And once producer adds something to the queue, there should be a way for it to notify the consumer telling it has added something to queue. So, consumer can again consume from the queue. And thus IndexError will never be raised.\r\n\r\n## About Condition\r\n- Condition object allows one or more threads to wait until notified by another thread. Taken from here.  \r\n\r\nAnd this is exactly what we want. We want consumer to wait when the queue is empty and resume only when it gets notified by the producer. Producer should notify only after it adds something to the queue. So after notification from producer, we can be sure that queue is not empty and hence no error can crop if consumer consumes.  \r\n\r\n- Condition is always associated with a lock.\r\n- A condition has acquire() and release() methods that call the corresponding methods of the associated lock.\r\n\r\nCondition provides acquire() and release() which calls lock's acquire() and release() internally, and so we can replace lock instances with condition instances and our lock behaviour will keep working properly.  \r\n\r\nConsumer needs to wait using a condition instance and producer needs to notify the consumer using the condition instance too. So, they must use the same condition instance for the wait and notify functionality to work properly.  \r\n\r\nLet's rewrite our Consumer and Producer code:  \r\n\r\n```python\r\nfrom threading import Condition\r\n\r\ncondition = Condition()\r\n\r\nclass ConsumerThread(Thread):\r\n    def run(self):\r\n        global queue\r\n        while True:\r\n            condition.acquire()\r\n            if not queue:\r\n                print \"Nothing in queue, consumer is waiting\"\r\n                condition.wait()\r\n                print \"Producer added something to queue and notified the consumer\"\r\n            num = queue.pop(0)\r\n            print \"Consumed\", num \r\n            condition.release()\r\n            time.sleep(random.random())\r\n```\r\n\r\nLet's rewrite Producer code:  \r\n\r\n```python\r\nclass ProducerThread(Thread):\r\n    def run(self):\r\n        nums = range(5)\r\n        global queue\r\n        while True:\r\n            condition.acquire()\r\n            num = random.choice(nums)\r\n            queue.append(num)\r\n            print \"Produced\", num \r\n            condition.notify()\r\n            condition.release()\r\n            time.sleep(random.random())\r\n```\r\n\r\nSample output:  \r\n\r\n```bash\r\nProduced 3\r\nConsumed 3\r\nProduced 1\r\nConsumed 1\r\nProduced 4\r\nConsumed 4\r\nProduced 3\r\nConsumed 3\r\nNothing in queue, consumer is waiting\r\nProduced 2\r\nProducer added something to queue and notified the consumer\r\nConsumed 2\r\nNothing in queue, consumer is waiting\r\nProduced 2\r\nProducer added something to queue and notified the consumer\r\nConsumed 2\r\nNothing in queue, consumer is waiting\r\nProduced 3\r\nProducer added something to queue and notified the consumer\r\nConsumed 3\r\nProduced 4\r\nConsumed 4\r\nProduced 1\r\nConsumed 1\r\n```\r\n\r\nExplanation:  \r\n\r\n- For consumer, we check if the queue is empty before consuming.\r\n- If yes then call wait() on condition instance.\r\n- wait() blocks the consumer and also releases the lock associated with the condition. This lock was held by consumer, so basically consumer loses hold of the lock.\r\n- Now unless consumer is notified, it will not run.\r\n- Producer can acquire the lock because lock was released by consumer.\r\n- Producer puts data in queue and calls notify() on the condition instance.\r\n- Once notify() call is made on condition, consumer wakes up. But waking up doesn't mean it starts executing.\r\n- notify() does not release the lock. Even after notify(), lock is still held by producer.\r\n- Producer explicitly releases the lock by using condition.release().\r\n- And consumer starts running again. Now it will find data in queue and no IndexError will be raised.\r\n\r\n## Adding a max size on the queue\r\nProducer should not put data in the queue if the queue is full.  \r\n \r\nIt can be accomplished in the following way:  \r\n\r\n- Before putting data in queue, producer should check if the queue is full.\r\n- If not, producer can continue as usual.\r\n- If the queue is full, producer must wait. So call wait() on condition instance to accomplish this.\r\n- This gives a chance to consumer to run. Consumer will consume data from queue which will create space in queue.\r\n- And then consumer should notify the producer.\r\n- Once consumer releases the lock, producer can acquire the lock and can add data to queue.\r\n\r\nFinal program looks like:  \r\n\r\n```python\r\nfrom threading import Thread, Condition\r\nimport time\r\nimport random\r\n\r\nqueue = []\r\nMAX_NUM = 10\r\ncondition = Condition()\r\n\r\nclass ProducerThread(Thread):\r\n    def run(self):\r\n        nums = range(5)\r\n        global queue\r\n        while True:\r\n            condition.acquire()\r\n            if len(queue) == MAX_NUM:\r\n                print \"Queue full, producer is waiting\"\r\n                condition.wait()\r\n                print \"Space in queue, Consumer notified the producer\"\r\n            num = random.choice(nums)\r\n            queue.append(num)\r\n            print \"Produced\", num\r\n            condition.notify()\r\n            condition.release()\r\n            time.sleep(random.random())\r\n\r\n\r\nclass ConsumerThread(Thread):\r\n    def run(self):\r\n        global queue\r\n        while True:\r\n            condition.acquire()\r\n            if not queue:\r\n                print \"Nothing in queue, consumer is waiting\"\r\n                condition.wait()\r\n                print \"Producer added something to queue and notified the consumer\"\r\n            num = queue.pop(0)\r\n            print \"Consumed\", num\r\n            condition.notify()\r\n            condition.release()\r\n            time.sleep(random.random())\r\n\r\n\r\nProducerThread().start()\r\nConsumerThread().start()\r\n```\r\n\r\nSample output:  \r\n\r\n```bash\r\nProduced 0\r\nConsumed 0\r\nProduced 0\r\nProduced 4\r\nConsumed 0\r\nConsumed 4\r\nNothing in queue, consumer is waiting\r\nProduced 4\r\nProducer added something to queue and notified the consumer\r\nConsumed 4\r\nProduced 3\r\nProduced 2\r\nConsumed 3\r\n```\r\n\r\n## Update:\r\nMany people on the internet suggested that I use Queue.Queue instead of using a list with conditions and lock. I agree, but I wanted to show how Conditions, wait() and notify() work so I took this approach.  \r\n\r\nLet's update our code to use Queue.  \r\n\r\nQueue encapsulates the behaviour of Condition, wait(), notify(), acquire() etc.  \r\n\r\nNow is a good time to read the [documentation for Queue](http://docs.python.org/2/library/queue.html) and the source code for it.  \r\n\r\nUpdated program:  \r\n\r\n```python\r\nfrom threading import Thread\r\nimport time\r\nimport random\r\nfrom Queue import Queue\r\n\r\nqueue = Queue(10)\r\n\r\nclass ProducerThread(Thread):\r\n    def run(self):\r\n        nums = range(5)\r\n        global queue\r\n        while True:\r\n            num = random.choice(nums)\r\n            queue.put(num)\r\n            print \"Produced\", num\r\n            time.sleep(random.random())\r\n\r\n\r\nclass ConsumerThread(Thread):\r\n    def run(self):\r\n        global queue\r\n        while True:\r\n            num = queue.get()\r\n            queue.task_done()\r\n            print \"Consumed\", num\r\n            time.sleep(random.random())\r\n\r\n\r\nProducerThread().start()\r\nConsumerThread().start()\r\n```\r\n\r\n#### Explanation\r\n- In place of list, we are using a Queue instance(hereafter queue).\r\n- queue has a Condition and that condition has its lock. You don't need to bother about Condition and Lock if you use Queue.\r\n- Producer uses put available on queue to insert data in the queue.\r\n- put() has the logic to acquire the lock before inserting data in queue.\r\n- Also put() checks whether the queue is full. If yes, then it calls wait() internally and so producer starts waiting.\r\n- Consumer uses get.\r\n- get() acquires the lock before removing data from queue.\r\n- get() checks if the queue is empty. If yes, it puts consumer in waiting state.\r\n- get() and put() has proper logic for notify() too. Why don't you check the source code for Queue now?"}}