在本食谱中，你将学习如何在 Flet 应用中使用 Python 内置的 multiprocessing 模块（包括 concurrent.futures.ProcessPoolExecutor）， 实现跨进程的真正 CPU 并行。

对于 I/O 密集型任务，或者只是需要脱离 UI 线程的工作，优先使用 异步或线程——它们更轻量，并且在所有平台都能运行。 当你需要多个 CPU 核心同时执行 Python 任务（数值计算、批处理、机器学习推理等）， 或者需要为可能失败或需要停止的任务提供进程隔离时，才使用 multiprocessing。 关于单进程内的多核 Python——包括 multiprocessing 无法运行的移动端——请见 子解释器（Python 3.14+）。

::: important 平台与 Flet 版本支持 multiprocessing 在开发阶段的 Flet 桌面应用中（flet run）以及使用 Flet v0.86.0 或更高版本、通过 flet build {macos,windows,linux} 或 flet debug {macos,windows,linux} 打包的桌面应用中可用。

它不支持 iOS 和 Android（移动操作系统不允许应用随意启动子进程），也不支持浏览器。 在这些平台上，I/O 密集型任务使用线程或 asyncio，CPU 密集型任务使用 子解释器（Python 3.14+）——它们在进程内运行，因此能在移动端跨核心并行执行 Python 代码，而 multiprocessing 做不到。 :::

它是如何工作的？

在使用 flet build 打包的桌面应用中，没有独立的 python 可执行文件—— 解释器被嵌入在你的应用二进制文件中。当 multiprocessing 派生一个工作进程时，它会用一个 CPython 辅助命令行重新执行该二进制文件；该二进制文件识别出这种形态，并以一个普通的、无窗口的 Python 解释器来提供服务。这也覆盖了 multiprocessing 的辅助进程（资源追踪器和 forkserver）。

准则

这些是标准的 Python multiprocessing 指南 ——但在打包后的 Flet 应用中，它们是强制要求，而不仅仅是好的风格。

始终守卫入口点

只在 if __name__ == "__main__": 守卫下启动你的应用。例如：

import flet as ft

def main(page: ft.Page):
    ...

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

使用 spawn 和 forkserver启动方式 时，工作进程/辅助进程需要安全地导入你的主模块。spawn 在 macOS 和 Windows 上是默认方式； 从 Python 3.14 起，forkserver 在 Linux 上是默认方式。如果没有守卫，子进程会尝试再次启动你的整个应用。

使用可导入、可序列化的工作函数

工作目标、参数和返回值必须是可 pickle 的（picklable），这样 Python 才能在进程之间发送它们。实践中：

• 在模块顶层定义工作函数，而不是在 main() 内部或按钮事件处理器内部
• 传递简单的数据，如数字、字符串、列表、字典或数据类
• 不要传递 Flet 控件、page、数据库连接、打开的文件、lambda 或嵌套函数

推荐：

def sort_chunk(chunk):
    return sorted(chunk)

避免：

def main(page: ft.Page):
    def sort_chunk(chunk):
        return sorted(chunk)

嵌套版本不可靠地可序列化，因为工作进程需要按名称从模块导入该函数。

不要从工作进程操作 GUI

工作进程运行在独立的解释器中，与你的应用页面没有连接。通过 multiprocessing.Queue、 Pipe 或 进程池的 future 把数据传回，并在主进程中更新 UI。

其他

