Python使用future处理并发问题方案详解

import os
import time
import sys
import requests
POP20_CC = ('CN IN US ID BR PK NG BD RU JP '
            'MX PH VN ET EG DE IR TR CD FR').split()
BASE_URL = 'http://flupy.org/data/flags'
DEST_DIR = 'downloads/'
def sava_flag(img, filename):
    path = os.path.join(DEST_DIR, filename)
    with open(path, 'wb') as fp:
        fp.write(img)
def get_flag(cc):
    url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
    resp = requests.get(url)
    return resp.content
def show(text):
    print(text, end=" ")
    sys.stdout.flush()  # 能在一行中显示
def download_many(cc_list):
    for cc in sorted(cc_list):
        image = get_flag(cc)
        show(cc)
        sava_flag(image, cc.lower() + ".gif")
    return len(cc_list)
def main():
    t0 = time.time()
    count = download_many(POP20_CC)
    elapsed = time.time() - t0
    msg = '\n{} flags download in {:.2f}s'
    print(msg.format(count, elapsed))
if __name__ == '__main__':
    main()

import os
import time
import sys
from concurrent import futures
import requests
POP20_CC = ('CN IN US ID BR PK NG BD RU JP '
            'MX PH VN ET EG DE IR TR CD FR').split()
BASE_URL = 'http://flupy.org/data/flags'
DEST_DIR = 'downloads/'
max_workers = 20  # 设定线程数
def sava_flag(img, filename):
    path = os.path.join(DEST_DIR, filename)
    with open(path, 'wb') as fp:
        fp.write(img)
def get_flag(cc):
    url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
    resp = requests.get(url)
    return resp.content
def show(text):
    print(text, end=" ")
    sys.stdout.flush()  # 能在一行中显示
def download_one(cc):
    image = get_flag(cc)
    show(cc)
    sava_flag(image, cc.lower() + ".gif")
def download_many(cc_list):
    works = min(len(cc_list), max_workers)  # 取其中的最小值，以免创建多余的线程
    with futures.ThreadPoolExecutor(works) as executor:
        res = executor.map(download_one, cc_list)
    return len(list(res))
def main():
    t0 = time.time()
    count = download_many(POP20_CC)
    elapsed = time.time() - t0
    msg = '\n{} flags download in {:.2f}s'
    print(msg.format(count, elapsed))
if __name__ == '__main__':
    main()

...其余代码省略
def download_one(cc):
    image = get_flag(cc)
    show(cc)
    sava_flag(image, cc.lower() + ".gif")
    return cc
def download_many(cc_list):
    cc_list = cc_list[:5]  # 这次演示5个国家
    future_list = []
    with futures.ThreadPoolExecutor(max_workers=3) as executor:  # 线程池为3，便于观察
        for cc in sorted(cc_list):
            future = executor.submit(download_one, cc)  # submit方法排定可调用对象的执行时间，然后返回一个future，表示待执行操作
            future_list.append(future)
            print("{}: {}".format(cc, future))
        res = []
        for future in futures.as_completed(future_list):  # as_completed函数在future都执行完毕后，返回future
            result = future.result()  # 获取future结果
            print('{} result: {!r}'.format(future, result))
            res.append(result)

import time
from concurrent import futures
def display(*args):
    """把参数打印前，加上时间显示"""
    print(time.strftime('[%H:%M:%S]'), end=" ")
    print(*args)
def loiter(n):
    """开始时显示一个消息，然后休眠n秒，最后再结束的时候在显示一个消息
        消息使用制表符缩进，缩进量由n值确定
        loiter：徘徊，闲着，闲荡
    """
    msg = '{}loiter({}):doing nothing for {}s'
    display(msg.format('\t' * n, n, n))
    time.sleep(n)
    msg = '{}loiter({}):done'
    display(msg.format('\t' * n, n))
    return n * 10  # 随意返回一个结果
def main():
    display('Script starting.')  # 脚本开始
    executor = futures.ThreadPoolExecutor(max_workers=3)
    results = executor.map(loiter, range(5))  # 把5个任务交个3个线程
    display('results :', results) # 打印调用executor.map的结果，是一个生成器
    display('waiting for individual results:')  # 等待个体结果
    for i, result in enumerate(results):
        display('result {}: {}'.format(i, result))
if __name__ == '__main__':
    main()

import time
from tqdm import tqdm
for i in tqdm(range(1000)):
    time.sleep(.01)
100%|██████████| 1000/1000 [00:10<00:00, 95.34it/s]

def get_flag(base_url, cc):
    url = '{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())
    resp = requests.get(url)
    if resp.status_code != 200:
        resp.raise_for_status()  # 如果状态码不是200，产生一个HttpError的异常
    return resp.content
def download_one(cc, base_url, verbose=False):
    try:
        image = get_flag(base_url, cc)
    except requests.exceptions.HTTPError as exc:
        res = exc.response
        if res.status_code == 404:
            status = HTTPStatus.not_found
            msg = 'not found'
        else:  # 如果不是404异常，向上冒泡，传给调用方
            raise
    else:
        sava_flag(image, cc.lower() + '.gif')
        status = HTTPStatus.ok
        msg = 'ok'
    if verbose:
        print(cc, msg)
    return Result(status, cc)

def download_many(cc_list, base_url, verbose, max_req):
    """实现依序下载"""
    counter = collections.Counter()  # 统计不同的下载状态：HTTPStatus.ok、HTTPStatus.not_found、HTTPStatus.error
    cc_iter = sorted(cc_list)
    if not verbose:
        cc_iter = tqdm.tqdm(cc_iter)  # 如果不需要详细模式，就使用进度条展示
    for cc in cc_iter:
        try:
            res = download_one(cc, base_url, verbose)
        except requests.exceptions.HTTPError as exc:
            error_msg = "HTTP error {res.status_code} - {res.reason}"
            error_msg = error_msg.format(res=exc.response)
        except requests.exceptions.ConnectionError as exc:
            error_msg = 'Connection error'
        else:
            error_msg = ''
            status = res.status
        if error_msg:
            status = HTTPStatus.error
        counter[status] += 1
        if verbose and error_msg:
            print('*** Error for {}:{}'.format(cc, error_msg))
    return counter

default_concur_req = 30  # 默认的线程池大小
max_concur_req = 1000  # 最大并发请求数，这是一个安全措施
def download_many(cc_list, base_url, verbose, concur_req):
    counter = collections.Counter()
    with futures.ThreadPoolExecutor(max_workers=concur_req) as executor:
        to_do_map = {}
        for cc in sorted(cc_list):
            future = executor.submit(download_one, cc, base_url, verbose)  # submit排定一个可调用对象的执行时间，返回一个Future实例
            to_do_map[future] = cc  # 把各个Future实例映射到国家代码上，在错误处理时使用
        done_iter = futures.as_completed(to_do_map) # 返回一个迭代器，在future运行结束后产出future
        if not verbose:
            done_iter = tqdm.tqdm(done_iter, total=len(cc_list)) # 如果不是详细模式，就显示进度条，因为done_iter没有len函数，只能通过total参数传入
        for future in done_iter:
            try:
                res = future.result()
            except requests.exceptions.HTTPError as exc:
                error_msg = 'HTTP {res.status_code} - {res.reason}'
                error_msg = error_msg.format(res=exc.response)
            except requests.exceptions.ConnectionError as exc:
                error_msg = "Connection error"
            else:
                error_msg = ""
                status = res.status
            if error_msg:
                status = HTTPStatus.error
            counter[status] +=1
            if verbose and error_msg:
                cc = to_do_map[future]
                print('*** Error for {}:{}'.format(cc, error_msg))
        return counter