批量注册三个导出任务，执行日志却出现三次同一个任务编号；回调数量正确，也没有异常。这时先检查回调如何绑定参数，再检查调度器是否重复投递。

本文使用进程内的零参数回调作为场景：注册阶段只保存函数，执行阶段才调用。示例不连接数据库、不访问网络，也不冒充真实生产事故。

现象：注册时正确，执行时全部变了

保存下面代码为 late_binding_demo.py，使用 Python 3.10 以上运行：

from collections.abc import Callable


def build_broken(job_ids: list[str]) -> list[Callable[[], str]]:
    """复现循环结束后才读取共享循环变量的问题。"""
    callbacks = []
    for job_id in job_ids:
        callbacks.append(lambda: job_id)
    return callbacks


if __name__ == "__main__":
    callbacks = build_broken(["report-a", "report-b", "report-c"])
    print([callback() for callback in callbacks])

实际输出：

['report-c', 'report-c', 'report-c']

问题与多线程无关：这个复现只有一个线程，所有回调都在循环结束后调用。

lambda: job_id 引用外层函数中的变量。三个闭包共享这个变量对应的闭包单元，调用时才读取它的内容；循环结束后，内容是 report-c。循环每轮创建了新的函数对象，却没有为每个函数保存当轮编号的独立快照。普通 def 如果在循环中引用相同的外层变量，也会遇到同样的问题。Python 官方 FAQ 对循环中定义函数的晚绑定行为有对应说明。

排查：把“注册”和“执行”分开观察

建议按以下顺序缩小范围：

  1. 注册时记录任务编号，执行时记录实际传入业务函数的编号。结构化日志可分别使用 job_id、callback_index，固定消息使用中文，不要输出完整报表或敏感参数。
  2. 去掉定时器与线程池，将回调收集到列表，等循环结束后逐个调用。如果仍然全部指向最后一项，故障就在参数绑定阶段。
  3. 查看函数体是否直接引用循环变量，或者引用循环外被反复修改的字典。增加重试、调整 Worker 数量都不能修复这类绑定错误。
  4. 测试至少使用三个不同编号。只检查“执行三次”会漏掉“同一任务执行三次”。

不要在注册循环里立刻调用回调来证明没有问题。那时变量尚未进入下一轮，测试时序恰好掩盖缺陷。

修复：注册时绑定不可变的业务编号

对于已有的具名业务函数，使用 functools.partial 绑定当前编号最直观。它保存调用所需的参数，之后调用回调才执行业务函数。官方文档说明了 partial 的参数绑定规则。

下面是完整实现与回归测试，保存为 callback_binding.py。这里的 sink 是可替换的执行边界：测试传入列表的 append，实际应用可以接入自己的导出服务。实现仅接受最多 1000 个、格式受限的字符串编号；这是演示的输入约定，接入生产时应使用真实业务规则。

import re
import unittest
from collections.abc import Callable
from functools import partial


def export_report(job_id: str, sink: Callable[[str], None]) -> None:
    """通过注入的执行边界处理一个已校验的任务编号。"""
    sink(job_id)


def build_callbacks(
    job_ids: list[str], sink: Callable[[str], None]
) -> list[Callable[[], None]]:
    """先完整校验，再为每个编号生成延迟执行的回调。"""
    if not isinstance(job_ids, list) or len(job_ids) > 1000:
        raise ValueError("任务编号必须是最多 1000 项的列表")
    if not callable(sink):
        raise TypeError("执行边界必须可调用")
    for job_id in job_ids:
        if not isinstance(job_id, str) or re.fullmatch(
            r"[A-Za-z0-9][A-Za-z0-9_-]{0,63}", job_id
        ) is None:
            raise ValueError("任务编号必须是 1 至 64 位字母、数字、下划线或短横线，首位为字母或数字")
    return [partial(export_report, job_id, sink) for job_id in job_ids]


