关于 Redis 的几种数据库设计方案的内存占用测试

最近在做一个项目，数据库使用的是 Redis。在设计数据结构时，不知道哪种实现是最优的，于是做了下测试。 测试环境如下： OS X10.8.3 Redis 2.6.12 Python 2.7.4 redis-py 2.7.2 hiredis 0.1.1 ujson 1.30 MessagePack 0.3.0 注意： 因为是拿 Python 测试的

最近在做一个项目，数据库使用的是 Redis。在设计数据结构时，不知道哪种实现是最优的，于是做了下测试。

测试环境如下：

OS X10.8.3
Redis 2.6.12
Python 2.7.4
redis-py 2.7.2
hiredis 0.1.1
ujson 1.30
MessagePack 0.3.0

注意：

1. 因为是拿 Python 测试的，所以可能对其他语言并不完全适用。
   
2. 使用的测试数据是特定的，可能对更小或更大的数据并不完全适用。
   

测试结果就不列出了，直接说结论吧。

1. 最差的存储方式就是用一个 hash 来存储一个实体（即一条记录）。时间上比其他方案慢 1 ~ 2 倍，空间占用较大。
   更重要的是拿出来的字段类型是字符串，还得自己转换类型。
   唯一的好处就是可以单独操作一个字段。
   
2. 使用 string 类型来存储也是不推荐的，不过稍好于前一种方式。在单个实体较小时，会暴露出 key 占用内存较多的缺点。
   
3. 用一个 hash 来存储一个类型的所有实体（即一张表），在实现上比较简单，内存占用尚可。
   
4. 用多个 hash 来存储一个类型的所有实体（即分表），在实现上稍微复杂点，但占用的内存最小。
   如果单个字段值较小（缺省值是 64 字节），单个 hash 存储的字段数不多（缺省值是 512 个）时，会采用 hash zipmap 来存储，内存占用会显著减小。
   单个 hash 存储的字段数建议为 2 的次方，例如 1024。略微超过这个值，会导致内存占用和延迟时间都增加。
   Instagram 的工程师认为，使用 hash zipmap 时，最佳的字段数为 1000 左右。不过据我测试，基本都是随字段数增加而变慢，而内存占用从 128 直到 1024 的变化基本可以忽略。
   
5. 存储为 JSON 格式是种不错的选择。对包含中文的内容来说，设置 ensure_ascii=False 可以节省大量内存。
   ujson 比 json 性能好很多，后者在设置 ensure_ascii=False 后性能急剧下降。
   
6. cPickle 比 ujson 的性能要差，不过支持更多类型（如 datetime）。
   
7. MessagePack 比 ujson 有一点不太明显的性能优势，不过丧失了可读性，且取回 unicode 需要自己 decode。
   号称比 Protocol Buffer 快 4 倍应该可以无视了，至少其 Python 库没有明显优势。
   
8. 使用 zlib 压缩可以节省更多内存，不过性能变慢 1 ~ 2 倍。
   

看这个测试结果，感觉还不如用 MongoDB 省事……

最后附上测试代码：

# -*- coding: utf-8 -*-
import cPickle
import json
import time
import zlib
import msgpack
import redis
import ujson
class Timer:    
	def __enter__(self):
		self.start = time.time()
		return self
	def __exit__(self, *args):
		self.end = time.time()
		self.interval = self.end - self.start
def test(function):
	def wrapper(*args, **kwargs):
		args_list = []
		if args:
			args_list.append(','.join((str(arg) for arg in args)))
		if kwargs:
			args_list.append(','.join('%s=%s' % (key, value) for key, value in kwargs.iteritems()))
		print 'call %s(%s):' % (function.func_name, ', '.join(args_list))
		redis_client.flushall()
		print 'memory:', redis_client.info()['used_memory_human']
		with Timer() as timer:
			result = function(*args, **kwargs)
		print 'time:', timer.interval
		print 'memory:', redis_client.info()['used_memory_human']
		print
		return result
	return wrapper
redis_client = redis.Redis()
pipe = redis_client.pipeline(transaction=False)
articles = [{
	'id': i,
	'title': u'团结全世界正义力量痛击日本',
	'content': u'近期日本社会有四种感觉极度高涨，即二战期间日本军国主义扩张战争的惨败在日本右翼*内心留下的耻辱感；被美国长期占领和控制的压抑感；经济长期停滞不前的焦虑感；对中国快速崛起引发的失落感。为此，日本为了找到一个发泄口，对中国采取了一系列挑衅行为，我们不能听之任之。现在全国13亿人要万众一心，团结起来，拿出决心、意志和能力，果断实施对等反击。在这场反击日本右翼*的反攻倒算中，中国不是孤立的，我们要团结全世界一切反法西斯战争的正义力量，痛击日本对国际正义的挑战。',
	'source_text': u'环球时报',
	'source_url': 'http://opinion.huanqiu.com/column/mjzl/2012-09/3174337.html',
	'time': '2012-09-13 09:23',
	'is_public': True
} for i in xrange(10000)]
@test
def test_hash():
	for article in articles:
		pipe.hmset('article:%d' % article['id'], article)
	pipe.execute()
@test
def test_json_hash():
	for article in articles:
		pipe.hset('article', article['id'], json.dumps(article))
	pipe.execute()
@test
def test_ujson_hash():
	for article in articles:
		pipe.hset('article', article['id'], ujson.dumps(article))
	pipe.execute()
@test
def test_ujson_string():
	for article in articles:
		pipe.set('article:%d' % article['id'], ujson.dumps(article))
	pipe.execute()
@test
def test_zlib_ujson_string():
	for article in articles:
		pipe.set('article:%d' % article['id'], zlib.compress(ujson.dumps(article, ensure_ascii=False)))
	pipe.execute()
@test
def test_msgpack():
	for article in articles:
		pipe.hset('article', article['id'], msgpack.packb(article))
	pipe.execute()
@test
def test_pickle_string():
	for article in articles:
		pipe.set('article:%d' % article['id'], cPickle.dumps(article))
	pipe.execute()
@test
def test_json_without_ensure_ascii():
	for article in articles:
		pipe.hset('article', article['id'], json.dumps(article, ensure_ascii=False))
	pipe.execute()
@test
def test_ujson_without_ensure_ascii():
	for article in articles:
		pipe.hset('article', article['id'], ujson.dumps(article, ensure_ascii=False))
	pipe.execute()
def test_ujson_shard_id():
	@test
	def test_ujson_shard_id_of_size(size):
		for article in articles:
			article_id = article['id']
			pipe.hset('article:%d' % (article_id / size), article_id % size, ujson.dumps(article, ensure_ascii=False))
		pipe.execute()
	for size in (2, 4, 8, 10, 16, 32, 64, 100, 128, 256, 500, 512, 513, 1000, 1024, 1025, 2048, 4096, 8092):
		test_ujson_shard_id_of_size(size)
	test_ujson_shard_id_of_size(512)
for key, value in sorted(globals().copy().iteritems(), key=lambda x:x[0]):
	if key.startswith('test_'):
		value()

原文地址：关于 Redis 的几种数据库设计方案的内存占用测试, 感谢原作者分享。