购物网站的redis相关实现

---

需求:

（1）登录和cookie缓存

对于一个大型网上商店，假设每天都会有大约500万不同的用户，这些用户会给网站带来1亿次点击，并从网站购买超过10万件商品。

我们需要存储用户登录信息，用户的访问时长和已浏览商品的数量，如果将其保存到数据库中，会导致大量的数据库写入。

大多数关系数据库在每台数据库服务器上面每秒只能插入、更新或者删除200～2000个数据行，尽管批量操作可以以更快的速度执行，但客户点每次浏览网页都只更新少数几行数据，所以高速的批量插入在这里并不适用。

而对于负载量相对比较大的系统，譬如平均情况下每秒大约1200次写入，高峰时期每秒接近6000次写入，所以它必须部署10台关系数据库服务器才能应对高峰期的负载量。

为了提升系统的处理速度，降低资源的占用量，可以将传统数据库的一部分数据处理任务以及存储任务转交给Redis来完成。

（2）使用redis实现购物车

我们把购物车的信息也存储到Redis，并且使用与用户会话令牌一样的cookie id来引用购物车。

将用户和购物车都存储到Redis里面，这种做法除了可以减少请求体积外，我们可以根据用户浏览过的商品，用户放入购物车的商品以及用户最终购买的商品进行统计计算，并构建起很多大型网络零售上都在提供的”在查看过这件商品的用户当中，有X%的用户最终购买了这件商品“”购买了这件商品的用户也购买了某某其他商品“等功能，这些功能可以帮助用户查找其他相关的商品，并最终提升网站的销售业绩。

（3）网页缓存

购物网站上多数页面实际上并不会经常发生大变化，虽然会向分类中添加新商品、移除旧商品、有时候特价促销、有时甚至还有”热卖商品“页面，但是在一般情况下，网站只有账号设置、以往订单、购物车（结账信息）以及其他少数几个页面才包含需要每次载入都要动态生成的内容。

对于不需要动态生成的页面，我们需要尽量不再生成，减少网站在动态生成内容上面所花的时间，可以降低网站处理相同负载所需的服务器数量，让网站速度加快。

python应用框架大都存在中间件，我们创建中间件来调用Redis缓存函数：对于不能被缓存的请求，直接生成并返回页面，对于可以被缓存的请求，先从缓存取出缓存页面，如果缓存页面不存在，那么会生成页面并将其缓存在Redis，最后将页面返回给函数调用者。

这样的方式可以让网站在5分钟之内无需再为他们动态地生成视图页面。

(4) 数据行缓存

为了清空旧库存和吸引客户消费，决定开始新一轮的促销活动，每天都会推出一些特价商品供用户抢购，所有特价商品的数量都是限定的，卖完为止。在这种情况下，网站是不能对整个促销页面进行缓存，这会导致用户看到错误的特价商品和商品剩余数，但每次载入页面都从数据库中取出特价商品的剩余数量的话，又会给数据库带来巨大的压力。

为了应付促销活动带来的大量负载，需要对数据行进行缓存，可以编写一个持续运行的守护进程函数，让这个函数将指定的数据行缓存到Redis里面，并不定期地对这些缓存进行更新。缓存函数将数据和编码为json字典并存储在Redis的字符串中。

我们还需要使用两个有序集合来记录应该在何时对缓存进行更新，第一个有序集合为调度有序集合，成员为数据行的ID,分值为时间戳，记录应该在何时将制定的数据行缓存到Redis里面。第二个有序集合为延时有序集合，成员为数据行的ID，而分值记录指定数据行的缓存需要每隔多少秒更新一次。

对于更新频率，如果数据行记录的是特价促销商品的剩余数量，并且参与促销活动的用户非常多，那么我么最好每隔几秒更新一次数据行缓存，如果数据并不经常改变，或者商品缺货是可以接受的，我们可以每分钟更新一次缓存。

（5）网页分析

之前对于网页的缓存，如果网站总共包含100000件商品，贸然缓存所有商品页面将耗尽整个网站的全部内存，所以我们可以只针对那些浏览量较高的商品页面进行缓存。