• 打包应用中的 sys.executable 指向你的应用二进制文件，而不是 python 可执行文件。这是有意为之——不要用 multiprocessing.set_executable() 去覆盖它。
• 在 Flet 应用中，你通常不需要 multiprocessing.freeze_support()。在 if __name__ == "__main__": 代码块中调用它并无害处，但 Flet 并不依赖 PyInstaller 风格的冻结可执行文件引导机制。
• 工作进程的 print() 输出不会连接到你的应用控制台日志；如果需要工作进程的诊断信息，请使用 Queue 或基于文件的日志。
• 在 Linux 上，避免强制使用 fork 启动方式：你的应用进程运行着 Flutter 引擎，并带有许多活跃线程，对它执行 fork 是不安全的。优先使用平台默认方式（通常是 forkserver 或 spawn），或者通过 multiprocessing.set_start_method() 或 multiprocessing.get_context() 显式指定。
• 在打包应用中启动工作进程比在普通 Python 中开销更大——每个派生的子进程在 Python 接管前都要先加载你的应用二进制库的库文件。请一次性创建进程池和长期存活的工作进程并复用它们，而不是每次点击按钮都派生（见保持常驻工作进程示例）。

示例

带实时进度的并行排序

跨所有 CPU 核心对数据分块排序，并将进度实时推送到页面：

import random
from concurrent.futures import ProcessPoolExecutor, as_completed

import flet as ft


def sort_chunk(chunk: list[float]) -> list[float]:
    """Sort one chunk of data."""
    return sorted(chunk)


def main(page: ft.Page):
    def run_sort():
        """Orchestrate the pool from a background thread, updating the UI
        from the main process as results come in."""
        chunks = [[random.random() for _ in range(250_000)] for _ in range(8)]
        completed = 0
        # Without max_workers, the pool sizes itself to the machine's CPUs.
        with ProcessPoolExecutor() as pool:
            futures = [pool.submit(sort_chunk, c) for c in chunks]
            # as_completed() yields each future as soon as its worker is done.
            for _ in as_completed(futures):
                completed += 1
                progress.value = completed / len(futures)
                status.value = f"Sorted {completed}/{len(futures)} chunks"
                page.update()
        status.value = "Done!"
        page.update()

    page.add(
        ft.SafeArea(
            content=ft.Column(
                controls=[
                    ft.Button(
                        "Start sorting", on_click=lambda: page.run_thread(run_sort)
                    ),
                    status := ft.Text("Idle"),
                    progress := ft.ProgressBar(value=0, width=300),
                ]
            )
        )
    )


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

注意，长时间运行的整体编排通过 page.run_thread 从 UI 事件处理器中移出，而 CPU 密集的工作则在进程池中运行。 工作函数可以放在你的主模块中（如上所示），也可以放在独立的、可导入的模块里——两者都可行。

从工作进程流式上报进度

要在单个长时间运行的任务内部展示细粒度进度，可以把一个 multiprocessing.Queue 传递给工作进程，并在后台线程中消费它。工作进程用 put() 上报进度值， 并在结束时放入一个 None 哨兵值：

import multiprocessing

import flet as ft


def crunch(progress_queue: multiprocessing.Queue, steps: int) -> None:
    """Do `steps` slices of CPU-heavy work, reporting progress after each one.

    Runs in a worker process, 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(steps):
        sum(range(50_000_000))  # one slice of real work
        progress_queue.put((i + 1) / steps)
    progress_queue.put(None)  # sentinel: no more updates


def main(page: ft.Page):
    def start():
        button.disabled = True
        status.value = "Starting worker…"
        page.update()
        queue = multiprocessing.Queue()
        worker = multiprocessing.Process(target=crunch, args=(queue, 20), daemon=True)
        worker.start()
        page.run_thread(drain, queue, worker)

    def drain(queue: multiprocessing.Queue, worker: multiprocessing.Process):
        """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"Crunching… {value:.0%}"
            page.update()
        worker.join()
        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)

保持常驻工作进程

启动一个进程并非没有代价——尤其是在打包应用中，每个子进程都要先加载你的应用二进制库的库文件。 对于重复性的任务，可以启动一个保持存活、通过 Pipe 提供服务的工作进程，只支付一次启动开销。昂贵的初始化（加载模型、打开数据集）也只需在 worker 中做一次：

import multiprocessing
import time

import flet as ft


