【发布时间】：2015-04-23 21:02:01
【问题描述】：
我在 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>
是的，它们是不同的，但为什么呢？
EDIT #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()
【问题讨论】：
标签： python multithreading python-2.7 queue web.py