对传递给函数的列表进行多重处理

Question

我有一个一次处理一个url的函数：

def sanity(url):
    try:
       if 'media' in url[:10]:
           url = "http://dummy.s3.amazonaws.com" + url
       req = urllib.request.Request(url, headers={'User-Agent' : "Magic Browser"})
       ret = urllib.request.urlopen(req)
       allurls.append(url)
       return 1
    except (urllib.request.HTTPError,urllib.request.URLError,http.client.HTTPException, ValueError) as e:
       print(e, url)
       allurls.append(url)
       errors.append(url)
       return 0

在main函数中，我有一个需要由上述函数处理的url列表。我试过了，但没用。你知道吗

start=0
allurls=[]
errors=[]
#arr=[0,100,200...]
for i in arr:
    p=Process(target=sanity,args=(urls[start:i],))
    p.start()
    p.join()

上面的代码应该处理一批100个url。但它不起作用。我知道它不起作用，因为我正在将列表allurls和errors写入两个不同的文件，当它们不应该是空的时候，它们是空的。我发现名单是空的。我不明白这种行为。你知道吗

Answer 1

如果我理解正确，您希望一次处理一个列表的块，但是并行处理这些块？其次，要将答案存储在全局变量中。问题是processes are not threads, so you much more involved to share memory between them。你知道吗

因此，另一种方法是返回答案，下面的代码可以帮助您这样做。首先，需要将列表转换为列表列表，每个列表都包含要在该块中处理的数据。然后可以将列表列表传递给处理其中每个列表的函数。每个块的输出是一个答案列表和一个错误列表（我建议将其转换为dict以跟踪哪个块抛出了错误）。然后在进程返回后，您可以解开列表列表，创建您的答案列表和错误列表。你知道吗

以下是实现上述目标的代码：

from multiprocessing import Pool

def f(x):
    try:
        return [x*x, None]  # 0 for sucess
    except Exception as e:
        return [None, e]  # 1 for failure

def chunk_f(x):
    output = []
    errors = []
    for xi in x:
        ans, err = f(xi)
        if ans:
            output.append(ans)
        if err:
            errors.append(err)
    return [output, errors]

n = 10  # chunk size
data = list(range(95))  # test data
data.extend(['a', 'b'])

l = [data[k*n:(k+1)*n] for k in range(int(len(data)/n+1))]

p = Pool(8)
d = p.map(chunk_f, l)

new_data = []
all_errors = []
for da, de in d:
    new_data.extend(da)
    all_errors.extend(de)
print(new_data)
print(all_errors)

您还可以查看this stack overflow answer on different methods of chunking your data。你知道吗