newpool

一个简化异步协程、线程池和进程池使用方式的 Python 工具库。

特性

• 基于 asyncio 的协程并发：async_gather()、async_taskgroup()；
• 批量线程执行：thread_map()；
• 批量进程执行：process_map()；
• 统一的 Pool API，可选择线程池或 loky 进程池；
• 将同步函数包装为异步函数：to_async()；
• 支持任务状态查询、异常统计和已完成任务回收。

安装

pip install -U newpool

• Python：>=3.11；
• 运行依赖：rich、loky。

快速选择

场景	推荐 API
已经是协程函数	async_gather() 或 async_taskgroup()
I/O 密集型同步函数	thread_map() 或线程模式 Pool
CPU 密集型函数	process_map() 或进程模式 Pool
单个同步函数需要在协程中执行	to_async()

协程并发

async_gather()

并发执行协程，返回与输入顺序一致的结果列表。单个任务发生异常时不会自动中断其他任务，异常对象会作为对应位置的结果返回。

import asyncio
import newpool


async def work(value):
    await newpool.async_sleep(0.1)
    return value * 2


async def main():
    results = await newpool.async_gather(
        [work(i) for i in range(20)],
        sem=5,  # 同时运行的协程数
    )
    print(results)


asyncio.run(main())

async_taskgroup()

基于 asyncio.TaskGroup 并发执行协程，返回已创建的 Task 列表。任务发生异常时，TaskGroup 会按 asyncio 语义取消其他未完成任务并抛出异常组。

import asyncio
import newpool


async def work(value):
    await newpool.async_sleep(0.1)
    return value * 2


async def main():
    tasks = await newpool.async_taskgroup(
        [work(i) for i in range(20)],
        sem=5,
    )
    print([task.result() for task in tasks])


asyncio.run(main())

参数：

• coros_or_futures：单个协程或协程列表；
• sem：并发限制。省略或传 None 时不使用信号量限制。

线程池

批量执行：thread_map()

import newpool


def square(value):
    return value * value


results = newpool.thread_map(square, range(10), n=4)
print(results)

参数：

• n：线程数。省略时使用 min(32, CPU 核心数 + 4)；
• timeout：等待映射结果的超时时间；
• chunksize：兼容统一 map 接口的参数。线程池通常不需要调整。

长生命周期线程池：Pool()

import time
import newpool


pool = newpool.Pool(n=5)


@pool.task
def work(value):
    time.sleep(0.1)
    return value * 2


try:
    futures = [work(i) for i in range(20)]
    pool.wait()
    print([future.result() for future in futures])
    print(pool.state_str())
finally:
    pool.shutdown()

线程池中的任务通过 Future 返回结果和异常：

future = pool.run(work, 1)
pool.wait()

try:
    print(future.result())
except Exception as exc:
    print("任务失败:", exc)

长期运行或需要隔离任务时，建议自己创建 Pool，这样可以独立调用 shutdown()。库也提供了一个内置共享线程池：

import newpool


@newpool.pool.task
def work(value):
    return value * 2


futures = [work(i) for i in range(10)]
newpool.pool.wait()
print([future.result() for future in futures])

newpool.pool 会在模块导入时创建，并由调用方负责管理；如果调用了 newpool.pool.shutdown() 或 newpool.pool.kill()，这个共享池就不能继续提交任务，需要重新创建独立的 Pool。

进程池

进程池基于 loky，适合 CPU 密集型任务。任务函数和参数需要能够被序列化。Windows 或使用进程启动方式的环境中，请把启动代码放在 if __name__ == "__main__": 保护下。

批量执行：process_map()

import newpool


def square(value):
    return value * value


if __name__ == "__main__":
    results = newpool.process_map(square, range(10), n=4)
    print(results)

长生命周期进程池：Pool(mode="process")

import newpool


def work(value):
    return value * value


if __name__ == "__main__":
    pool = newpool.Pool(n=4, mode="process")
    try:
        futures = [pool.run(work, i) for i in range(10)]
        pool.wait()
        print([future.result() for future in futures])
        print(pool.state_dict())
    finally:
        pool.shutdown()

也可以使用装饰器：

import newpool


if __name__ == "__main__":
    pool = newpool.Pool(n=4, mode="process")
    try:
        @pool.task
        def work(value):
            return value * value

        futures = [work(i) for i in range(10)]
        pool.wait()
        print([future.result() for future in futures])
    finally:
        pool.shutdown()

