每日一题_Python.利用gevent和pipeline快速导出近千万Redis字段值?

pqwsa

#!/usr/bin/env python　　
# -*- coding: utf-8 -*-
　　
"""
　　
#
　　
# Authors: limanman
　　
# OsChina: http://xmdevops.blog.51cto.com/
　　
# Purpose:
　　
#
　　
"""
　　
# 说明: 导入公共模块
　　
import os
　　
from gevent import monkey
　　
from gevent.pool import Pool
　　
from itertools import product
　　
from redis.exceptions import ResponseError
　　
from redis import StrictRedis, ConnectionPool
　　
# 说明: 导入其它模块
　　

　　

　　
host = '127.0.0.1'
　　
port = 5123
　　
monkey.patch_all()
　　

　　

　　
def write_serialdata(key):
　　
    rds_pool = ConnectionPool(host=host, port=port)
　　
    rds_inst = StrictRedis(connection_pool=rds_pool, retry_on_timeout=True, max_connections=10)
　　
    rds_pipe = rds_inst.pipeline()
　　
    prefix = ''.join(key)
　　
    redis_key = '{0}*'.format(prefix)
　　
    rds_pipe.keys(redis_key)
　　
    redis_val = rds_pipe.execute()
　　
    if not redis_val:
　　
        return
　　
    with open(prefix, 'a+b') as fd:
　　
        for index, serial_key in enumerate(redis_val[0]):
　　
            if index % 1000 == 0:
　　
                serial_val = rds_pipe.execute()
　　
                if not serial_val:
　　
                    continue
　　
                for item in serial_val:
　　
                    if 'WanIP' not in item:
　　
                        continue
　　
                    ip = item['WanIP']
　　
                    print 'record ip => {0} to file {1}'.format(ip, prefix)
　　
                    fd.write(''.join([ip, os.linesep]))
　　
            if not serial_key.isalnum():
　　
                continue
　　
            rds_pipe.hgetall(serial_key)
　　
        serial_val = rds_pipe.execute()
　　
        if not serial_val:
　　
            return
　　
        for item in serial_val:
　　
            if 'WanIP' not in item:
　　
                continue
　　
            ip = item['WanIP']
　　
            print 'record ip => {0} to file {1}'.format(ip, prefix)
　　
            fd.write(''.join([ip, os.linesep]))
　　

　　

　　
def generat_product():
　　
    s_bit = 'abcdef0123456789'
　　
    return product(s_bit, repeat=2)
　　

　　
if __name__ == '__main__':
　　
    pool = Pool(100)
　　
    keys = generat_product()
　　
    path = []
　　
    for key_pairs in keys:
　　
        fpath = ''.join(key_pairs)
　　
        path.append(fpath)
　　
        if not os.path.exists(fpath):
　　
            file(fpath, 'w+b').close()
　　
    pool.map(write_serialdata, path)