Python 中使用可调用类（Functor）实现带状态的回调函数

霞舞 2026-01-25 00:00:00 次阅读

本文介绍如何通过定义带 `__call__` 方法的类来替代全局变量，安全、清晰地实现有状态的 kafka 消息投递回调计数器，并对比类变量与实例变量在共享状态场景下的适用性。

在 Python 中，当第三方库（如 confluent-kafka）要求传入一个回调函数（如 callback=delivery_callback），而该回调需维护内部状态（例如成功投递消息的数量）时，直接使用全局变量虽能工作，但存在可维护性差、线程不安全、测试困难等明显缺陷。更优雅的替代方案是使用可调用对象（functor）——即实现了 __call__ 方法的类实例。

最直观的做法是将计数器作为实例属性：

class DeliveryCallbackCounter:
    def __init__(self):
        self.count_callback = 0

    def __call__(self, error, message):
        if error:
            print(f'ERROR: Kafka: Message delivery failure: {error}')
        else:
            self.count_callback += 1

    def __str__(self):
        return f'DeliveryCallbackCounter: callback count: {self.count_callback}'

此设计支持多实例隔离——每个 DeliveryCallbackCounter() 实例拥有独立计数器，适用于需要按主题、分区或业务上下文分别统计的场景。但若目标是完全复现原全局函数的行为（即所有调用共享同一计数器，无论创建多少个 functor 实例），则应改用类变量（class variable）：

class DeliveryCallbackCounter:
    count_callback = 0  # ← 类变量：所有实例共享

    def __call__(self, error, message):
        if error:
            print(f'ERROR: Kafka: Message delivery failure: {error}')
        else:
            DeliveryCallbackCounter.count_callback += 1  # ← 显式通过类名访问

    def __str__(self):
        return f'DeliveryCallbackCounter: callback count: {DeliveryCallbackCounter.count_callback}'

✅ 关键点说明：类变量 count_callback 在类定义时初始化一次，被所有实例共享，语义上等价于原全局变量；推荐显式使用 ClassName.attribute（而非 self.__class__.attribute 或 type(self).attribute）访问类变量，避免继承歧义，提升可读性与健壮性；若需线程安全（如高并发 Kafka 生产者场景），应配合 threading.Lock 或 atomic 操作（如 threading.local() 或 concurrent.futures 工具），类变量本身不提供线程安全性； @classmethod 不适用于本场景——它定义的是“类方法”，接收 cls 参数，无法直接作为回调函数被 Kafka 库调用（因其签名需严格匹配 (error, message)）。

最后，使用方式简洁一致：

# 单例共享计数（类变量版）
counter = DeliveryCallbackCounter()

producer.produce(
    topic="my-topic",
    key=b"key",
    value=b"value",
    callback=counter
)

# 后续可随时检查状态
print(counter)  # 输出：DeliveryCallbackCounter: ca
llback count: 1