本文介绍如何在 python 中通过定义带 `__call__` 方法的类来替代全局变量回调，实现线程安全、可复用且状态隔离的回调逻辑，特别适用于 kafka 等异步生产者场景。
在异步消息系统（如 Apache Kafka）中，生产者常需传入回调函数以处理消息投递结果。传统做法是使用带全局变量的普通函数，例如：
callback_count = 0
def delivery_callback(error, message_payload):
if error:
print(f'ERROR: {error}')
else:
global callback_count
callback_count += 1这种方式虽简单，但存在明显缺陷：全局状态污染、不可重入、难以单元测试、多线程下非线程安全，且无法为不同生产者实例维护独立计数。
更优解是使用 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 get_count(self):
return self.count_callback
def __str__(self):
return f'DeliveryCallbackCounter(count={self.count_callback})'使用时只需实例化并传入：
counter_a = DeliveryCallbackCounter() counter_b = DeliveryCallbackCounter() producer_a.produce(topic="logs", value=b"msg1", callback=counter_a) producer_b.produce(topic="alerts", value=b"msg2", callback=counter_b) # 各自独立计数，互不影响 print(counter_a.get_count()) # 可能为 1 print(counter_b.get_count()) # 可能为 1
✅ 优势显著：
- ✅ 状态封装在对象内，避免全局污染；
- ✅ 支持多实例并行，天然适配多生产者或多主题场景；
- ✅ 易于扩展（如添加日志、错误统计、超时监控等）；
- ✅ 可直接继承、mock 或注入依赖，大幅提升可测试性。
⚠️ 注意：若你确实需要所有实例共享同一计数器（即模拟“静态全局语义”），可改用类变量（class variable），而非实例变量：
class SharedDeliveryCallbackCounter:
count_callback = 0 # 类变量，所有实例共享
def __call__(self, error, message):
if not error:
SharedDeliveryCallbackCounter.count_callback += 1
@classmethod
def reset(cls):
cls.count_callback = 0
@classmethod
def get_count(cls):
return cls.count_callback此时无论创建多少实例，count_callback 均为同一份内存地址的数据。但需谨慎使用——它本质上仍是全局状态，仅语法上更清晰，不解决线程安全问题。如需并发安全，应配合 threading.Lock 或使用 atomic 操作（如 threading.local() 或 concurrent.futures 工具）。
总结：Python 中的 functor 是替代全局回调的理想模式。优先使用实例变量实现状态隔离；仅当业务明确要求跨实例共享状态时，才考虑类变量，并务必评估并发与可维护性风险。
