Python多线程，生产者-消费者-终结者

threading.Thread

run()

__init__()

__init__()

threading.Thread

__init__()

import threading

class MyThread(threading.Thread):
    def __init__(self, para1, para2):
        threading.Thread.__init__(self)
        self.para1 = para1
        self.para2 = para2

    def run(self):
        print self.para1
        print self.para2

thread1 = MyThread('Hello, earthman.', 'Goodbye.')
thread2 = MyThread('Hello, ET.', 'Bye.')
thread1.start()
thread2.start()

deque

maxlen

threading

Condition()

Condition()

acquire()

release()

wait()

notify()

notifyAll()

wait()

notify()

notifyAll()

wait()

#!/usr/bin/python2
from collections import deque
import threading
import time

BUFFER_SIZE = 5

class State(object):
    """A set of mechanism for synchronization and notification between threads.

    Attributes:
        shutting: A boolean indicating if the system is shutting down. Its 
            value should only be changed by calling the shutdown() method.
    """

    def __init__(self):
        self.shutting = False
        self.__cv = threading.Condition()

    def shutdown(self):
        """Set shutting to True and notify other threads who are waiting."""
        self.shutting = True
        self.__cv.acquire()
        self.__cv.notifyAll()
        self.__cv.release()

    def gotdata(self):
        """Notify other threads who are waiting."""
        if self.__cv.acquire(False):
            # if self.shutting:
                # self.__cv.release()
                # return
            self.__cv.notifyAll()
            self.__cv.release()

    def waitfordata(self):
        """Sleep until another thread calls either gotdata() or shutdown()."""
        if self.__cv.acquire(False):
            if self.shutting:
                self.__cv.release()
                return
            self.__cv.wait()
            self.__cv.release()         

class MyThread(threading.Thread):
    def __init__(self, buffer, state):
        threading.Thread.__init__(self)
        self.buffer = buffer
        self.state = state

    def run(self):
        """This is for consumers. The producer should override this method."""
        self.state.waitfordata()
        while not self.state.shutting:
            self.consume(self.buffer)
            self.state.waitfordata()

    def consume(self, buffer):
        """This method should be overridden for a consumer thread."""
        return

class Producer(MyThread):
    """A thread who produces data periodically and save them into buffer."""
    def run(self):
        data = 0
        while not self.state.shutting:
            time.sleep(0.2)
            data += 1
            self.buffer.append(data)
            self.state.gotdata()

class Consumer1(MyThread):
    """A thread who analyzes the data in buffer."""
    def consume(self, buffer):
        for data in buffer:
            print data,
        print ''

class Consumer2(MyThread):
    """A thread who analyzes the data in buffer."""
    def consume(self, buffer):
        time.sleep(0.1)
        print 'Buffer length: ', len(buffer)

def main():
    thread_classes = (Producer, Consumer1, Consumer2)
    buffer = deque(maxlen = BUFFER_SIZE)
    state = State()
    threads = []
    for cls in thread_classes:
        threads.append(cls(buffer, state))
    for thread in threads:
        thread.start()
    time.sleep(2)
    state.shutdown()
    for thread in threads:
        thread.join()

if __name__ == '__main__':
    main()

deque

maxlen

append()