在本食谱中，你将学习如何在 Flet 应用中使用 Python 3.14 的 子解释器（subinterpreters） （concurrent.interpreters 和 concurrent.futures.InterpreterPoolExecutor）， 实现真正的多核 CPU 并行。

子解释器是运行在同一进程内的独立 Python 解释器。从 Python 3.12 起，每个子解释器都拥有 自己的 GIL，因此多个子解释器可以同时 在多个 CPU 核心上运行纯 Python 代码——而无需启动独立的进程。

何时使用哪种

	运行于	真正的 CPU 并行	说明
线程	一个解释器，一个 GIL	❌（纯 Python）	最适合 I/O，或会释放 GIL 的 C 库
子解释器	一个进程，N 个解释器	✅	进程内运行，支持移动端；数据共享受限；无法强制取消
多进程	N 个进程	✅	完全隔离，可硬取消工作进程；更重，在 Flet 中仅限桌面端

当你需要多个核心执行 Python 任务、又想留在单一进程内时，就使用子解释器——尤其是移动端， 因为在那里 multiprocessing 根本无法派生子进程。

::: important 平台与版本支持 子解释器需要 Python 3.14 或更高版本。使用 flet build 打包应用时， 请确保内置的 Python 版本满足此要求。 在开发阶段（例如使用 flet run 时），你虚拟环境中的 Python 解释器也必须满足此要求。

它们适用于 Flet 在 macOS、Windows、Linux、iOS 和 Android 上的应用。

在 Web 端，则取决于你的 Python 实际运行在哪里：

• 动态网站 把你的应用作为普通的 CPython 进程（FastAPI/Uvicorn）在服务端运行， 因此子解释器的行为与桌面端完全一致——只要服务端运行的是 Python 3.14。
• 静态网站 完全在浏览器中通过 Pyodide 运行，这是一个 单线程的 WebAssembly 运行时， 没有按解释器划分的 GIL——因此在那里子解释器不可用。 :::

规则

在模块顶层定义工作函数

要在另一个解释器中运行工作函数，CPython 会把它复制过去——连同它的代码，以及它所引用的模块级函数和常量。 请在模块的顶层定义工作函数（可以是像下面示例那样写在你的 main.py 中，也可以是单独的文件）； 在 macOS、Windows、Linux、iOS 和 Android 上行为完全一致。

把工作函数嵌套在 main() 或按钮事件处理器内部，只有在它无状态时才有效——即没有捕获的变量、也没有引用模块全局变量—— 因此一旦它引用了某个辅助函数或常量，就会失败并抛出 NotShareableError: only stateless functions are shareable。 顶层函数没有此限制：它可以自由调用其他模块级辅助函数。（另外，不要在生成器表达式中调用辅助函数——见注意事项。）

只传递可序列化 / 可共享的数据

参数和返回值需要被 pickle 才能跨解释器边界，因此它们必须是可 pickle 的。 底层的 Queue 还可以直接接受可共享对象——数字、str、bytes、None、由这些类型组成的元组，以及队列本身。 不要传递 Flet 控件、page、打开的文件或数据库连接。

不要从子解释器操作 GUI

工作进程运行在隔离的解释器中，无法访问你的页面。请返回数据（或通过 Queue 流式上报）， 并在主解释器中更新 UI。

示例

跨核心的并行映射

InterpreterPoolExecutor 是 ProcessPoolExecutor 的直接替代品：它在每个子解释器中运行一个任务，并且由于每个子解释器都有自己的 GIL， 能同时利用多个核心——全部在同一个进程内。

import time
from concurrent.futures import InterpreterPoolExecutor, as_completed

import flet as ft


def _is_prime(n: int) -> bool:
    """Returns True if `n` is prime."""
    if n < 2:
        return False
    for d in range(2, int(n**0.5) + 1):
        if n % d == 0:
            return False
    return True


def count_primes(limit: int) -> int:
    """Count the primes below `limit` (CPU-bound, pure Python)."""
    count = 0
    for n in range(2, limit):
        if _is_prime(n):
            count += 1
    return count


