Python：队列和线程的阻塞问题

Question

我在Python队列和线程方面遇到了一个奇怪的问题。在

我有一个网页.py调度作业的应用程序，因此具有global incoming_queue = Queue(maxsize=10)。在

我有一个url和一个GET处理程序，可以添加到队列中（我还可以添加到列表中，因为我需要知道队列的内容）：

class ProcessRequest:
    def GET(self):
        global incoming_queue, incoming_jobs
        if incoming_queue.full():
            print "Queue is full"
            return web.InternalError("Queue is full, please try submitting later.")
        else:
            job_id = getNextInt()
            req_folder = "req" + str(job_id)
            incoming_queue.put(job_id)
            incoming_jobs.append(job_id)
            print "Received request, assigning Drop Folder {0}".format(req_folder)
            web.header('Drop-Folder', req_folder)
            return req_folder

我还运行一个线程来处理这些作业：

^{pr2}$

启动程序时，我运行以下程序：

if __name__ == '__main__':
    job_processor_thread = threading.Thread(target=processJobs)
    job_processor_thread.start()
    app.run()

然后调用添加到队列中的URL。使用另一个url，我可以检查该项是否确实添加到列表中，并将以下代码添加到原始url处理程序（print incoming_queue.get()）的GET方法中，我验证了该项确实被添加到队列中。在

作业处理线程只是在current_job = incoming_queue.get(block=True)处阻塞。这是有意的。但是，它永远不会解除阻止，即使在将项目添加到队列中时也是如此。它只是永远被封锁。在

为什么？就像它有一个单独的队列对象副本。在

编辑：根据Martin的建议，我决定尝试查看GET方法和processJobs方法中引用了什么对象。在

processJobs(): <Queue.Queue instance at 0x7f32b6958a70>GET(): <Queue.Queue instance at 0x7f32b5ec5368>

是的，它们是不同的，但为什么呢？在

编辑2：以下是整个脚本供参考：

'''
Created on Apr 20, 2015

@author: chris
'''
import web
import time
import threading
import json
from Queue import Queue, Empty
import os

urls = (
        '/request', 'ProcessRequest',
        '/status', 'CheckStatus',
    )

current_job_thread = threading.Thread()

app = web.application(urls, globals())

incoming_jobs = []
incoming_queue = Queue(maxsize=10)

current_job = None

finished_jobs = []

next_int = 0

def getNextInt():
    global next_int, incoming_queue
    the_int = next_int
    next_int += 1
    return the_int

class ProcessRequest:
    def GET(self):
        global incoming_queue, incoming_jobs
        if incoming_queue.full():
            print "Queue is full"
            return web.InternalError("Queue is full, please try submitting later.")
        else:
            job_id = getNextInt()
            req_folder = "req" + str(job_id)
            print incoming_queue
            incoming_queue.put(job_id)
            incoming_jobs.append(job_id)
            print "Received request, assigning Drop Folder {0}".format(req_folder)
            web.header('Drop-Folder', req_folder)
            return req_folder

class CheckStatus:
    def GET(self):
        global incoming_queue, incoming_jobs, current_job, finished_jobs
        if str(web.input().jobid) == 'all':
            # Construct JSON to return
            web.header('Content-Type', 'application/json')
            return {'In Queue': incoming_jobs,
                            'Currently Processing': current_job,
                            'Finished': finished_jobs
                    }
        try:
            jobid = int(web.input().jobid)
        except ValueError:
            jobid = -1
        print jobid
        if jobid in finished_jobs:
            file_string = "results{0}.json".format(jobid)
            try:
                json_file = open(file_string)
                finished_jobs.remove(jobid)
                os.remove(file_string)
                web.header('Process-Status', 'Complete')
                web.header('Content-Type', 'application/json')
                return json.load(json_file)
            except IOError:
                web.header('Process-Status', 'Complete, but failed to retrieve file, saving')
                return ""

        elif jobid is current_job:
            web.header('Process-Status', 'Processing')
        elif jobid in incoming_jobs:
            web.header('Process-Status', 'In Queue')
        else:
            web.header('Process-Status', 'Unknown')
        return ""         

def processJobs():
    global incoming_queue, incoming_jobs, current_job, finished_jobs
    while True:
        print incoming_queue
        print "Job processor thread active"
        current_job = incoming_queue.get(block=True)
        incoming_jobs.remove(current_job)
        print "Processing job {0}".format(current_job)
        # Do magical Spark stuff here
        time.sleep(10)  # Simulate a Spark Job
        finished_jobs.append(current_job)
        current_job = None
        print "Job processor thread ready for next job"
    print "Job processor thread finished"

if __name__ == '__main__':
    job_processor_thread = threading.Thread(target=processJobs)
    job_processor_thread.start()
    app.run()

Answer 1

只需打印对象，就可以测试它们是不同队列的假设：

def processJobs():
    global incoming_queue, incoming_jobs, current_job, finished_jobs
    print incoming_queue # print something like <__main__.Queue instance at 0x7f556d93f830>


class ProcessRequest:
    def GET(self):
        global incoming_queue, incoming_jobs
        print incoming_queue # print something like <__main__.Queue instance at 0x7f556d93f830>

确保内存地址（0x7f556d93f830）匹配。在

您从未提及是否使用框架来处理web请求，因此可能是该框架正在执行一些分叉操作，从而导致您的队列成为单独的实例。在

顺便说一句，你可能想把Redis或beanstalk看作一个队列—这些都非常简单，即使重新启动应用程序，你的队列也会持续存在。在

Answer 2

在马丁的指导下，我用这里的想法解决了这个问题：https://groups.google.com/forum/#!topic/webpy/u-cfL7jLywo。在

基本上，网页.py在发出请求时重新创建全局变量，因此如果要在框架和其他线程之间共享数据，则不能使用全局变量。解决方案是创建另一个模块，在该模块中创建一个类，然后向其中添加变量定义。我最后得出的结论是：

在jobqueue.py公司名称：

'''
Created on Apr 23, 2015

@author: chris
'''
import Queue

class JobManagement:
    incoming_queue = Queue.Queue(maxsize=10)
    incoming_jobs = []
    current_job = None
    finished_jobs = []

在主.py公司名称：

^{pr2}$