store.py 文件源码

def reschedule(self, now=None):
        now = now or time()
        items, _, size = (self.client.pipeline()
                          .zrangebyscore(SCHEDULE_KEY, '-inf', now)
                          .zremrangebyscore(SCHEDULE_KEY, '-inf', now)
                          .zcard(SCHEDULE_KEY)
                          .execute())

        for chunk in iter_chunks(items, 5000):
            pipe = self.client.pipeline(False)
            for r in chunk:
                queue, _, task = r.partition(b':')
                pipe.rpush(rqname(queue), task)
            pipe.execute()

        return size