def main(page: ft.Page):
    def start():
        button.disabled = True  # block a second run while this one is in flight
        page.update()
        page.run_thread(run)

    def run():
        """Time the same work sequentially and across a pool, then report the
        speedup. Runs on a background thread so the UI stays responsive."""
        limits = [200_000 + i * 30_000 for i in range(8)]

        # Baseline: run every chunk in this one interpreter (one core).
        status.value = "Sequential…"
        progress.value = 0
        page.update()
        started = time.perf_counter()
        for done, limit in enumerate(limits, 1):
            count_primes(limit)
            progress.value = done / len(limits)
            page.update()
        seq_time = time.perf_counter() - started

        # Parallel: one subinterpreter per chunk, each with its own GIL.
        status.value = "Parallel…"
        progress.value = 0
        page.update()
        primes = completed = 0
        started = time.perf_counter()
        with InterpreterPoolExecutor() as pool:  # sizes itself to the CPU count
            futures = [pool.submit(count_primes, n) for n in limits]
            for future in as_completed(futures):
                primes += future.result()
                completed += 1
                progress.value = completed / len(futures)
                page.update()
        par_time = time.perf_counter() - started

        status.value = (
            f"{primes} primes · sequential {seq_time:.1f}s · "
            f"parallel {par_time:.1f}s · {seq_time / par_time:.1f}× faster"
        )
        button.disabled = False
        page.update()

    page.add(
        ft.SafeArea(
            content=ft.Column(
                controls=[
                    button := ft.Button(
                        "Count primes: sequential vs parallel",
                        on_click=start,
                    ),
                    progress := ft.ProgressBar(value=0, width=300),
                    status := ft.Text(),
                ]
            )
        )
    )


if __name__ == "__main__":
    ft.run(main)

该示例对同样的任务先按序运行、再放进进程池并行运行，并报告加速比。整体编排通过 page.run_thread 移出 UI 线程，并行结果则随着每个任务完成， 通过 as_completed 收集。 只有当每个任务的工作量足以抵消启动一个子解释器的开销时，进程池才有优势——加速比在多核桌面上最大， 在移动端较小，因为那里启动开销更高、核心也更少。

从子解释器流式上报进度

要在单个长时间任务中展示细粒度进度，可以与子解释器共享一个 Queue。 工作进程 put 进度值；一个后台线程把它们取出来驱动 UI：

import threading
from concurrent import interpreters

import flet as ft


def _is_prime(n: int) -> bool:
    """Returns True if `n` is prime."""
    if n < 2:
        return False
    for d in range(2, int(n**0.5) + 1):
        if n % d == 0:
            return False
    return True


def _count_in_range(lo: int, hi: int) -> int:
    """Returns the number of primes in the half-open range [lo, hi)."""
    count = 0
    for n in range(lo, hi):
        if _is_prime(n):
            count += 1
    return count


def stream_primes(progress_queue, chunks: int, per_chunk: int) -> None:
    """Count primes in `chunks` slices, reporting progress after each one.

    Runs in a subinterpreter, which has no access to the page — the queue is
    the only channel back to the UI. Values are fractions 0..1; a final `None`
    tells the consumer there is nothing more to read.
    """
    for i in range(chunks):
        lo = i * per_chunk + 2
        _count_in_range(lo, lo + per_chunk)
        progress_queue.put((i + 1) / chunks)
    progress_queue.put(None)  # sentinel: no more updates


def main(page: ft.Page):
    def start():
        button.disabled = True
        status.value = "Working…"
        page.update()

        # A queue shared between this interpreter and the subinterpreter. Only
        # "shareable" objects cross it (numbers, str, bytes, None, tuples of
        # those, and the queue itself).
        queue = interpreters.create_queue()

        interp = interpreters.create()
        drained = threading.Event()

        # The worker runs the job in the subinterpreter and blocks that thread,
        # so it goes on its own background thread…
        page.run_thread(work, interp, queue, drained)
        # …while a second thread drains progress and drives the UI.
        page.run_thread(drain, queue, drained)

    def work(interp, queue, drained):
        """Run the job in the subinterpreter, then close it once the UI is done.

        On its own thread because interp.call() blocks until the job finishes.
        The interpreter is closed only after `drained` is set — a subinterpreter
        Queue's pending items go invalid the moment its interpreter closes.
        """
        interp.call(stream_primes, queue, 20, 100_000)
        drained.wait()
        interp.close()

    def drain(queue, drained):
        """Forward the worker's progress reports to the UI.

        Runs on a background thread: queue.get() blocks until the worker
        reports again, so it must stay off the UI event loop.
        """
        while (value := queue.get()) is not None:
            progress.value = value
            status.value = f"Counting… {value:.0%}"
            page.update()
        drained.set()
        status.value = "Done!"
        button.disabled = False
        page.update()

    page.add(
        ft.SafeArea(
            content=ft.Column(
                controls=[
                    button := ft.Button("Start", on_click=start),
                    progress := ft.ProgressBar(value=0, width=300),
                    status := ft.Text(),
                ]
            )
        )
    )


