问题
在基于 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 参数，时间窗口越大，则平均排队时间变化曲线越平滑。另外，为了降低对内存的消耗，每次记录入队或出队时间的时候，会清理一天之前的入队出队时间记录。
网友评论