如果我有一个以回调函数为参数的函数,如果回调函数带有一些参数集,我如何创建一个函数器对象来代替回调函数?

Context:

卡夫卡制作人接受回调函数.

producer.produce(
  topic=topic,
  key=key,
  value=value,
  callback=delivery_callback
)
callback_count = 0

def delivery_callback(error, message_payload):
    if error:
        print(f'{error}')
    else:
        global callback_count
        callback_count += 1
        # count the number of successfully
        # delivered messages

此回调函数统计成功传递的消息数.

Problems with this design and proposed solution

这种设计的问题是显而易见的--使用全局变量.

我的第一个本能是设计某种仿函数对象,它可能会定义__call__使其可调用.

class DeliveryCallbackCounter():

    def __init__(self) -> None:
        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) -> str:
        return f'DeliveryCallbackCounter: callback count: {self.count_callback}'

这可以通过以下方式使用,并且确实有效.

delivery_callback_counter = DeliveryCallbackCounter()

producer.produce(
  topic=topic,
  key=key,
  value=value,
  callback=delivery_callback_counter
)

但是,这个解决方案的工作方式与前面的选项不完全相同,因为类DeliveryCallbackCounter不是静态的.以前,当使用函数而不是函子(类)作为回调函数时,修改数据count_callback是具有静态生命周期 的全局变量.

有了函数器,我们就有了一些稍有不同的东西,并且可以创建类DeliveryCallbackCounter的许多实例化.

Python类有一些类似于静态方法的东西,由@classmethoddecorator 表示.

有没有可能通过在Python中构建一个"静态类"来重现"函数选项"所提供的相同语义?我对Python不太熟悉,不知道该怎么做.

推荐答案

您可以使用类变量而不是全局变量:

class DeliveryCallbackCounter:
    count_callback = 0

    def __init__(self) -> None:
        pass

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

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

Python相关问答推荐

具有多个选项的计数_匹配

更改matplotlib彩色条的字体并勾选标签?

Deliveryter Notebook -无法在for循环中更新matplotlib情节(保留之前的情节),也无法使用动画子功能对情节进行动画

ModuleNotFound错误:没有名为flags.State的模块; flags不是包

计算组中唯一值的数量

如何请求使用Python将文件下载到带有登录名的门户网站?

组/群集按字符串中的子字符串或子字符串中的字符串轮询数据框

实现自定义QWidgets作为QTimeEdit的弹出窗口

计算每个IP的平均值

如何合并两个列表,并获得每个索引值最高的列表名称?

如何指定列数据类型

lityter不让我输入左边的方括号,'

为什么if2/if3会提供两种不同的输出?

关于两个表达式的区别

如何按row_id/row_number过滤数据帧

删除特定列后的所有列

简单 torch 模型测试:ModuleNotFoundError:没有名为';Ultralytics.yolo';

如何使用加速广播主进程张量?

如何批量训练样本大小为奇数的神经网络?

多个布尔条件的`jax.lax.cond`等效项