注意：进程池任务异常不会在提交任务时直接抛出。请通过 future.result() 读取结果，或通过 future.exception() 检查异常。不要把连接、锁、打开的文件句柄等不可序列化对象作为进程任务参数。

将同步函数包装为异步：to_async()

to_async() 使用线程池执行同步函数，不会把同步函数变成真正的协程函数。位置参数和关键字参数都支持：

import asyncio
import newpool


@newpool.to_async()
def add(a, b=0):
    return a + b


async def main():
    print(await add(2, 3))
    print(await add(2, b=3))
    print(await add(a=2, b=3))


asyncio.run(main())

可以通过 n 指定线程数：

@newpool.to_async(n=5)
def work(value):
    return value * 2

长期运行程序中，建议注意显式线程池的生命周期；to_async(n=5) 创建的线程池没有单独的公开关闭接口。

Pool 状态、清理和关闭

wait()

等待当前池中跟踪的任务完成。默认等待全部任务完成：

pool.wait()

Pool 支持动态提交任务，等待期间追加到 results 的任务可能会被继续纳入等待。当前版本的 timeout 和 return_when 参数不建议作为复杂控制流程的唯一依据；需要精确超时控制时，请直接使用对应的 Future 或标准库等待接口。

state_dict()、state_str()、state()

state = pool.state_dict()  # 字典
text = pool.state_str()    # 多行字符串
pool.state()               # Rich Panel 输出

状态分为累计统计和当前任务状态：

• 总任务、已完成、成功、取消、报错：累计统计，包含已经归档和当前仍保留的 Future；
• 已归档：已经通过 clear_done() 清理的 Future 数量；
• 当前任务：仍保留在 results 中的 Future 数量；
• 当前已完成：已完成但尚未调用 clear_done() 的 Future 数量；
• 正在运行：当前正在执行的任务数量；
• 剩余任务：已经提交但尚未开始执行的任务数量。

clear_done()

归档并清理已经完成的 Future，保留正在运行和等待中的任务。归档只保存数字，不保存 Future 或结果对象，适合大批量或长期运行的任务。

summary = pool.clear_done()

调用时会直接输出一条摘要，例如：

已归档 10 个已完成任务：成功 8,报错 1,取消 1

同时返回：

{
    "已归档": 10,
    "成功": 8,
    "报错": 1,
    "取消": 1,
}

没有可清理任务时会输出：

没有可归档的已完成任务

调用方如果还需要结果，应在清理前读取 Future.result()：

for future in pool.results:
    if future.done():
        try:
            save_result(future.result())
        except Exception as exc:
            save_error(exc)

pool.clear_done()

clear()

彻底清空当前任务和统计归档：

• 尝试取消尚未开始的任务；
• 清空 results；
• 将总任务、成功、报错和取消统计归零。

已经开始执行的线程任务不一定能够被强制取消，调用后底层线程可能继续运行，但不会再被池跟踪。只想清理已完成任务并保留累计统计时，应使用 clear_done()。

cancel_all()、shutdown() 和 kill()

• cancel_all()：尝试取消所有尚未开始的任务；正在运行的任务通常不能被取消；
• shutdown(wait=True, cancel_futures=False)：关闭池。传入 cancel_futures=True 时，会先调用 cancel_all()；进程池还会终止 worker；
• kill()：等价于 shutdown(wait=False, cancel_futures=True)。进程池会终止 worker；线程池只能取消尚未开始的任务，不能安全强杀正在运行的线程。

关闭后不要继续向同一个池提交任务；如果还要执行新任务，请重新创建 Pool。

API 概览

async_gather(coros_or_futures, sem=None)
async_taskgroup(coros_or_futures, sem=None)
asyncio_run(coro)
thread_map(func, *iterables, n=None, timeout=None, chunksize=1)
process_map(func, *iterables, n=None, timeout=None, chunksize=1)
to_async(n=None)

Pool(n=None, mode="thread")
Pool.run(func, *args, **kwargs)
Pool.task(func)
Pool.callback(func)
Pool.wait(timeout=None, return_when="ALL_COMPLETED")
Pool.cancel_all()
Pool.clear_done()
Pool.clear()
Pool.state_dict()
Pool.state_str(extend=None)
Pool.state(extend=None)
Pool.shutdown(wait=True, cancel_futures=False)
Pool.kill()

许可证

MIT License。