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

def get_twemproxies_status(server, status):
    cmd = "status\r\n"
    n_send = server.tws_conn.send(cmd)
    if n_send != len(cmd):
        return

    buf = ""
    while 1:
        data = server.tws_conn.recv(1024)
        if not data:
            # connection close, let's kill it and raise
            raise Exception("connection closed by server")

        buf += data
        if buf.endswith("\n"):
            break

    tws_status = json.loads(buf)

    parse_twemproxies_status(tws_status, status)
    return

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-c", "--config", type=str, help="Configuration file")
    args = parser.parse_args()
    starttime = time.time()

    # Store opened shell sessions
    shells = {}

    # FIXME redis connection settings should be in config
    redis_conn = redis.StrictRedis()
    p = redis_conn.connection_pool

    publish = gevent.spawn(publisher, redis_conn)
    # FIXME: use config
    workers = 2
    log.info(_("Spawning %s greenlets connecting to Redis..."), workers)
    redis_greenlets = [gevent.spawn(execute_workflow, redis_conn, _id, shells)
                       for _id in xrange(workers)]
    # Wait until all greenlets have started and connected.
    gevent.sleep(1)

    log.info(_("# active `threading` threads: %s") % threading.active_count())
    log.info(_("# Redis connections created: %s") % p._created_connections)
    log.info(_("# Redis connections in use: %s") % len(p._in_use_connections))
    log.info(_("# Redis connections available: %s") % len(p._available_connections))
    log.info(_("Waiting for Redis connection greenlets to terminate..."))
    gevent.joinall(redis_greenlets)

    d = time.time() - starttime
    log.info(_("All Redis connection greenlets terminated. Duration: %.2f s.") % d)
    publish.kill()

def redis_connection_create():
    import redis.connection
    redis_conn = redis.StrictRedis(
        host=host,
        port=port,
        password=password)

    return redis_conn

def pika_connection_create():
    import redis.connection
    try:
        pika_conn = redis.StrictRedis(
            host=host,
            port=port,
            password=password)
    except Exception,e:
        server.err = 1
        server.errmsg = get_exception_message(sys._getframe().f_code.co_name,sys._getframe().f_lineno, e)

    return pika_conn

def usage():
    print 'python sss.py [options]'
    print ''
    print 'options:'
    print '-h,--help: show this help message'
    print '-v,--version: show the version'
    print '-H: target host'
    print '-P: target port to get the service status'
    print '-u: target service user'
    print '-p: target user password'
    print '-T: target service type, default is '+support_types[0]
    print '-s: sections to show, use comma to split'
    print '-a: addition sections to show, use comma to split'
    print '-d: removed sections for the showing, use comma to split'
    print '-I,--instructions: show the support sections\' instructions'
    print '-o: output the status to this file'
    print '-D: separate output files by day, suffix of the file name is \'_yyyy-mm-dd\''
    print '-e: output error message to this file'
    print '-i: time interval to show the status, unit is second'
    print '-n: the count of the status to collect, default is forever'
    print '-S: speed is calculated by the remote monitor system, like the open-falcon'
    print '--socket: the socket file to use for connection'
    print '--falcon: upload the status to the open-falcon, the address is like \''+open_falcon+'\''
    print '--net-face: set the net device face name for os_net_* sections, default is \'lo\''
    print '--disk-name: set the disk device name for os_disk sections, default is \'vda\''
    print '--proc-pid: set the process pid number for proc_* sections, default is 0'
    print '--service-port: set the server service port, default is same as the port setted by \'-P\' option'
    print '\r\n'
    support_services=""
    for service in support_types:
        support_services+=service+" "
    print 'Support services: '+support_services
    print ''

def connection_pool_class(self):
        cls = self.options.get('POOL_CLASS', 'redis.ConnectionPool')
        mod_path, cls_name = cls.rsplit('.', 1)
        try:
            mod = import_object(mod_path)
            pool_class = getattr(mod, cls_name)
        except (AttributeError, ImportError):
            raise ConfigError("Could not find connection pool class '%s'" % cls)
        return pool_class

def __call__(self):
        """
        Entry point.
        """

        # Establish a connection with Redis server.
        self._redis = redis.StrictRedis(**connection.get_redis_settings())
        self._pubsub = self._redis.pubsub(ignore_subscribe_messages=True)
        self._pubsub.subscribe(connection.QUERYOBSERVER_REDIS_CHANNEL)

        while self._pubsub.subscribed:
            event = self._pubsub.get_message(ignore_subscribe_messages=True, timeout=0.1)
            if not event:
                continue

            # Events are assumed to be pickled data.
            try:
                event = pickle.loads(event['data'])
            except ValueError:
                logger.error("Ignoring received malformed event '{}'.", event['data'][:20])
                continue

            # Handle event.
            try:
                event_name = event.pop('event')
                handler = getattr(self, 'event_%s' % event_name)
            except AttributeError:
                logger.error("Ignoring unimplemented event '{}'.", event_name)
                continue
            except KeyError:
                logger.error("Ignoring received malformed event '{}'.", event)
                continue

            try:
                handler(**event)
            except:
                logger.error("Unhandled exception while executing event '{}'.", event_name)
                logger.error(traceback.format_exc())
            finally:
                db.close_old_connections()

        self._pubsub.close()