Python redis 模块，keys() 实例源码

def Queue(self, queue_type="python_queue", name=None, **kwargs): #???????????
        if queue_type not in ["python_queue", "redis_queue"]: #????
            raise Exception(queue_type + " wrong")
        if queue_type == "redis_queue" and redis_enable == False:
            raise RedisImportException

        if name in self.queue_dict.keys(): #???????
            return self.queue_dict[name]
        else: #????????
            if not name: #???name?????????
                max_name_id = 0
                if self.queue_name_counter.has_key(queue_type):
                    name = queue_type + str(self.queue_name_counter[queue_type] + 1)
                    self.queue_name_counter[queue_type] += 1
                else:
                    name = queue_type + "0"
                    self.queue_name_counter[queue_type] = 0

            #???????????
            if queue_type == "python_queue":
                queue = PythonQueue(name, **kwargs)
            elif queue_type == "redis_queue":
                queue = RedisQ(name, **kwargs)
            self.queue_dict[name] = queue
            return queue

def resubmit_jobs():
    '''
    Examines the fetch and gather queues for items that are suspiciously old.
    These are removed from the queues and placed back on them afresh, to ensure
    the fetch & gather consumers are triggered to process it.
    '''
    if config.get('ckan.harvest.mq.type') != 'redis':
        return
    redis = get_connection()

    # fetch queue
    harvest_object_pending = redis.keys(get_fetch_routing_key() + ':*')
    for key in harvest_object_pending:
        date_of_key = datetime.datetime.strptime(redis.get(key),
                                                 "%Y-%m-%d %H:%M:%S.%f")
        # 3 minutes for fetch and import max
        if (datetime.datetime.now() - date_of_key).seconds > 180:
            redis.rpush(get_fetch_routing_key(),
                json.dumps({'harvest_object_id': key.split(':')[-1]})
            )
            redis.delete(key)

    # gather queue
    harvest_jobs_pending = redis.keys(get_gather_routing_key() + ':*')
    for key in harvest_jobs_pending:
        date_of_key = datetime.datetime.strptime(redis.get(key),
                                                 "%Y-%m-%d %H:%M:%S.%f")
        # 3 hours for a gather
        if (datetime.datetime.now() - date_of_key).seconds > 7200:
            redis.rpush(get_gather_routing_key(),
                json.dumps({'harvest_job_id': key.split(':')[-1]})
            )
            redis.delete(key)

def __init__(self, redis, routing_key):
        self.redis = redis
        # Routing keys are constructed with {site-id}:{message-key}, eg:
        # default:harvest_job_id or default:harvest_object_id
        self.routing_key = routing_key
        # Message keys are harvest_job_id for the gather consumer and
        # harvest_object_id for the fetch consumer
        self.message_key = routing_key.split(':')[-1]

def queue_purge(self, queue=None):
        '''
        Purge the consumer's queue.

        The ``queue`` parameter exists only for compatibility and is
        ignored.
        '''
        # Use a script to make the operation atomic
        lua_code = b'''
            local routing_key = KEYS[1]
            local message_key = ARGV[1]
            local count = 0
            while true do
                local s = redis.call("lpop", routing_key)
                if s == false then
                    break
                end
                local value = cjson.decode(s)
                local id = value[message_key]
                local persistance_key = routing_key .. ":" .. id
                redis.call("del", persistance_key)
                count = count + 1
            end
            return count
        '''
        script = self.redis.register_script(lua_code)
        return script(keys=[self.routing_key], args=[self.message_key])

def dump_all(redis=r):
    keys = redis.keys('*')
    pairs = {}
    for key in keys:
        type = redis.type(key)
        val = redis.get(key)
        try:
            pairs[key] = eval(val)
        except Exception as e:
            print pairs, key, val, e
    return pairs

def del_all(redis=r):
    keys = redis.keys('*')
    for k in keys:
        print 'Deleting:', k, 'result is', redis.delete(k)

def main():
    # del_all()
    # print json.dumps(dump_all(), indent=4)
    keys = r.keys('*')
    print keys
    print len(keys)

def test_aredis(i):
    start = time.time()
    client = aredis.StrictRedis(host=HOST)
    res = None
    for i in range(i):
        res = await client.keys('*')
    print(time.time() - start)
    return res

def test_asyncio_redis(i):
    connection = await asyncio_redis.Connection.create(host=HOST, port=6379)
    start = time.time()
    res = None
    for i in range(i):
        res = await connection.keys('*')
    print(time.time() - start)
    connection.close()
    return res

def test_conn(i):
    start = time.time()
    client = redis.StrictRedis(host=HOST)
    res = None
    for i in range(i):
        res = client.keys('*')
    print(time.time() - start)
    return res

def test_aioredis(i, loop):
    start = time.time()
    redis = await aioredis.create_redis((HOST, 6379), loop=loop)
    val = None
    for i in range(i):
        val = await redis.keys('*')
    print(time.time() - start)
    redis.close()
    await redis.wait_closed()
    return val

def pull_redis_queue(self, host="localhost", port=6379, **kwargs): #????redis????
        if not redis_enable:
            raise RedisImportException
        redis = redis.Redis(host = host, port = port, **kwargs)
        for key in redis.keys():
            if name[:11] == "redis_queue":
                self.queue_dict[key] = RedisQ(key, **kwargs)
                if self.queue_name_counter.has_key(queue_type):
                    self.queue_name_counter["redis_queue"] += 1
                else:
                    self.queue_name_counter["redis_queue"] = 0

def remove(self, queue_object=None, name = None): #????
        if (not queue_object) and (not name): #????????
            for queue in self.queue_dict.values():
                queue = None
            self.queue_dict = dict()
        elif queue_object in self.queue_dict.values(): #????????
            del self.queue_dict[queue_object.name]
        elif name in self.queue_dict.keys(): #????????
            del self.queue_dict[name]
        else:
            raise Exception("queue error")