测试 python3.2+ 中的 ThreadPoolExecutor

test_threadpool.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


"""测试使用 concurrent 提供的线程池.
"""

import time
import random
import concurrent.futures
from concurrent.futures import ThreadPoolExecutor


def worker(name):
    print('First %s' % (name,))
    time.sleep(random.randint(1, 3))
    print('Final %s' % (name,))


with ThreadPoolExecutor(max_workers=2) as executor:
    for i in range(10):
        executor.submit(worker, i)