首页 / PYTHON / python-由于阻塞Queue.get()方法而导致死锁

python-由于阻塞Queue.get()方法而导致死锁

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了python-由于阻塞Queue.get()方法而导致死锁，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含6688字，纯文字阅读大概需要10分钟。

内容图文

如标题所示,我陷入僵局,不知道为什么.我有多个生产者,只有一个消费者.线程调用队列的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()方法.并行处理可能很困难,而且数十年来,我发现保持线程间/进程间通信尽可能简单且显而易见是非常有价值的.这意味着,除其他外,直接使用消息队列,而不是例如将消息隐藏在方法中.在这种情况下,抽象层通常使正确的代码难以创建,扩展,调试和维护.

内容总结

以上是互联网集市为您收集整理的python-由于阻塞Queue.get()方法而导致死锁全部内容，希望文章能够帮你解决python-由于阻塞Queue.get()方法而导致死锁所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/658468.html

来源：【匿名】