一日一技：等待多个线程同时结束的两种方法

随风 • 2021-10-06 • Python知识 • 474 阅读

摄影：产品经理

这是「进击的Coder」的第 433 篇技术分享

作者：kingname

来源：未闻Code

“

阅读本文大概需要 8 分钟。

”

我们在写多线程代码的时候，可能会需要等待多个线程同时结束，然后再进行后续的流程。例如，我做了一个聚合搜索引擎，用户输入一个关键词，我需要同时在很多个搜索引擎上搜索，然后把搜索结果汇总以后返回给用户。

示例代码如下：

@app.get('/api/search')
def search(keyword: str):
    google_result = requests.get('Google 搜索地址').text
    baidu_result = requests.get('百度搜索地址').text
    bing_result = requests.get('Bing搜索地址').text
    result = combine(google_result, baidu_result, bing_result)
    return {'success': True, 'result': result}

从上面这段代码，大家可能会发现一个问题，就是在请求多个搜索引擎的时候是串行的，先访问 Google，访问完成再访问百度，访问完成最后访问 Bing。这样显然会浪费大量的时间。

如果你不会async/await，那么为了解决这个问题，你能想到的显然就是使用多线程。使用 3 个线程同时访问 Google、百度和 Bing，然后把结果汇总传入combine函数，不就解决问题了吗？

如果仅仅是启动多个线程，那么做法很简单：

import threading

def get_url(url):
    result = requests.get(url, headers=HEADERS).text
    return result

@app.get('/api/search')
def search(keyword: str):
    google_thead = threading.Thread(target=get_url, 'Google 搜索地址')
    baidu_thread = threading.Thread(target=get_url, '百度搜索地址')
    bing_thread = threading.Thread(target=get_url, 'Bing搜索地址')
    google_thread.start()
    baidu_thread.start()
    bing_thread.start()
    ...

现在问题来了，三个线程确实已经启动了，但你怎么知道到什么时候为止，所有线程都运行完毕？

这里我们给出几个方法。

使用 join

调用线程的.join()方法，就可以卡住主线程，直到这个子线程运行完毕才能让主线程继续运行后面的代码。所以我们可以修改代码为：

import threading

def get_url(url):
    result = requests.get(url, headers=HEADERS).text
    return result

@app.get('/api/search')
def search(keyword: str):
    google_thead = threading.Thread(target=get_url, 'Google 搜索地址')
    baidu_thread = threading.Thread(target=get_url, '百度搜索地址')
    bing_thread = threading.Thread(target=get_url, 'Bing搜索地址')
    google_thread.start()
    baidu_thread.start()
    bing_thread.start()
    
    google_thread.join()
    baidu_thread.join()
    bing_thread.join()

但等一等，我怎么拿到子线程的返回呢？在默认情况下，你确实拿不到返回的数据。所以你需要传入一个东西去子线程接收结果。所以代码可以改为：

import threading

def get_url(url, output):
    result = requests.get(url, headers=HEADERS).text
    output.append(result)

@app.get('/api/search')
def search(keyword: str):
    result = []
    google_thead = threading.Thread(target=get_url, args=['Google 搜索地址', result])
    baidu_thread = threading.Thread(target=get_url, args=['百度搜索地址', result])
    bing_thread = threading.Thread(target=get_url, args=['Bing搜索地址', result])
    google_thread.start()
    baidu_thread.start()
    bing_thread.start()
    
    google_thread.join()
    baidu_thread.join()
    bing_thread.join()
    combine(*result)

因为线程是共享内存的，所以他们可以直接修改主线程传入的列表。

在使用.join()的时候，需要小心不要把.join()放错了地方，否则你的多线程就会变成单线程。详情可以看我的这篇文章：等一等，你的多线程可别再乱 join 了。

ThreadPoolExecutor

Python 自带了一个concurrent模块，它就是专门用来处理并发问题的。我们也可以使用这个模块中的ThreadPoolExecutor来解决问题：

from concurrent.futures import ThreadPoolExecutor, as_completed

def get_url(url):
    result = requests.get(url, headers=HEADERS).text
    return result

@app.get('/api/search')
def search(keyword: str):
    tasks = []
    with ThreadPoolExecutor() as executor:
        for url in ['Google 搜索地址', '百度搜索地址', 'Bing搜索地址']
            task = executor.submit(get_url, url)
            tasks.append(task)
        result = [x.result() for x in as_completed(tasks)]
 
    combine(*result)
    ...