每日一题_Python.利用gevent和pipeline快速导出近千万Redis字段值?
#!/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):
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())
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())
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)
页:
[1]