線程/隊列掛起問題

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 

（未經測試）

[/編輯]