关于 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 的几种数据库设计方案的内存占用测试, 感谢原作者分享。