线程/队列挂起问题

Question

新手线程在这里。我是从这个thread借用大量的代码，同时试图使用线程/队列建立我的第一个脚本：

import threading, urllib2 
import Queue 
import sys 
from PIL import Image 
import io, sys 

def avhash(url,queue): 
    if not isinstance(url, Image.Image): 
     try: 
      im = Image.open(url) 
     except IOError: 
      fd=urllib2.urlopen(url) 
      image_file=io.BytesIO(fd.read()) 
      im=Image.open(image_file) 
      im = im.resize((8, 8), Image.ANTIALIAS).convert('L') 
      avg = reduce(lambda x, y: x + y, im.getdata())/64. 
    hash = reduce(lambda x, (y, z): x | (z << y), 
        enumerate(map(lambda i: 0 if i < avg else 1, im.getdata())), 
        0) 

    queue.put({url:hash}) 
    queue.task_done() 

def fetch_parallel(job_list): 
    q = Queue.Queue() 
    threads = [threading.Thread(target=avhash, args = (job,q)) for job in job_list[0:50]] 
    for t in threads: 
     t.daemon = True 
     t.start() 

    for t in threads: 
     t.join() 
    return [q.get() for _ in xrange(len(job_list))] 

在这种情况下，job_list是URL列表。我发现这个代码在这个列表等于或小于50时工作正常，但当> 50时它会挂起。我一定有一些事情我根本不了解线程是如何工作的？

Answer 1

你的问题是这样的一行：

return [q.get() for _ in xrange(len(job_list))] 

如果job_list有超过50个元素，然后尝试读取您的队列更多的结果比你投入，因此：

return [q.get() for _ in xrange(len(job_list[:50]))] 

或者，甚至更好：

MAX_LEN = 50 
... 
threads = [... for job in job_list[:MAXLEN]] 
... 
return [q.get() for _ in job_list[:MAXLEN]] 

[编辑]

看起来你希望你的程序做一些与它不同的事情。您的程序将采用job_list中的前50个条目，在线程中处理每个条目并忽略所有其他作业。从您的评论下面我假设你想处理所有的工作，但一次只有50。为此，您应该使用线程池。在Python> = 3.2中，您可以使用concurrent.futures.ThreadPoolExecutor[link]。

在Python < 3.2你必须推出自己的：

CHUNK_SIZE = 50 

def fetch_parallel(job_list): 
    results = [] 
    queue = Queue.Queue() 
    while job_list: 
     threads = [threading.Thread(target=avhash, args=(job, queue)) 
         for job in job_list[:CHUNK_SIZE]] 
     job_list = job_list[CHUNK_SIZE:] 
     for thread in threads: 
      thread.daemon = True 
      thread.start() 
     for thread in threads: 
      thread.join() 
     results.extend(queue.get() for _ in threads) 
    return results 

（未经测试）

[/编辑]