每个用户都有一个相应的记录用户浏览商品历史的有序集合，我们在记录的过程中，我们也痛死记录所有商品的浏览次数，根据浏览次数对商品进行排序，被浏览得最多的商品放到有序集合的索引0位置上，并且具有整个有序集合最少的分值。

除了缓存最常被浏览的商品外，我们还需要发现那些变得越来越流畅的新商品，于是我们需要定期修剪有序集合的长度并调整已有元素的分值，才能使得新流行的商品在排行榜中占据一席之地。

Redis数据结构设计

（1）登录令牌与用户映射关系的散列 "login:"
（2）记录最近登录用户的有序集合 "recent:"
（3）记录各个用户最近浏览商品的有序集合 "viewed:94233rhsYRIq3yi3qryrye"
（4）每个用户的购物车散列，存储商品ID与商品订购数量之间的映射。"cart:94233rhsYRIq3yi3qryrye"
（5）请求页面缓存集合 "cache:wre9w3rieruerwe3" (wre9w3rieruerwe3代表请求ID)

(94233rhsYRIq3yi3qryrye假设为某个用户的令牌)
（6）数据行缓存字符串，数据列(column)的名字会被映射为json字典的键，而数据行的值会被映射为json字典的值，"inv:273" (其中273为数据行id)。
（7）数据行缓存调度有序集合，成员为数据行的ID,分值为时间戳，记录应该在何时将制定的数据行缓存到Redis里面，"schedule："。
（8）数据行缓存延时有序集合，成员为数据行的ID，而分值记录指定数据行的缓存需要每隔多少秒更新一次，"delay:"。

（9）商品浏览次数有序集合，成员为商品，分值为浏览次数负值，方便保持在有序集合的较前的索引位置，"viewed"。

Redis实现

（1）使用散列来存储登录cookie令牌与已登录用户之前的映射。根据给定的令牌查找与之相应的用户，检查用户是否登录，并返回该用户的ID。

"""
获取并返回令牌对应的用户@param {object}
@param {string} token@return {string} 用户id
"""
def checkToken(conn, token):return conn.hget('login:', token)

（2）用户每次浏览页面的时候，需要更新“登录令牌与用户映射关系的散列”里面的信息，
并将用户的令牌和当前时间戳添加到 “记录最近登录用户的有序集合” 里面，
将浏览商品添加到记录“记录各个用户最近浏览商品的有序集合”中，如果记录的商品数量超过25个，对这个有序集合进行修剪。

"""
更新令牌时，需要更改用户令牌信息，将用户记录到最近登录用户的有序集合中，
如果用户浏览的是商品，则需要将浏览商品写入该用户浏览过商品的有序集合中，并保证该集合不超过25个@param {object}
@param {string} token
@param {string} user
@param {string} item"""
def updateToken(conn, token, user, item = None):timestamp = time.time()# 更新用户令牌登录对应的用户信息conn.hset('login:', token, user)# 增加最近访问的用户到有序集合conn.zadd('recent:', token, timestamp)# 如果浏览产品，记录该用户最近访问的25个产品if item:conn.zadd('viewed:' + token, item, timestamp)conn.zremrangebyrank('viewed:' + token, 0, -26)# 记录每个商品的浏览量conn.zincrby('viewed:', item, -1)

（3）存储会话的内存会随着时间的推移而不断增加，需要定期清理会话数据，我们决定只保留最新的1000万个会话。

我们可以用 守护进程的方式来运行或者定义一个cron job每隔一段时间运行 ，
检查最近 “记录最近登录用户的有序集合” 大小是否超过了限制，超过限制每秒从集合中删除最旧的100个令牌，并且移除相应的“登录令牌与用户映射关系的散列”的信息和对应的“记录各个用户最近浏览商品的有序集合”，对应的”美国用户的购物车散列“。

