Python中获取连续函数调用链的最后一次调用以优化性能

现有如下代码，Foo类的update_something方法会被短时间内重复调用数百次，每次调用都会执行逻辑，带来不必要的性能损耗。我们需要实现：仅执行该方法最后一次调用的逻辑，忽略之前的所有重复调用。

原代码 #

class Foo:

    def __init__(self, name):
        self.name = name

    def update_something(self, number):
        """
        该方法负责更新对象状态，会被同时调用数百次
        因此仅执行最后一次调用更新状态即可。
        """
        if number != 100:
            print(f"This method has been called for: ({number}) of times ---> [{self.name}]")
        else:
            print(f"This is the Final call ---> [{self.name}]")


if __name__ == '__main__':
    obj1 = Foo("obj1")
    obj2 = Foo("obj2")
    obj3 = Foo("obj3")
    obj4 = Foo("obj4")
    list_of_objects = [obj1, obj2, obj3, obj4]

    # 模拟短时间内重复调用
    for i in range(101):
        print("---")
        for obj in list_of_objects:
            obj.update_something(i)

当前输出 #

This method has been called for: (0) of times ---> [obj1]
This method has been called for: (0) of times ---> [obj2]
This method has been called for: (0) of times ---> [obj3]
This method has been called for: (0) of times ---> [obj4]
...
This is the Final call ---> [obj1]
This is the Final call ---> [obj2]
This is the Final call ---> [obj3]
This is the Final call ---> [obj4]

优化方案：实现防抖机制 #

通过给每个Foo实例添加定时器，每次调用方法时取消之前未触发的定时器，仅保留最后一次调用的定时器，待延迟结束后执行最终逻辑。

修改后的代码 #

import threading

class Foo:

    def __init__(self, name):
        self.name = name
        self.timer = None  # 用于存储定时器实例

    def _actual_update(self, number):
        """实际执行更新逻辑的内部方法"""
        if number != 100:
            print(f"This method has been called for: ({number}) of times ---> [{self.name}]")
        else:
            print(f"This is the Final call ---> [{self.name}]")

    def update_something(self, number):
        """
        防抖处理：短时间内重复调用时，仅执行最后一次调用的逻辑
        """
        # 取消之前的定时器（如果存在）
        if self.timer is not None:
            self.timer.cancel()
        
        # 创建新的定时器，延迟0.1秒执行实际逻辑
        # 延迟时间可根据实际场景调整，确保覆盖所有重复调用的时间段
        self.timer = threading.Timer(0.1, self._actual_update, args=(number,))
        self.timer.start()


if __name__ == '__main__':
    obj1 = Foo("obj1")
    obj2 = Foo("obj2")
    obj3 = Foo("obj3")
    obj4 = Foo("obj4")
    list_of_objects = [obj1, obj2, obj3, obj4]

    # 模拟短时间内重复调用
    for i in range(101):
        for obj in list_of_objects:
            obj.update_something(i)
    
    # 等待定时器执行完成
    input("Press Enter to exit...")

期望输出 #

This is the Final call ---> [obj1]
This is the Final call ---> [obj2]
This is the Final call ---> [obj3]
This is the Final call ---> [obj4]

说明 #

• 防抖延迟时间0.1秒可根据实际调用频率调整，需确保该时间段内能覆盖所有重复调用。
• 使用threading.Timer实现异步延迟执行，避免阻塞主线程。
• 每个Foo实例独立维护自己的定时器，互不干扰。

内容的提问来源于stack exchange，提问作者Uncle Bob