基于 Redis 的任务排队时间统计

ゞ浴缸里的玫瑰 2022-08-23 12:53 14阅读 0赞

## 问题 ##

在[基于 Redis 的优先级队列][Redis]这篇文章中，我们使用 sorted set 实现了一个优先级队列。现在，考虑如下问题：

> 在一个任务调度系统中，任务进入队列排队，然后等待被调度出队列。为了衡量调度效率，一个指标是任务的平均排队时间。那么，如何统计每个任务的排队时间以及这些任务的平均排队时间呢？

## 解决方案 ##

为了统计每个任务的排队时间，我们需要知道任务的入队时间以及出队时间，这两个时间的差值就是这个任务的排队时间。我们仍然使用 sorted set 来记录任务的入队或出队时间，只不过这个时候 sorted set 中元素的 score 不再是优先级，而是任务的入队或出队时间。

## 代码实现 ##

下面是一个可能的代码实现：

class QueuingTimeStat(object):
    
        def __init__(self, redis_client, redis_key, namespace='schedule'):
            self.client = redis_client
            self.key = '%s:%s' % (namespace, redis_key)
    
        def _clean(self):
            """清理一天之前的记录"""
            now = time.time()
            self.client.zremrangebyscore(self.key, 0, now - 3600 * 24)
    
        def add_start(self, task_id, timestamp=None):
            self._clean()
            member = '%s:%s' % (task_id, 'start')
            timestamp = time.time() if timestamp is None else timestamp
            self.client.zadd(self.key, timestamp, member)
    
        def add_end(self, task_id, timestamp=None):
            self._clean()
            member = '%s:%s' % (task_id, 'end')
            timestamp = time.time() if timestamp is None else timestamp
            self.client.zadd(self.key, timestamp, member)
    
        def average_delay(self, window):
            now = time.time()
            results = self.client.zrangebyscore(self.key, now - window, now, withscores=True)
            # [(value, score), ...]
            sorted_results = sorted(results, key=lambda x: x[0])
            group_results = groupby(sorted_results, key=lambda x: x[0].rsplit(':', 1)[0])
            total = 0.0
            count = 0
            for task_id, sub_group in group_results:
                pairs = tuple(sub_group)
                if len(pairs) == 2:
                    count += 1
                    total += abs(pairs[0][1] - pairs[1][1])
                else:
                    if pairs[0][0].endswith('start'):
                        count += 1
                        total += now - pairs[0][1]
            return total / count if count > 0 else 0

在上面的代码中，task\_id 是任务的唯一标识。task\_id:start 用于记录任务的入队时间，task\_id:end 用于记录任务的出队时间。average\_delay 用于统计最近任务的平均排队时间，该方法提供了时间窗口 window 参数，时间窗口越大，则平均排队时间变化曲线越平滑。另外，为了降低对内存的消耗，每次记录入队或出队时间的时候，会清理一天之前的入队出队时间记录。

[Redis]: https://www.jianshu.com/p/9a455194bef6