我们也可以使用EXPIRE命令，为用户令牌设记录用户商品浏览记录的有序集合设置过期时间，让Redis在一段时间之后自动删除它们，这样就不用使用有序集合来记录最近出现的令牌了，但是这样我们就没办法将会话数限制在1000万之内了。

"""
定期清理会话数据，只保留最新的1000万个会话。使用 *守护进程的方式来运行或者定义一个cron job每隔一段时间运行* ，
检查最近 “记录最近登录用户的有序集合” 大小是否超过了限制，超过限制每秒从集合中删除最旧的100个令牌，
并且移除相应的“登录令牌与用户映射关系的散列”的信息和对应的“记录各个用户最近浏览商品的有序集合”。@param {object}
"""# 循环判断，如果是cron job可以不用循环
QUIT = False
# 限制保留的最大会话数据
LIMIT = 10000000def cleanFullSession(conn):# 循环判断，如果是cron job可以不用循环while not QUIT:# 查询最近登录用户会话数size = conn.zcard('recent:')# 没有超过限制，休眠1秒再继续执行if size <= LIMIT:time.sleep(1)continue# 查询最旧登录的最多100个令牌范围end_index = min(size - LIMIT, 100)tokens = conn.zrange('recent:', 0, end_index - 1)# 将要删除的key都推入到数组中，要时候一起删除session_keys = []for token in tokens:session_keys.append('viewed:' + token)session_keys.append('cart:' + token)# 批量删除相应的用户最近浏览商品有序集合，用户的购物车，登录令牌与用户映射关系的散列和记录最近登录用户的有序集合conn.delete(*session_keys)conn.hdel('login:', *tokens)conn.zrem('recent:', *tokens)

（4）对购物车进行更新，如果用户订购某件商品数量大于0，将商品信息添加到 “用户的购物车散列”中，如果购买商品已经存在，那么更新购买数量。

"""
对购物车进行更新，如果用户订购某件商品数量大于0，将商品信息添加到 “用户的购物车散列”中，如果购买商品已经存在，那么更新购买数量@param {object}
@param {string} session
@param {string} item
@param {float}  count"""
def addToCart(conn, session, item, count):if count <= 0:# 从购物车移除指定商品conn.hrem('cart:' + session, item)else:# 将指定商品添加到对应的购物车中conn.hset('cart:' + session, item, count)

（5）在用户请求页面时，对于不能被缓存的请求，直接生成并返回页面，对于可以被缓存的请求，先从缓存取出缓存页面，如果缓存页面不存在，那么会生成页面并将其缓存在Redis，最后将页面返回给函数调用者。

