Redis作为高性能的Nosql数据库越来越受欢迎，并在使用很多应用场景。常见的一种用法就是缓存一些用户数据。通常我们也会因为有特殊要求，手动插入一些缓存数据。
数据量小的时候，可以直接插入。由于redis是单线程的模式，数据量很大的时候也是一个线程处理并发，单条的插入的时间也会随着数据规模线性增长。同时还会无疑给redis带来类似ddos的操作，虽然redis的性能足够抗住这些并发。
一个行之有效的方式就是使用redis的Redis Mass Insertion – Redis)方式。
pipline模式插入
将需要插入的redis命令按照redis协议进行编码，然后通过管道传给redis-cli，配置—pipe选项即可。
例如有下面的一些命令：
SET Key0 Value0
SET Key1 Value1
...
SET KeyN ValueN
保存为文本文件，然后通过redis协议编码。
cat data.txt | redis-cli --pipe
执行成功会返回：
All data transferred. Waiting for the last reply...
Last reply received from server.
errors: 0, replies: 1000000
redis协议编码
可见，使用pipe的关键在于redis协议的编码。上面的mass文章中使用了ruby的例子。实际上根据redis协议的文档Redis Protocol specification – Redis，我们可以轻松的使用python实现。也可以参考之前redis.py 源码分析的编码协议解析Redispy 源码学习（二） — RESP协议简介
字串编码
Redis的指令为 命令+参数。所有指令中，无怪乎就只有字串和数字。编码字串的格式为，以$符号开头，然后紧跟着字串的长度，接下来是字串本身，最后再加上一个CRLF分隔符即可。大致形式如下：
$ + string_length + string + CRLF
例如hello可以编码为$5\r\nhello\r\n， 字串’21.7’编码成$3\r\n21.7\r\n。
数字编码
数字类型和字串类似，使用:符号开头，紧跟着数字本身，最后以CRLF结尾。不过在执行命令的时候，即使是数字，也都是字串类型，只有redis返回的响应，才需要针对数字编码的方式解码。
组合编码
针对redis的命令，通过字符编码组合起来，就是最终的redis命令编码，每个指令都使用CRLF分割。即：
*<参数数量> CR LF
$<参数 1 的字节数量> CR LF
<参数 1 的数据> CR LF
...
$<参数 N 的字节数量> CR LF
<参数 N 的数据> CR LF
再如：
PING 编码字符串的方式为$4\r\nPING\r\n，组合的编码为*1\r\n$4\r\nPING\r\n。
SET 中国 21.7的将编码成
*3\r\n$3\r\nSET\r\n$6\r\n\xe4\xb8\xad\xe5\x9b\xbd\r\n$4\r\n21.7\r\n
实现
实现一个RedisProto的class，然后提供pack_command方法打包命令进行编码，其参数就是redis命令各部分，redis命令可以使用空格分开成为一个元组。
例如： SET hello world为 (‘SET’, ‘hello’, ‘world’)
GET hello为 (‘GET’, ‘hello’)
HSET key name python 为 (‘HSET’, ‘key’, ‘name’, ‘python’)
下面为 python2的实现。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
class Token(object):
def __init__(self, value):
if isinstance(value, Token):
value = value.value
self.value = value
def __repr__(self):
return self.value
def __str__(self):
return self.value
def b(x):
return x
SYM_STAR = b('*')
SYM_DOLLAR = b('$')
SYM_CRLF = b('\r\n')
SYM_EMPTY = b('')
class RedisProto(object):
def __init__(self, encoding='utf-8', encoding_errors='strict'):
self.encoding = encoding
self.encoding_errors = encoding_errors
def pack_command(self, *args):
"""将redis命令安装redis的协议编码,返回编码后的数组,如果命令很大,返回的是编码后chunk的数组"""
output = []
command = args[0]
if ' ' in command:
args = tuple([Token(s) for s in command.split(' ')]) + args[1:]
else:
args = (Token(command),) + args[1:]
buff = SYM_EMPTY.join(
(SYM_STAR, b(str(len(args))), SYM_CRLF))
for arg in map(self.encode, args):
# 数据量特别大的时候,分成部分小的chunk
if len(buff) > 6000 or len(arg) > 6000:
buff = SYM_EMPTY.join((buff, SYM_DOLLAR, b(str(len(arg))), SYM_CRLF))
output.append(buff)
output.append(arg)
buff = SYM_CRLF
else:
buff = SYM_EMPTY.join((buff, SYM_DOLLAR, b(str(len(arg))), SYM_CRLF, arg, SYM_CRLF))
output.append(buff)
return output
def encode(self, value):
if isinstance(value, Token):
return b(value.value)
elif isinstance(value, bytes):
return value
elif isinstance(value, int):
value = b(str(value))
elif not isinstance(value, str):
value = str(value)
if isinstance(value, str):
value = value.encode(self.encoding, self.encoding_errors)
return value
if __name__ == '__main__':
""" python redis_insert.py | redis-cli --pipe """
commands_args = [('SET', 'Hello', 'World'),
('Set', 'Country', '{"name": 中国"}'),
('HSET', 'Key', 'Name', 'Python'),
('HMSET', 'user', 'name', 'Rsj217', 'Age', '21')]
commands = ''.join([RedisProto().pack_command(*args)[0] for args in commands_args])
print commands
最后执行 python redis_insert.py | redis-cli --pipe 即可，返回如下：
$ python redis_insert.py | redis-cli --pipe
All data transferred. Waiting for the last reply...
Last reply received from server.
errors: 0, replies: 4
最后实测了一下，单条循环插入100w数据大概需要20多分钟，使用pipline方式只需要不到15s。
