python-由于阻塞Queue.get()方法而导致死锁
作者：互联网
如标题所示,我陷入僵局,不知道为什么.我有多个生产者,只有一个消费者.线程调用队列的get方法后,schedule_task方法将被多个进程调用
from logging import getLogger
from time import sleep
from threading import Event, Thread
from multiprocessing import Process
from Queue import Queue
class TaskExecutor(object):
def __init__(self):
print("init taskExecutor")
self.event = Event()
self.taskInfos = Queue()
task_thread = Thread(target=self._run_worker_thread)
self._instantEnd = False
self._running = True
task_thread.daemon = True
task_thread.start()
def _run_worker_thread(self):
print("start running taskExcecutor worker Thread")
while self.is_running():
try:
print("try to get queued task from %s" % str(self.taskInfos))
msg, task = self.taskInfos.get()
print("got task %s for msg: %s" % str(task), str(msg))
task.execute(msg)
self.taskInfos.task_done()
except Exception, e:
logger.error("Error: %s" % e.message)
print("shutting down TaskExecutor!")
def is_running(self):
return True
def schedule_task(self, msg, task):
try:
print("appending task '%s' for msg: %s" % (str(task), str(msg)))
self.taskInfos.put((msg, task))
print("into queue: %s " % str(self.taskInfos))
except Exception, e:
print("queue is probably full: %s" % str(e))
class Task(object):
def execute(self, msg):
print(msg)
executor = TaskExecutor()
def produce():
cnt = 0
while True:
executor.schedule_task("Message " + str(cnt), Task())
cnt += 1
sleep(1)
for i in range(4):
p = Process(target=produce)
p.start()
从我的日志中,我得到：
init taskExecutor
start running taskExcecutor worker Thread
try to get queued task from <Queue.Queue instance at 0x7fdd09830cb0>
appending task '<__main__.Task object at 0x7fdd098f8f10>' for msg: Message 0
into queue: <Queue.Queue instance at 0x7fdd09830cb0>
appending task '<__main__.Task object at 0x7fdd098f8f10>' for msg: Message 0
into queue: <Queue.Queue instance at 0x7fdd09830cb0>
appending task '<__main__.Task object at 0x7fdd098f8f10>' for msg: Message 0
into queue: <Queue.Queue instance at 0x7fdd09830cb0>
appending task '<__main__.Task object at 0x7fdd098f8f10>' for msg: Message 0
into queue: <Queue.Queue instance at 0x7fdd09830cb0>
appending task '<__main__.Task object at 0x7fdd086f35d0>' for msg: Message 1
into queue: <Queue.Queue instance at 0x7fdd09830cb0>
appending task '<__main__.Task object at 0x7fdd086f3490>' for msg: Message 1
into queue: <Queue.Queue instance at 0x7fdd09830cb0>
appending task '<__main__.Task object at 0x7fdd086f3b10>' for msg: Message 1
into queue: <Queue.Queue instance at 0x7fdd09830cb0>
appending task '<__main__.Task object at 0x7fdd086f3b10>' for msg: Message 1
into queue: <Queue.Queue instance at 0x7fdd09830cb0>
有人可以解释一下,我想念什么？
解决方法:
尽管其他人无法运行此代码(这不是独立的),但是您显示的内容没有明显的问题.因此,问题出在未显示的问题上-可能是在创建和使用TaskExecutor实例的代码中.
当我凭空插入缺少的部分时,此代码可以正常工作.
因此,您不仅需要展示更多.如何更换：
logger.debug("try to get queued task")
与
logger.debug("try to get queued task from queue %s", self.taskInfos)
？然后至少我们可以看到您的生产者是否正在使用与您的消费者相同的队列.
下一个
感谢您添加.接下来：这是一个独立的程序供您尝试.这非常像您的代码.看看它是否为您正确运行(对我有用)：
from threading import Thread, Lock
from Queue import Queue
class Logger:
def __init__(self):
self.iolock = Lock()
def debug(self, str, *msg):
with self.iolock:
print str % msg
error = debug
logger = Logger()
class TaskExecutor(object):
def __init__(self):
logger.debug("init taskExecutor")
self.taskInfos = Queue()
task_thread = Thread(target=self._run_worker_thread)
task_thread.daemon = True
task_thread.start()
def is_running(self):
return True
def _run_worker_thread(self):
logger.debug("start running taskExcecutor worker Thread")
while self.is_running():
try:
logger.debug("try to get queued task from queue %s", self.taskInfos)
msg, task = self.taskInfos.get()
logger.debug("got task %s for msg: %s", str(task), str(msg))
#task.execute(msg)
self.taskInfos.task_done()
except Exception, e:
logger.error("Error: %s", e.message)
logger.debug("shutting down TaskExecutor!")
def schedule_task(self, msg, task):
try:
logger.debug("appending task '%s' for msg: %s", str(task), str(msg))
self.taskInfos.put((msg, task))
logger.debug("into queue: %s ", str(self.taskInfos))
except Exception, e:
logger.debug("queue is probably full: %s", str(e))
te = TaskExecutor()
def runit():
for i in range(10):
te.schedule_task("some task", i)
main = Thread(target=runit)
main.start()
下一个
好的,此代码可能无法正常工作.在Linux-y系统上,仅在以下位置创建了一个TaskExecutor实例：
executor = TaskExecutor()
这发生在主程序中.每次您执行以下操作：
p = Process(target=produce)
您的主程序是fork()的.尽管分叉的进程也可以看到执行程序,但这是主程序执行程序的地址空间副本,与主程序中的执行程序没有任何关系(通常为写时复制fork()语义).
每个子进程还具有执行者数据成员的副本,包括其队列.所有子进程都将数据放在其自己的唯一执行者副本上,但是使用者线程仅在主程序中运行,并且工作进程对其执行者副本所做的任何操作都不会对主程序的使用者线程所见内容产生任何影响.
所以这真的很困惑.我现在必须停止尝试在这里弄清楚您可能真正想做的事情;-)如果您想发挥创意,请使用multiprocessing.Queue进行调查.在进程之间进行通信的唯一方法是使用从头开始构建的对象,以支持进程间的通信. Queue.Queue将永远无法工作.
还有一个
这是一个跨进程,甚至在Windows上都可以正常运行的;-)
from time import sleep
from threading import Thread
from multiprocessing import Process, JoinableQueue
class TaskExecutor(Thread):
def __init__(self):
print("init taskExecutor")
Thread.__init__(self)
self.taskInfos = JoinableQueue()
def getq(self):
return self.taskInfos
def run(self):
print("start running taskExcecutor worker Thread")
while self.is_running():
try:
print("try to get queued task from %s" % self.taskInfos)
msg, task = self.taskInfos.get()
print("got task %s for msg: %s" % (task, msg))
task.execute(msg)
self.taskInfos.task_done()
except Exception, e:
print("Error: %s" % e.message)
print("shutting down TaskExecutor!")
def is_running(self):
return True
class Task(object):
def execute(self, msg):
print(msg)
def produce(q):
cnt = 0
while True:
q.put(("Message " + str(cnt), Task()))
cnt += 1
sleep(1)
if __name__ == "__main__":
executor = TaskExecutor()
executor.start()
for i in range(4):
p = Process(target=produce, args=(executor.getq(),))
p.start()
if __name__ ==“ __main__”部分不仅允许它在Windows上运行,它也具有很大的“文档”值,一眼就可以看出执行程序实际上仅在主程序中运行.
您的问题是,这是否就是您想要的分工.您是否真的要主要流程-并且仅主要流程-来完成所有
task.execute(msg)
工作？从这里无法猜测是否就是您想要的.这就是代码的作用.
风格点：请注意,这摆脱了schedule_task()方法.并行处理可能很困难,而且数十年来,我发现保持线程间/进程间通信尽可能简单且显而易见是非常有价值的.这意味着,除其他外,直接使用消息队列,而不是例如将消息隐藏在方法中.在这种情况下,抽象层通常使正确的代码难以创建,扩展,调试和维护.
标签：producer-consumer,python-2-7,multiprocessing,queue,python 来源： https://codeday.me/bug/20191122/2059319.html