本文介绍如何通过定义带 `__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）：
macOS UI 自动化技能包 —— 用 Python 控制你的 Mac
下载
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
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)）。
最后，使用方式简洁一致：
立即学习“Python免费学习笔记（深入）”；
# 单例共享计数（类变量版）
counter = DeliveryCallbackCounter()
producer.produce(
topic="my-topic",
key=b"key",
value=b"value",
callback=counter
)
# 后续可随时检查状态
print(counter) # 输出：DeliveryCallbackCounter: callback count: 1综上，Python 的 functor 机制不仅消除了全局变量的副作用，还通过灵活选择实例属性或类变量，精准匹配不同状态共享需求——是构建可测试、可复用、面向对象回调逻辑的推荐实践。