if __name__ == "__main__":
    ft.run(main)

有两个细节值得注意：工作进程运行在它自己的线程上，因为 interp.call() 会阻塞直到任务完成；而解释器只有在 UI 取完所有条目之后才会关闭——子解释器 Queue 的待处理项 在它的解释器关闭的那一刻就会失效。

保持常驻、有状态解释器

用 interpreters.create() 创建解释器并非没有代价，因此不要每个任务都创建一个。请只创建一次并复用它： 工作进程建立的状态（此处是缓存在模块全局变量中的素数表）会在调用之间保存在该解释器中， 因此昂贵的初始化只发生在第一次调用时。

import atexit
import time
from concurrent import interpreters

import flet as ft

_UPPER = 2_000_000
_primes: list[int] | None = None


def _is_prime(n: int) -> bool:
    """Returns True if `n` is prime."""
    if n < 2:
        return False
    for d in range(2, int(n**0.5) + 1):
        if n % d == 0:
            return False
    return True


def nth_prime(n: int) -> dict:
    """Returns the n-th prime (1-indexed), building a prime table on first call.

    Runs in a long-lived subinterpreter that keeps its state across calls, so
    the table is built once (cached in a module global) and reused after that.
    The build stands in for genuinely expensive setup — loading a model, opening
    a dataset, warming a cache.
    """
    global _primes
    built = _primes is None
    if built:
        table = []
        for x in range(2, _UPPER):
            if _is_prime(x):
                table.append(x)
        _primes = table
    return {
        "prime": _primes[n - 1],
        "built_this_call": built,
        "table_size": len(_primes),
    }


def main(page: ft.Page):
    # One long-lived subinterpreter, created once and reused for every query,
    # so its cached state survives.
    interp = interpreters.create()
    atexit.register(interp.close)  # close it at exit

    def query():
        button.disabled = True
        page.update()
        page.run_thread(run)

    def run():
        """Call the interpreter on a background thread and show the result."""
        started = time.perf_counter()
        result = interp.call(nth_prime, 100_000)
        elapsed = time.perf_counter() - started
        how = "built the table" if result["built_this_call"] else "reused cache"
        status.value = f"100,000th prime = {result['prime']}\n{how} in {elapsed:.2f}s"
        button.disabled = False
        page.update()

    page.add(
        ft.SafeArea(
            content=ft.Column(
                controls=[
                    ft.Text(
                        "Click twice: the first query builds a prime table, the "
                        "second reuses it from the live interpreter."
                    ),
                    button := ft.Button("Find the 100,000th prime", on_click=query),
                    status := ft.Text(),
                ]
            )
        )
    )


if __name__ == "__main__":
    ft.run(main)

点击两次：第一次查询构建素数表（较慢），第二次从存活的解释器中复用它（瞬时完成）。在真实应用中， 这张表就相当于一个已加载的模型、一个已打开的数据集，或一个已预热的缓存。

因为这个解释器会存活整个会话，所以该示例用 atexit 在关闭时 close() 它——否则 Python 会警告有一个子解释器没有被关闭。

注意事项

• 你无法强制取消子解释器： 它运行在线程上，因此与 multiprocessing.Process 不同，没有 terminate() 可用。如果你需要中止一个失控的任务，请改用多进程。
• 并非每个 C 扩展都支持子解释器： 一个扩展必须主动选择加入（支持多阶段初始化、每个解释器拥有独立 GIL）； 一些第三方原生库目前还不支持，在子解释器中导入时会抛出 ImportError。纯 Python 代码和 标准库则没有问题。
• 复用解释器和进程池： 启动解释器是有代价的（每个都会重新导入它的模块）；请创建一个进程池或常驻解释器一次， 而不是每个任务都创建。
• 在循环或列表推导式中调用模块级辅助函数，而不要在生成器表达式中调用： 生成器表达式查找的名字不会被带入子解释器， 因此 sum(f(x) for x in ...) 若引用了一个模块级的 f，会抛出 NameError——无论是在模块中还是在 main.py 中都是如此。 普通的 for 循环或列表推导式（会被 Python 内联）则可以正常工作，本示例就是这么做的。