class CallbackBindingTests(unittest.TestCase):
    """用延迟执行验证编号绑定，而非只检查调用次数。"""

    def test_each_job_keeps_its_identity(self) -> None:
        seen: list[str] = []
        callbacks = build_callbacks(["report-a", "report-b", "report-c"], seen.append)
        for callback in callbacks:
            callback()
        self.assertEqual(seen, ["report-a", "report-b", "report-c"])

    def test_registration_does_not_execute(self) -> None:
        seen: list[str] = []
        callbacks = build_callbacks(["report-a"], seen.append)
        self.assertEqual(len(callbacks), 1)
        self.assertEqual(seen, [])

    def test_source_list_change_does_not_retarget(self) -> None:
        seen: list[str] = []
        job_ids = ["report-a", "report-b"]
        callbacks = build_callbacks(job_ids, seen.append)
        job_ids[:] = ["report-z"]
        for callback in callbacks:
            callback()
        self.assertEqual(seen, ["report-a", "report-b"])

    def test_reverse_execution_keeps_binding(self) -> None:
        seen: list[str] = []
        callbacks = build_callbacks(["report-a", "report-b"], seen.append)
        for callback in reversed(callbacks):
            callback()
        self.assertEqual(seen, ["report-b", "report-a"])

    def test_empty_input(self) -> None:
        seen: list[str] = []
        self.assertEqual(build_callbacks([], seen.append), [])

    def test_invalid_ids_fail_before_execution(self) -> None:
        seen: list[str] = []
        for invalid in ("", "../report", "a" * 65):
            with self.subTest(invalid=invalid):
                with self.assertRaises(ValueError):
                    build_callbacks(["report-a", invalid], seen.append)
        self.assertEqual(seen, [])

    def test_batch_limit(self) -> None:
        seen: list[str] = []
        self.assertEqual(len(build_callbacks(["a"] * 1000, seen.append)), 1000)
        with self.assertRaises(ValueError):
            build_callbacks(["a"] * 1001, seen.append)

    def test_sink_error_is_propagated(self) -> None:
        def fail(job_id: str) -> None:
            raise RuntimeError("模拟导出失败")

        callback = build_callbacks(["report-a"], fail)[0]
        with self.assertRaisesRegex(RuntimeError, "模拟导出失败"):
            callback()

    def test_default_argument_also_binds_strings(self) -> None:
        callbacks = [lambda job_id=job_id: job_id for job_id in ["a", "b"]]
        self.assertEqual([callback() for callback in callbacks], ["a", "b"])


if __name__ == "__main__":
    unittest.main(verbosity=2)

运行：

python .\late_binding_demo.py
python .\callback_binding.py

关键断言是编号序列等于 report-a、report-b、report-c。反序执行测试进一步确认，每个回调绑定自己的任务，而不是依赖执行顺序。

lambda job_id=job_id: job_id 也能绑定当前字符串：右侧默认值在函数创建时求值。它适合简单表达式；具名业务函数配合 partial 通常更容易定位调用关系。若框架会向回调传入事件参数，要按框架规定设计签名；这个零参数例子不能原样用于所有回调接口。

证明测试能挡住原缺陷

把 build_callbacks 的最后一行临时改为下面的错误版本，再运行同一套测试：

return [lambda: export_report(job_id, sink) for job_id in job_ids]

虽然列表推导式有自己的作用域，其中生成的闭包仍共享该作用域里的 job_id。三项不同编号最终都会使用最后一个编号。test_each_job_keeps_its_identity 必须失败；如果只断言长度为三，错误实现仍会通过。

本次在 Windows、Python 3.14.6 上执行了两个完整示例：原始复现输出三个 report-c，修复后的九项 unittest 全通过；恢复上述错误实现后，三个编号绑定测试失败，未发生测试执行错误。未安装 Ruff，因此未执行 Ruff 格式化和静态检查，也未为本文安装额外依赖。

上线边界：绑定参数不等于复制对象

partial 和默认参数都保存对象引用，不会自动深拷贝对象。本例绑定字符串，所以修改原列表或替换其中元素不会改变已绑定的编号。如果绑定的是一个不断被复用和修改的字典，延迟回调仍可能读到后来写入的内容。建议提取不可变的业务编号，执行时通过明确的服务边界读取数据；读取的是注册时快照还是执行时最新数据，必须由业务契约决定。

还要分别处理下面几件事：

  • 参数校验不能代替权限校验。多租户服务需要绑定并校验租户身份，执行时仍要检查访问权限。
  • 正确绑定不会自动去重。本文保留重复编号，重复投递、失败重试和幂等仍由任务系统处理。
  • 回调异常继续向调用方传播。批处理应明确中止、继续或重试策略，不能捕获后直接记成功。
  • 绑定的 sink 也会被回调持有，应随任务生命周期释放回调，避免长期引用业务对象。
  • 分布式任务应传输明确的任务名和可序列化参数。不要据此假设进程内闭包或 partial 能直接跨进程、跨机器调度。

本文没有验证真实定时器、线程池、Celery、权限系统、持久队列及跨进程序列化。先在实际框架中核对回调签名和失败策略，再灰度替换注册代码。若需回退，应暂停新任务注册并排查已登记回调；已经生成的错误闭包不会因代码文件更新而自动修复。对已误执行的任务，先核对业务编号与结果，再按幂等规则补偿，避免整批盲目重跑。

总结

批量回调全部处理最后一项，通常要检查变量在什么时候被读取。注册时绑定不可变编号，执行时调用业务函数，再用不同编号、延迟调用和反序执行做回归测试，才能同时验证任务数量与任务身份。