4. 核心参数详解（下）：chunksize参数的作用、map与submit的选择、超时设置与异常处理

好，咱们接着聊。上一节我们把线程池的核心参数拆了个七七八八，今天要聊的这几个，说实话，才是真正决定你线上代码稳不稳的关键。

我见过太多人，线程池配得漂漂亮亮，一上生产就崩。为什么？因为忽略了chunksize，用错了map和submit，或者压根没处理超时和异常。嗯，今天咱们就把这几个坑填平。

4.1 chunksize：大任务拆小，效率翻倍

先问个问题：你用concurrent.futures的map()时，有没有想过它是怎么把数据分给各个线程的？

默认情况下，map()会把可迭代对象拆成一个个元素，每个元素丢给一个线程。这在小数据量时没问题，但如果你有10万个URL要爬，每个线程只拿一个URL，那线程切换的开销会大得吓人。

chunksize就是干这个的。它决定了每个线程一次拿多少个任务。

我个人习惯，在IO密集型任务里，把chunksize设成10~100。比如爬虫任务，每个请求耗时0.5秒，如果chunksize=1，线程切换频繁，反而慢。设成50，一个线程一次拿50个URL，一口气处理完，效率高很多。

举个例子：

from concurrent.futures import ThreadPoolExecutor
import time

def fetch_url(url):
    # 模拟网络请求
    time.sleep(0.5)
    return f"OK: {url}"

urls = [f"http://example.com/{i}" for i in range(1000)]

# 不指定chunksize，默认每个线程拿1个
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(fetch_url, urls))

# 指定chunksize=50
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(fetch_url, urls, chunksize=50))

我在项目中遇到过，一个数据处理任务，数据量200万条，默认chunksize跑了3分钟。我改成chunksize=1000，直接降到40秒。你想想看，这差距有多大。

4.2 map vs submit：选对工具，事半功倍

这两个方法，初学者经常混着用。其实它们各有各的脾气。

特性	map	submit
返回值	迭代器，按输入顺序返回结果	Future对象，可单独获取结果
异常处理	抛出第一个异常，后续任务可能不执行	每个Future可单独捕获异常
任务控制	一次性提交所有任务	可逐个提交，可取消
适用场景	批量处理，结果顺序重要	需要精细控制，或结果顺序不重要

map的优势：代码简洁，结果顺序和输入顺序一致。适合你有一批数据，每个数据做同样的操作，而且你关心结果的顺序。

submit的优势：灵活。你可以拿到Future对象，随时检查任务状态、取消任务、单独设置超时。适合任务之间有依赖，或者你需要动态提交任务的场景。

举个例子，你有个任务列表，但其中某些任务可能超时，你想单独处理：

from concurrent.futures import ThreadPoolExecutor, as_completed

def task(n):
    import time
    time.sleep(n)
    return n

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(task, i): i for i in [1, 2, 3, 4, 5]}
    for future in as_completed(futures):
        try:
            result = future.result(timeout=3)
            print(f"任务完成: {result}")
        except TimeoutError:
            print(f"任务超时: {futures[future]}")

你看，用submit + as_completed，你可以逐个处理结果，哪个先完成就先处理哪个。而map只能等所有任务完成，再按顺序返回。

我个人建议：批量无状态任务用map，需要精细控制用submit。别纠结，选对了能省一半代码量。

4.3 超时设置：别让线程死等

线程池最怕什么？怕某个任务卡住不返回。比如网络请求超时、数据库连接挂掉、死循环。如果不设超时，整个线程池可能被一个坏任务拖垮。

Python的Future.result(timeout)可以设置超时。但要注意，这个超时是等待结果的时间，不是任务执行的时间。

from concurrent.futures import ThreadPoolExecutor, TimeoutError

def slow_task():
    import time
    time.sleep(10)
    return "done"

with ThreadPoolExecutor(max_workers=2) as executor:
    future = executor.submit(slow_task)
    try:
        result = future.result(timeout=3)  # 最多等3秒
        print(result)
    except TimeoutError:
        print("任务超时了，但线程还在跑！")

嗯，这里要注意：超时只是你不再等结果了，但任务本身还在后台运行。如果你不处理，那个线程会一直被占用，直到任务完成。

我在项目中遇到过，一个爬虫任务，某个网站响应慢，设了5秒超时。结果超时后，那个线程还在后台等响应，导致线程池慢慢被占满。后来我加了个future.cancel()，但cancel只能取消还没开始执行的任务。对于正在执行的任务，cancel无效。

怎么办？我的做法是：在任务函数内部自己实现超时逻辑，比如用timeout_decorator库，或者用signal模块（仅限Unix）。

import signal

class TimeoutError(Exception):
    pass

def handler(signum, frame):
    raise TimeoutError("任务执行超时")

def task_with_timeout(seconds):
    signal.signal(signal.SIGALRM, handler)
    signal.alarm(seconds)  # 设置超时
    try:
        # 你的业务逻辑
        import time
        time.sleep(10)
        return "done"
    finally:
        signal.alarm(0)  # 取消闹钟

# 然后在线程池里调用
with ThreadPoolExecutor(max_workers=2) as executor:
    future = executor.submit(task_with_timeout, 3)
    try:
        print(future.result())
    except TimeoutError:
        print("任务被强制终止了")

4.4 异常处理：别让一个坏蛋毁了整锅粥

线程池里的异常，默认是不会打印出来的。你提交一个任务，它抛了个异常，然后……就没了。你根本不知道发生了什么。

为什么会这样？因为异常被封装在Future对象里了。只有你调用future.result()时，异常才会被重新抛出。

看个例子：

from concurrent.futures import ThreadPoolExecutor

def bad_task(x):
    if x == 3:
        raise ValueError("x不能是3")
    return x * 2

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(bad_task, i) for i in range(5)]
    for future in futures:
        try:
            result = future.result()
            print(result)
        except Exception as e:
            print(f"捕获到异常: {e}")

如果你用map，异常处理更麻烦。map会在遇到第一个异常时停止，并且抛出异常，后续任务可能还没执行就没了。

with ThreadPoolExecutor(max_workers=3) as executor:
    try:
        results = list(executor.map(bad_task, range(5)))
    except Exception as e:
        print(f"map抛出异常: {e}")
        # 注意：此时部分任务可能已经执行，部分还没执行

我的建议：尽量用submit + as_completed，每个Future单独捕获异常。这样即使某个任务挂了，其他任务不受影响。

另外，别忘了在任务函数内部也加try/except。有些异常（比如KeyboardInterrupt、SystemExit）在线程里是捕获不到的。最好在任务函数最外层包一层：

def safe_task(x):
    try:
        # 你的业务逻辑
        return do_something(x)
    except Exception as e:
        # 记录日志
        print(f"任务失败: {x}, 错误: {e}")
        return None  # 或者返回一个默认值

这样，即使任务挂了，你也能知道原因，而且不会影响线程池的其他任务。

4.5 知识体系总览

说了这么多，咱们用一张图把今天的内容串起来：

好了，今天的内容就到这。chunksize、map与submit的选择、超时和异常处理，这几个点看似简单，但用好了，你的线程池代码会稳如老狗。下次咱们聊聊更高级的话题——线程池的监控与动态调整。

可可精品资料库