"""
在用户请求页面时，对于不能被缓存的请求，直接生成并返回页面，
对于可以被缓存的请求，先从缓存取出缓存页面，如果缓存页面不存在，那么会生成页面并将其缓存在Redis，最后将页面返回给函数调用者。@param {object} conn
@param {string} request
@param {callback}@return 
"""
def cacheRequest(conn, request, callback):# 判断请求是否能被缓存，不能的话直接调用回调函数if not canCache(conn, request):return callback(request)# 将请求转换为一个简单的字符串健，方便之后进行查找page_key = 'cache:' + hashRequest(request)content = conn.get(page_key)# 没有缓存的页面，调用回调函数生成页面，并缓存到redis中if not content:content = callback(request)conn.setex(page_key, content, 300)return content"""
判断页面是否能被缓存，检查商品是否被缓存以及页面是否为商品页面，根据商品排名来判断是否需要缓存@param {object} conn
@param {string} request@return {boolean}
"""
def canCache(conn, request):# 根据请求的URL，得到商品IDitem_id = extractItemId(request)# 检查这个页面能否被缓存以及这个页面是否为商品页面if not item_id or isDynamic(request):return False# 商品的浏览排名rank = conn.zrank('viewed:', item_id)return rank is not None and rank < 10000"""
解析请求的URL,取得query中的item id@param {string} request@return {string}
"""
def extractItemId(request):parsed = urlparse.urlparse(request)# 返回query字典query  = urlparse.parse_qs(parsed.query)return (query.get('item') or [None])[0]"""
判断请求的页面是否动态页面@param {string} request@return {boolean}
"""
def isDynamic(request):parsed = urlparse.urlparse(request)query = urlparse.parse_qs(parsed.query)return '_' in query"""
将请求转换为一个简单的字符串健，方便之后进行查找
@param {string} request@return {string}
"""
def hashRequest(request):return str(hash(request))

（6）为了让缓存函数定期地缓存数据行，首先需要将行ID和给定的延迟值添加到延迟有序集合中，再将行ID和当前时间的时间戳添加到调度有序集合中。如果某个数据行的延迟值不存在，那么程序将取消对这个数据行的调度。如果我们想要移除某个数据行已有的缓存并且不再缓存那个数据行，只需要把那个数据行的延迟值设置为小于或等于0即可。

"""
设置数据行缓存的延迟值和调度时间@param {object} conn
@param {int}    row id
@param {int}    delay"""
def scheduleRowCache(conn, row_id, delay):conn.zadd('delay:', row_id, delay)conn.zadd('schedule:', row_id, time.time())

（7）尝试读取”数据行缓存调度有序集合“的第一个元素以及该元素的分支，如果”数据行缓存调度有序集合“没有包含任何元素，或者分值存储的时间戳所指定的时间尚未来临，那么函数先休眠50毫秒，然后再重新进行检查。

当发现一个需要立即进行更新的数据行时，如果数据行的延迟值小于或者等于0，会从”数据行缓存延时有序集合“和”数据行缓存调度有序集合“移除这个数据行的ID，并从缓存里面删除这个数据行已有的缓存，再重新进行检查。

对于延迟值大于0的数据行来说，从数据库里面取出这些行，将他们编码为json格式并存储到Redis里面，然后更新这些行的调度时间。

"""
守护进程，根据调度时间有序集合和延迟值缓存数据行@param {object} conn"""
def cacheRow(conn):while not QUIT:# 需要读取”数据行缓存调度有序集合“的第一个元素，如果没有包含任何元素，或者分值存储的时间戳所指定的时间尚未来临，那么函数先休眠50毫秒，然后再重新进行检查next = conn.zrange('schedule:', 0, 0, withscores=True)now = time.time()if not next or next[0][1] > now:time.sleep(.05)continuerow_id = next[0][0]# 取出延迟值delay = conn.zscore('delay:', row_id)# 如果延迟值小于等于0，则不再缓存该数据行if delay <= 0:conn.zrem('schedule:', row_id)conn.zrem('delay:', row_id)conn.delete('inv:' + row_id) continue;# 需要缓存的，更新缓存调度的有序集合，并缓存该数据行row = Inventory.get(row_id)conn.zadd('schedule:', row_id, now + delay)conn.set('inv:' + row_id, json.dumps(row.toDict()))"""
库存类，库存的商品信息
"""
class Inventory(object):def __init__(self, id):self.id = id@classmethoddef get(cls, id):return Inventory(id)def toDict(self):return {'id':self.id, 'data':'data to cache...','cached':time.time()}

（8）我们需要在用户浏览页面时，“商品浏览次数有序集合”对应的商品中需要减一，使得保持在有序集合较前的索引位置。

同时我们需要开启一个守护进程，每隔5分钟，删除所有排名在20000名之后的商品浏览数，并使用ZINTERSTORE将删除之后剩余的所有商品的浏览次数减半。

而判断页面是否需要缓存，我们需要通过ZRANK取出商品的浏览次数排名，如果排名在10000内，那么说明该页面需要缓存。

"""
守护进程，删除所有排名在20000名之后的商品，并将删除之后剩余的所有商品浏览次数减半，5分钟执行一次@param {object} conn"""
def rescaleViewed(conn):while not QUIT:conn.zremrangebyrank('viewed:', 20000, -1)conn.zinterstore('viewed:', {'viewed:', .5})time.sleep(300)

测试代码

"""
测试
"""
import time
import urlparse
import uuid
import threading
import unittest
import jsonclass TestShoppingWebsite(unittest.TestCase):def setUp(self):import redisself.conn = redis.Redis(db=15)def tearDown(self):conn = self.connto_del = (conn.keys('login:*') + conn.keys('recent:*') + conn.keys('viewed:*') +conn.keys('cart:*') + conn.keys('cache:*') + conn.keys('delay:*') + conn.keys('schedule:*') + conn.keys('inv:*'))if to_del:conn.delete(*to_del)del self.connglobal QUIT, LIMITQUIT = FalseLIMIT = 10000000printprintdef testLoginCookies(self):conn = self.connglobal LIMIT, QUITtoken = str(uuid.uuid4())updateToken(conn, token, 'username', 'itemX')print "We just logged-in/updated token:", tokenprint "For user:", 'username'printprint "What username do we get when we look-up that tokan?"r = checkToken(conn, token)print rprintself.assertTrue(r)print "Let's drop the maximun number of cookies to 0 to clear them out"print "We will start a thread to do the cleaning, while we stop it later"LIMIT = 0t = threading.Thread(target = cleanFullSession, args = (conn,))t.setDaemon(1)t.start()time.sleep(1)QUIT = Truetime.sleep(2)if t.isAlive():raise Exception("The clean sessions thread is still slive?!?")s = conn.hlen('login:')print "The current number of session still available is:", sself.assertFalse(s)def testShoppingCartCookies(self):conn = self.connglobal LIMIT, QUITtoken = str(uuid.uuid4())print "We'll refresh our session..."updateToken(conn, token, 'username', 'itemX')print "And add an item to the shopping cart"addToCart(conn, token, "itemY", 3)r = conn.hgetall('cart:' + token)print "Our Shopping cart currently has:", rprintself.assertTrue(len(r) >= 1)print "Let's clean out our sessions an carts"LIMIT = 0t = threading.Thread(target=cleanFullSession, args=(conn,))t.setDaemon(1)t.start()time.sleep(1)QUIT = Truetime.sleep(2)if t.isAlive():raise Exception("The clean sessions thread is still alive?!?")r = conn.hgetall('cart:' + token)print "Our shopping cart now contains:", rself.assertFalse(r)def testCacheRequest(self):conn = self.conntoken = str(uuid.uuid4())def callback(request):return "content for " + requestupdateToken(conn, token, 'username', 'itemX')url = 'http://test.com/?item=itemX'print "We are going to cache a simple request against", urlresult = cacheRequest(conn, url, callback)print "We got initial content:", repr(result)printself.assertTrue(result)print "To test that we've cached the request, we'll pass a bad callback"result2 = cacheRequest(conn, url, None)print "We ended up getting the same response!", repr(result2)self.assertEquals(result, result2)self.assertFalse(canCache(conn, 'http://test.com/'))self.assertFalse(canCache(conn, 'http://test.com/?item=itemX&_=1234567'))def testCacheRows(self):import pprintconn = self.connglobal  QUITprint "First, let's schedule caching of itemX every 5 seconds"scheduleRowCache(conn, 'itemX', 5)print "Our schedule looks like:"s = conn.zrange('schedule:', 0, -1, withscores = True)pprint.pprint(s)self.assertTrue(s)print "We'll start a caching thread that will cache the data..."t = threading.Thread(target=cacheRow, args=(conn,))t.setDaemon(1)t.start()time.sleep(1)print "Our cached data looks like:"r = conn.get('inv:itemX')print repr(r)self.assertTrue(r)printprint "We'll check again in 5 seconds..."time.sleep(5)print "Notice that the data has changed..."r2 = conn.get('inv:itemX')print repr(r2)printself.assertTrue(r2)self.assertTrue(r != r2)print "Let's force un-caching"scheduleRowCache(conn, 'itemX', -1)time.sleep(1)r = conn.get('inv:itemX')print "The cache was cleared?", not rprintself.assertFalse(r)QUIT = Truetime.sleep(2)if t.isAlive():raise Exception("The database caching thread is still alive?!?")if __name__ == '__main__':unittest.main()

完整示例代码地址：https://github.com/NancyLin/r...