In Producer/Consumer pattern, how could I kill the consumer thread?

Question

I will run the consumer in another work thread, the code is as following:

def Consumer(self):
        while True:
            condition.acquire()
            if not queue:
                condition.wait()
            json = queue.pop()
            clients[0].write_message(json)
            condition.notify()
            condition.release()


t = threading.Thread(target=self.Consumer);
t.start()

However, I find that I could not kill this work thread, the thread will be wait() all the time after the job...

I try to send a single from Procedurer to Consumer whenever finish the procedure work, if the consumer receive the single, the work thread should exit(), is it possible to do that ?

be careful, currently you aren't running the consumer task in a thread, you need to drop the `()`. `threading.Thread(target=self.Consumer)` — GP89, Dec 24 '14 at 14:03
http://stackoverflow.com/questions/323972/is-there-any-way-to-kill-a-thread-in-python — GP89, Dec 24 '14 at 14:03
Thank you so much !!! I am just also wondering why consumer is not in work thread ....:) — liuzhidong, Dec 24 '14 at 14:04
Is `queue` an instance of `Queue` standard python lib? Do you use `condition` to synchronizing the queue or to synchronizing clients? — Michele d'Amico, Dec 24 '14 at 14:05
Yes, queue is a Queue instance, and condition is to synchronize the queue. Whenever producer put an item into queue, condition will notify() — liuzhidong, Dec 24 '14 at 14:09
You are not using `queue.Queue()` object but a list... `queue.Queue()` has not `pop()` method. Do you use a list as queue. — Michele d'Amico, Dec 24 '14 at 14:21
Yea `queue` uses a threading condition internally, so you don't need to worry in this case. — GP89, Dec 24 '14 at 14:22

Michele d'Amico · Accepted Answer · 2014-12-24T14:57:53.163

My standard way to notify a consumer thread that should stop its work is send a fake message (I rewrite it to make it runnable):

import threading
condition = threading.Condition()
queue = []
class Client():
    def write_message(self,msg):
        print(msg)

clients=[Client()]

jobdone=object()

def Consumer():
    while True:
        condition.acquire()
        try:
            if not queue:
                condition.wait()
            json = queue.pop()
            if json is jobdone:
                break;
            clients[0].write_message(json)
        finally:
            condition.release()

t = threading.Thread(target=Consumer);
t.start()
import time
time.sleep(2)
condition.acquire()
queue.append(jobdone)
condition.notify()
condition.release()

Anyway consider to use queue.Queue that is standard and make synchronization simple. Here is how my example become:

import threading
import queue
import time
queue = queue.Queue()

class Client():
    def write_message(self,msg):
        print(msg)
clients=[Client()]

jobdone=object()
def Consumer():
    while True:
        json = queue.get()
        if json is jobdone:
            break;
        clients[0].write_message(json)

t = threading.Thread(target=Consumer);
t.start()
queue.put("Hello")
queue.put("Word")
time.sleep(2)
queue.put(jobdone)

t.join()
#You can use also q.join()
print("Job Done")

I like this idea ! just one point... we should release the condition before break :) — liuzhidong, Dec 24 '14 at 14:55
Unfortunatelly this will not reliably work if there are more than one consumer... — socketpair, Jul 10 '16 at 21:04
@socketpair of course it not work if you have more than one consumer. That's just a simple example. If you have more than one consumer you should implement something like register/unregister that send the fake message when no more consumers are registered. IMHO one consumer case is just enough for this question. — Michele d'Amico, Jul 10 '16 at 21:14

In Producer/Consumer pattern, how could I kill the consumer thread?

1 Answers1