def calc_worker(conn) -> None:
    """Serve calculation jobs over the pipe until told to stop.

    Runs in a worker process that stays alive across jobs, so the process
    startup cost is paid once. Expensive setup (loading models, opening
    datasets, warming caches) could be done once here, before the loop,
    and reused by every job. Receiving `None` is the shutdown signal.
    """
    while (job := conn.recv()) is not None:
        started = time.perf_counter()
        result = sum(i * i for i in range(job))
        conn.send((result, time.perf_counter() - started))
    conn.close()


def main(page: ft.Page):
    # One end of the pipe goes to the worker, the other stays with the UI.
    parent_conn, child_conn = multiprocessing.Pipe()
    worker = multiprocessing.Process(
        target=calc_worker, args=(child_conn,), daemon=True
    )
    worker.start()

    def submit():
        # One job in flight at a time: the button stays disabled until the
        # worker replies (a bare Pipe is not multiplexed).
        button.disabled = True
        status.value = "Working…"
        page.update()
        page.run_thread(request)

    def request():
        """Send a job to the worker and wait for its reply.

        Should be run on a background thread: conn.recv() blocks until the worker
        answers, so it must stay off the UI event loop.
        """
        parent_conn.send(100_000_000)
        result, duration = parent_conn.recv()
        status.value = (
            f"sum of squares = {result}\n{duration:.2f}s in process {worker.pid}"
        )
        button.disabled = False
        page.update()

    page.add(
        ft.SafeArea(
            content=ft.Column(
                controls=[
                    button := ft.Button("Compute in worker", on_click=submit),
                    status := ft.Text(f"Worker ready (pid {worker.pid})"),
                ]
            )
        )
    )


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

该工作进程以 daemon=True 启动，因此当你的应用退出时会自动终止。

取消失控的任务

线程无法从外部强制停止——而工作进程 Process 可以随时通过 terminate() 停止。这使得进程成为适合可能需要中断的任务的工具，例如对外部库的长耗时调用：

import multiprocessing
import time

import flet as ft


def long_job(seconds: int) -> None:
    """Burn CPU for roughly `seconds` seconds.

    Stands in for work you cannot interrupt from Python itself, e.g. a tight
    loop inside a C extension library. Runs in a worker process, which is why
    it can be cancelled at any point — threads can't be stopped like that.
    """
    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        sum(range(1_000_000))


def main(page: ft.Page):
    process = None

    def watch(p: multiprocessing.Process):
        """Wait (on a background thread) for the worker to end, then report
        whether it finished on its own or was cancelled."""
        p.join()
        if p.exitcode == 0:
            status.value = "Process finished normally."
        else:
            # terminate() ends the child with a negative exit code (-SIGTERM).
            status.value = f"Process was terminated (exit code {p.exitcode})."
        start.disabled = False
        cancel.disabled = True
        page.update()

    def start_job():
        nonlocal process
        process = multiprocessing.Process(target=long_job, args=(30,), daemon=True)
        process.start()
        status.value = f"Running in process {process.pid}…"
        start.disabled = True
        cancel.disabled = False
        page.update()
        page.run_thread(watch, process)

    def cancel_job():
        if process is not None and process.is_alive():
            process.terminate()  # threads can't do this

    page.add(
        ft.SafeArea(
            content=ft.Column(
                controls=[
                    ft.Row(
                        controls=[
                            start := ft.Button("Start 30s job", on_click=start_job),
                            cancel := ft.Button(
                                "Cancel", on_click=cancel_job, disabled=True
                            ),
                        ]
                    ),
                    status := ft.Text(),
                ]
            )
        )
    )


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

一个后台线程用 join() 等待工作进程并报告它如何结束——正常结束（退出码 0）或经由取消（负的退出码）。

INFO

Python 的 multiprocessing 指南 建议避免终止进程，或者仅对从不使用任何共享资源的进程这样做。