如何在Python中统计指定时长内的操作执行次数？含循环场景优化问询

Great question! Storing every single timestamp in a list works, but it can get memory-heavy if your loop runs for hours or days. Let's quickly cover the basic approach you mentioned (for context), then dive into more efficient alternatives that avoid keeping every single entry around:

先快速过一下基础列表法（对比用） #

这是你一开始想到的方式：每次执行foo()就记录当前时间，检查时过滤出最近60秒的时间戳并计数。但它的问题在于，列表会随着运行时间无限增长，清理旧数据的pop(0)操作也是O(n)的，效率不高：

import time

timestamps = []
window_seconds = 60

def check_number_of_foo_executed_last_60_sec():
    current_time = time.time()
    # 过滤出60秒内的时间戳
    valid_timestamps = [t for t in timestamps if current_time - t <= window_seconds]
    timestamps[:] = valid_timestamps  # 更新列表只保留有效数据
    print(f"最近60秒执行了{len(timestamps)}次foo()")

for i in range(1, 9999):
    foo(i)
    timestamps.append(time.time())
    check_number_of_foo_executed_last_60_sec()

更高效的替代方案 #

1. 用双端队列（collections.deque）实现滑动窗口 #

deque的popleft()操作是O(1)的，比列表的pop(0)快太多。我们可以维护一个只保留最近60秒时间戳的队列，每次检查时先清理过期数据，队列长度就是执行次数：

import time
from collections import deque

timestamps = deque()
window_seconds = 60

def check_number_of_foo_executed_last_60_sec():
    current_time = time.time()
    # 移除队列中超过60秒的旧时间戳
    while timestamps and current_time - timestamps[0] > window_seconds:
        timestamps.popleft()
    print(f"最近60秒执行了{len(timestamps)}次foo()")

for i in range(1, 9999):
    foo(i)
    timestamps.append(time.time())
    check_number_of_foo_executed_last_60_sec()

这个方法内存占用可控，实现也简单，是大多数场景下的最优选择。

2. 分桶计数法（适合高频率执行场景） #

如果foo()每秒执行上百次，存储每个时间戳还是有点浪费。可以按时间片分桶（比如每1秒一个桶），记录每个桶内的执行次数，大幅减少存储的数据量：

import time
from collections import defaultdict

window_seconds = 60
# 键是秒级时间片的起始时间，值是该时间片内的执行次数
bucket_counts = defaultdict(int)

def check_number_of_foo_executed_last_60_sec():
    current_time = time.time()
    window_start = current_time - window_seconds
    total = 0
    to_delete = []
    
    # 遍历桶，累加有效次数并标记过期桶
    for bucket_time, count in bucket_counts.items():
        if bucket_time < window_start:
            to_delete.append(bucket_time)
        else:
            total += count
    # 删除过期桶释放内存
    for t in to_delete:
        del bucket_counts[t]
    print(f"最近60秒执行了{total}次foo()")

for i in range(1, 9999):
    foo(i)
    current_bucket = int(time.time())
    bucket_counts[current_bucket] += 1
    check_number_of_foo_executed_last_60_sec()

缺点是统计结果有时间片级的误差（比如上面是1秒误差），但可以通过缩小时间片（比如0.1秒）来提升精度，同时仍然保持低内存占用。

3. 环形缓冲区法（内存严格受限场景） #

如果你的程序需要长期运行，且内存配额非常紧张，可以用固定大小的环形数组来实现滑动窗口，内存占用永远固定：

import time

window_seconds = 60
# 环形缓冲区，每个位置对应1秒的执行计数，初始全0
ring_buffer = [0] * window_seconds
current_pos = 0
last_second = int(time.time())

def check_number_of_foo_executed_last_60_sec():
    global current_pos, last_second
    current_time = int(time.time())
    
    # 如果当前时间和上次记录的时间差超过1秒，填充中间的空桶为0
    while current_time - last_second > 0:
        last_second += 1
        current_pos = (current_pos + 1) % window_seconds
        ring_buffer[current_pos] = 0
    
    # 缓冲区所有元素的和就是最近60秒的执行次数
    total = sum(ring_buffer)
    print(f"最近60秒执行了{total}次foo()")

for i in range(1, 9999):
    foo(i)
    current_time = int(time.time())
    
    # 跨时间片时更新指针和桶
    if current_time != last_second:
        last_second = current_time
        current_pos = (current_pos + 1) % window_seconds
        ring_buffer[current_pos] = 0
    ring_buffer[current_pos] += 1
    check_number_of_foo_executed_last_60_sec()

这个方法的内存占用永远是window_seconds个整数的大小，适合后台守护进程这类长期运行的任务，但实现稍复杂，同样有时间片精度限制。

快速选型指南 #

• 执行频率不高、需要精准统计：选双端队列法
• 执行频率极高、内存敏感：选分桶计数法
• 长期运行、内存严格受限：选环形缓冲区法

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