也许这会有帮助：import threading

class MyQueue:

def __init__(self):

self.tasks = []

self.tlock = threading.Semaphore(0)

self.dlock = threading.Lock()

self.aborted = False

def put(self, arg):

try:

self.dlock.acquire()

self.tasks.append(arg)

finally:

self.dlock.release()

self.tlock.release()

def get(self):

if self.aborted:

return None

self.tlock.acquire()

if self.aborted:

self.tlock.release()

return None

try:

self.dlock.acquire()

if self.tasks:

return self.tasks.pop()

else: # executed abort

return None

finally:

self.dlock.release()

def abort(self):

self.aborted = True

self.tlock.release()

# TESTING

mq = MyQueue()

import sys

def tlog(line):

sys.stdout.write("[ %s ] %s\n" % (threading.currentThread().name, line))

sys.stdout.flush()

def reader():

arg = 1

while arg is not None:

tlog("start reading")

arg = mq.get()

tlog("read: %s" % arg)

tlog("END")

import time, random

def writer():

try:

pos = 1

while not mq.aborted:

x = random.random() * 5

tlog("writer sleep (%s)" % x)

pending = x

while pending > 0:

tosleep = min(0.5, pending)

if mq.aborted:

return

time.sleep(tosleep)

pending -= tosleep

tlog("write: %s" % x)

mq.put("POS %s val=%s" % (pos, x))

pos += 1

finally:

tlog("writer END")

def testStart():

try:

for i in xrange(9):

th = threading.Thread(None, reader, "reader %s" % i, (), {}, None)

th.start()

for i in xrange(3):

th = threading.Thread(None, writer, "writer %s" % i, (), {}, None)

th.start()

time.sleep(30) # seconds for testing

finally:

print "main thread: abort()"

mq.abort()

if __name__ == "__main__":

testStart()