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

Question

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

我有一个用于安排作业的 web.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

我还运行一个线程来处理作业:

def processJobs():
    global incoming_queue, incoming_jobs, current_job, finished_jobs
    while True:
        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 stuff here
        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()

然后我调用添加到队列中的 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 请求，因此该框架可能正在执行一些 fork ，导致您的队列成为单独的实例。

顺便说一句，您可能需要考虑将 Redis 或 beanstalk 作为队列 - 这些使用起来非常简单，并且即使您重新启动应用程序，您的队列也会持续存在。