摘要：購(gòu)物網(wǎng)站的相關(guān)實(shí)現(xiàn)需求登錄和緩存對(duì)于一個(gè)大型網(wǎng)上商店，假設(shè)每天都會(huì)有大約萬不同的用戶，這些用戶會(huì)給網(wǎng)站帶來億次點(diǎn)擊，并從網(wǎng)站購(gòu)買超過萬件商品。根據(jù)給定的令牌查找與之相應(yīng)的用戶，檢查用戶是否登錄，并返回該用戶的。

對(duì)于一個(gè)大型網(wǎng)上商店，假設(shè)每天都會(huì)有大約500萬不同的用戶，這些用戶會(huì)給網(wǎng)站帶來1億次點(diǎn)擊，并從網(wǎng)站購(gòu)買超過10萬件商品。

我們需要存儲(chǔ)用戶登錄信息，用戶的訪問時(shí)長(zhǎng)和已瀏覽商品的數(shù)量，如果將其保存到數(shù)據(jù)庫(kù)中，會(huì)導(dǎo)致大量的數(shù)據(jù)庫(kù)寫入。

大多數(shù)關(guān)系數(shù)據(jù)庫(kù)在每臺(tái)數(shù)據(jù)庫(kù)服務(wù)器上面每秒只能插入、更新或者刪除200～2000個(gè)數(shù)據(jù)行，盡管批量操作可以以更快的速度執(zhí)行，但客戶點(diǎn)每次瀏覽網(wǎng)頁(yè)都只更新少數(shù)幾行數(shù)據(jù)，所以高速的批量插入在這里并不適用。

而對(duì)于負(fù)載量相對(duì)比較大的系統(tǒng)，譬如平均情況下每秒大約1200次寫入，高峰時(shí)期每秒接近6000次寫入，所以它必須部署10臺(tái)關(guān)系數(shù)據(jù)庫(kù)服務(wù)器才能應(yīng)對(duì)高峰期的負(fù)載量。

為了提升系統(tǒng)的處理速度，降低資源的占用量，可以將傳統(tǒng)數(shù)據(jù)庫(kù)的一部分?jǐn)?shù)據(jù)處理任務(wù)以及存儲(chǔ)任務(wù)轉(zhuǎn)交給Redis來完成。

我們把購(gòu)物車的信息也存儲(chǔ)到Redis，并且使用與用戶會(huì)話令牌一樣的cookie id來引用購(gòu)物車。

將用戶和購(gòu)物車都存儲(chǔ)到Redis里面，這種做法除了可以減少請(qǐng)求體積外，我們可以根據(jù)用戶瀏覽過的商品，用戶放入購(gòu)物車的商品以及用戶最終購(gòu)買的商品進(jìn)行統(tǒng)計(jì)計(jì)算，并構(gòu)建起很多大型網(wǎng)絡(luò)零售上都在提供的”在查看過這件商品的用戶當(dāng)中，有X%的用戶最終購(gòu)買了這件商品“”購(gòu)買了這件商品的用戶也購(gòu)買了某某其他商品“等功能，這些功能可以幫助用戶查找其他相關(guān)的商品，并最終提升網(wǎng)站的銷售業(yè)績(jī)。

購(gòu)物網(wǎng)站上多數(shù)頁(yè)面實(shí)際上并不會(huì)經(jīng)常發(fā)生大變化，雖然會(huì)向分類中添加新商品、移除舊商品、有時(shí)候特價(jià)促銷、有時(shí)甚至還有”熱賣商品“頁(yè)面，但是在一般情況下，網(wǎng)站只有賬號(hào)設(shè)置、以往訂單、購(gòu)物車（結(jié)賬信息）以及其他少數(shù)幾個(gè)頁(yè)面才包含需要每次載入都要?jiǎng)討B(tài)生成的內(nèi)容。

對(duì)于不需要?jiǎng)討B(tài)生成的頁(yè)面，我們需要盡量不再生成，減少網(wǎng)站在動(dòng)態(tài)生成內(nèi)容上面所花的時(shí)間，可以降低網(wǎng)站處理相同負(fù)載所需的服務(wù)器數(shù)量，讓網(wǎng)站速度加快。

python應(yīng)用框架大都存在中間件，我們創(chuàng)建中間件來調(diào)用Redis緩存函數(shù)：對(duì)于不能被緩存的請(qǐng)求，直接生成并返回頁(yè)面，對(duì)于可以被緩存的請(qǐng)求，先從緩存取出緩存頁(yè)面，如果緩存頁(yè)面不存在，那么會(huì)生成頁(yè)面并將其緩存在Redis，最后將頁(yè)面返回給函數(shù)調(diào)用者。

這樣的方式可以讓網(wǎng)站在5分鐘之內(nèi)無需再為他們動(dòng)態(tài)地生成視圖頁(yè)面。

為了清空舊庫(kù)存和吸引客戶消費(fèi)，決定開始新一輪的促銷活動(dòng)，每天都會(huì)推出一些特價(jià)商品供用戶搶購(gòu)，所有特價(jià)商品的數(shù)量都是限定的，賣完為止。在這種情況下，網(wǎng)站是不能對(duì)整個(gè)促銷頁(yè)面進(jìn)行緩存，這會(huì)導(dǎo)致用戶看到錯(cuò)誤的特價(jià)商品和商品剩余數(shù)，但每次載入頁(yè)面都從數(shù)據(jù)庫(kù)中取出特價(jià)商品的剩余數(shù)量的話，又會(huì)給數(shù)據(jù)庫(kù)帶來巨大的壓力。

為了應(yīng)付促銷活動(dòng)帶來的大量負(fù)載，需要對(duì)數(shù)據(jù)行進(jìn)行緩存，可以編寫一個(gè)持續(xù)運(yùn)行的守護(hù)進(jìn)程函數(shù)，讓這個(gè)函數(shù)將指定的數(shù)據(jù)行緩存到Redis里面，并不定期地對(duì)這些緩存進(jìn)行更新。緩存函數(shù)將數(shù)據(jù)和編碼為json字典并存儲(chǔ)在Redis的字符串中。

我們還需要使用兩個(gè)有序集合來記錄應(yīng)該在何時(shí)對(duì)緩存進(jìn)行更新，第一個(gè)有序集合為調(diào)度有序集合，成員為數(shù)據(jù)行的ID,分值為時(shí)間戳，記錄應(yīng)該在何時(shí)將制定的數(shù)據(jù)行緩存到Redis里面。第二個(gè)有序集合為延時(shí)有序集合，成員為數(shù)據(jù)行的ID，而分值記錄指定數(shù)據(jù)行的緩存需要每隔多少秒更新一次。

對(duì)于更新頻率，如果數(shù)據(jù)行記錄的是特價(jià)促銷商品的剩余數(shù)量，并且參與促銷活動(dòng)的用戶非常多，那么我么最好每隔幾秒更新一次數(shù)據(jù)行緩存，如果數(shù)據(jù)并不經(jīng)常改變，或者商品缺貨是可以接受的，我們可以每分鐘更新一次緩存。

之前對(duì)于網(wǎng)頁(yè)的緩存，如果網(wǎng)站總共包含100000件商品，貿(mào)然緩存所有商品頁(yè)面將耗盡整個(gè)網(wǎng)站的全部?jī)?nèi)存，所以我們可以只針對(duì)那些瀏覽量較高的商品頁(yè)面進(jìn)行緩存。

每個(gè)用戶都有一個(gè)相應(yīng)的記錄用戶瀏覽商品歷史的有序集合，我們?cè)谟涗浀倪^程中，我們也痛死記錄所有商品的瀏覽次數(shù)，根據(jù)瀏覽次數(shù)對(duì)商品進(jìn)行排序，被瀏覽得最多的商品放到有序集合的索引0位置上，并且具有整個(gè)有序集合最少的分值。

除了緩存最常被瀏覽的商品外，我們還需要發(fā)現(xiàn)那些變得越來越流暢的新商品，于是我們需要定期修剪有序集合的長(zhǎng)度并調(diào)整已有元素的分值，才能使得新流行的商品在排行榜中占據(jù)一席之地。

（1）登錄令牌與用戶映射關(guān)系的散列 "login:"
（2）記錄最近登錄用戶的有序集合 "recent:"
（3）記錄各個(gè)用戶最近瀏覽商品的有序集合 "viewed:94233rhsYRIq3yi3qryrye"
（4）每個(gè)用戶的購(gòu)物車散列，存儲(chǔ)商品ID與商品訂購(gòu)數(shù)量之間的映射。"cart:94233rhsYRIq3yi3qryrye"
（5）請(qǐng)求頁(yè)面緩存集合 "cache:wre9w3rieruerwe3" (wre9w3rieruerwe3代表請(qǐng)求ID)

(94233rhsYRIq3yi3qryrye假設(shè)為某個(gè)用戶的令牌)
（6）數(shù)據(jù)行緩存字符串，數(shù)據(jù)列(column)的名字會(huì)被映射為json字典的鍵，而數(shù)據(jù)行的值會(huì)被映射為json字典的值，"inv:273" (其中273為數(shù)據(jù)行id)。
（7）數(shù)據(jù)行緩存調(diào)度有序集合，成員為數(shù)據(jù)行的ID,分值為時(shí)間戳，記錄應(yīng)該在何時(shí)將制定的數(shù)據(jù)行緩存到Redis里面，"schedule："。
（8）數(shù)據(jù)行緩存延時(shí)有序集合，成員為數(shù)據(jù)行的ID，而分值記錄指定數(shù)據(jù)行的緩存需要每隔多少秒更新一次，"delay:"。

（9）商品瀏覽次數(shù)有序集合，成員為商品，分值為瀏覽次數(shù)負(fù)值，方便保持在有序集合的較前的索引位置，"viewed"。

（1）使用散列來存儲(chǔ)登錄cookie令牌與已登錄用戶之前的映射。根據(jù)給定的令牌查找與之相應(yīng)的用戶，檢查用戶是否登錄，并返回該用戶的ID。

"""
獲取并返回令牌對(duì)應(yīng)的用戶

@param {object}
@param {string} token

@return {string} 用戶id
"""
def checkToken(conn, token):
    return conn.hget("login:", token)

（2）用戶每次瀏覽頁(yè)面的時(shí)候，需要更新“登錄令牌與用戶映射關(guān)系的散列”里面的信息，
并將用戶的令牌和當(dāng)前時(shí)間戳添加到 “記錄最近登錄用戶的有序集合” 里面，
將瀏覽商品添加到記錄“記錄各個(gè)用戶最近瀏覽商品的有序集合”中，如果記錄的商品數(shù)量超過25個(gè)，對(duì)這個(gè)有序集合進(jìn)行修剪。

"""
更新令牌時(shí)，需要更改用戶令牌信息，將用戶記錄到最近登錄用戶的有序集合中，
如果用戶瀏覽的是商品，則需要將瀏覽商品寫入該用戶瀏覽過商品的有序集合中，并保證該集合不超過25個(gè)

@param {object}
@param {string} token
@param {string} user
@param {string} item

"""
def updateToken(conn, token, user, item = None):
    timestamp = time.time()
    # 更新用戶令牌登錄對(duì)應(yīng)的用戶信息
    conn.hset("login:", token, user)
    # 增加最近訪問的用戶到有序集合
    conn.zadd("recent:", token, timestamp)

    # 如果瀏覽產(chǎn)品，記錄該用戶最近訪問的25個(gè)產(chǎn)品
    if item:
        conn.zadd("viewed:" + token, item, timestamp)
        conn.zremrangebyrank("viewed:" + token, 0, -26)
        # 記錄每個(gè)商品的瀏覽量
        conn.zincrby("viewed:", item, -1)

（3）存儲(chǔ)會(huì)話的內(nèi)存會(huì)隨著時(shí)間的推移而不斷增加，需要定期清理會(huì)話數(shù)據(jù)，我們決定只保留最新的1000萬個(gè)會(huì)話。

我們可以用 守護(hù)進(jìn)程的方式來運(yùn)行或者定義一個(gè)cron job每隔一段時(shí)間運(yùn)行 ，
檢查最近 “記錄最近登錄用戶的有序集合” 大小是否超過了限制，超過限制每秒從集合中刪除最舊的100個(gè)令牌，并且移除相應(yīng)的“登錄令牌與用戶映射關(guān)系的散列”的信息和對(duì)應(yīng)的“記錄各個(gè)用戶最近瀏覽商品的有序集合”，對(duì)應(yīng)的”美國(guó)用戶的購(gòu)物車散列“。

我們也可以使用EXPIRE命令，為用戶令牌設(shè)記錄用戶商品瀏覽記錄的有序集合設(shè)置過期時(shí)間，讓Redis在一段時(shí)間之后自動(dòng)刪除它們，這樣就不用使用有序集合來記錄最近出現(xiàn)的令牌了，但是這樣我們就沒辦法將會(huì)話數(shù)限制在1000萬之內(nèi)了。

"""
定期清理會(huì)話數(shù)據(jù)，只保留最新的1000萬個(gè)會(huì)話。

使用 *守護(hù)進(jìn)程的方式來運(yùn)行或者定義一個(gè)cron job每隔一段時(shí)間運(yùn)行* ，
檢查最近 “記錄最近登錄用戶的有序集合” 大小是否超過了限制，超過限制每秒從集合中刪除最舊的100個(gè)令牌，
并且移除相應(yīng)的“登錄令牌與用戶映射關(guān)系的散列”的信息和對(duì)應(yīng)的“記錄各個(gè)用戶最近瀏覽商品的有序集合”。

@param {object}
"""

# 循環(huán)判斷，如果是cron job可以不用循環(huán)
QUIT = False
# 限制保留的最大會(huì)話數(shù)據(jù)
LIMIT = 10000000

def cleanFullSession(conn):
    # 循環(huán)判斷，如果是cron job可以不用循環(huán)
    while not QUIT:
        # 查詢最近登錄用戶會(huì)話數(shù)
        size = conn.zcard("recent:")
        # 沒有超過限制，休眠1秒再繼續(xù)執(zhí)行
        if size <= LIMIT:
            time.sleep(1)
            continue
        
        # 查詢最舊登錄的最多100個(gè)令牌范圍
        end_index = min(size - LIMIT, 100)
        tokens = conn.zrange("recent:", 0, end_index - 1)
        
        # 將要?jiǎng)h除的key都推入到數(shù)組中，要時(shí)候一起刪除
        session_keys = []
        for token in tokens:
            session_keys.append("viewed:" + token)
            session_keys.append("cart:" + token)
        
        # 批量刪除相應(yīng)的用戶最近瀏覽商品有序集合，用戶的購(gòu)物車，登錄令牌與用戶映射關(guān)系的散列和記錄最近登錄用戶的有序集合
        conn.delete(*session_keys)
        conn.hdel("login:", *tokens)
        conn.zrem("recent:", *tokens)

（4）對(duì)購(gòu)物車進(jìn)行更新，如果用戶訂購(gòu)某件商品數(shù)量大于0，將商品信息添加到 “用戶的購(gòu)物車散列”中，如果購(gòu)買商品已經(jīng)存在，那么更新購(gòu)買數(shù)量。

"""
對(duì)購(gòu)物車進(jìn)行更新，如果用戶訂購(gòu)某件商品數(shù)量大于0，將商品信息添加到 “用戶的購(gòu)物車散列”中，如果購(gòu)買商品已經(jīng)存在，那么更新購(gòu)買數(shù)量

@param {object}
@param {string} session
@param {string} item
@param {float}  count

"""
def addToCart(conn, session, item, count):
    if count <= 0:
        # 從購(gòu)物車移除指定商品
        conn.hrem("cart:" + session, item)
    else:
        # 將指定商品添加到對(duì)應(yīng)的購(gòu)物車中
        conn.hset("cart:" + session, item, count)

（5）在用戶請(qǐng)求頁(yè)面時(shí)，對(duì)于不能被緩存的請(qǐng)求，直接生成并返回頁(yè)面，對(duì)于可以被緩存的請(qǐng)求，先從緩存取出緩存頁(yè)面，如果緩存頁(yè)面不存在，那么會(huì)生成頁(yè)面并將其緩存在Redis，最后將頁(yè)面返回給函數(shù)調(diào)用者。

"""
在用戶請(qǐng)求頁(yè)面時(shí)，對(duì)于不能被緩存的請(qǐng)求，直接生成并返回頁(yè)面，
對(duì)于可以被緩存的請(qǐng)求，先從緩存取出緩存頁(yè)面，如果緩存頁(yè)面不存在，那么會(huì)生成頁(yè)面并將其緩存在Redis，最后將頁(yè)面返回給函數(shù)調(diào)用者。

@param {object} conn
@param {string} request
@param {callback}

@return 
"""
def cacheRequest(conn, request, callback):
    # 判斷請(qǐng)求是否能被緩存，不能的話直接調(diào)用回調(diào)函數(shù)
    if not canCache(conn, request):
        return callback(request)
    
    # 將請(qǐng)求轉(zhuǎn)換為一個(gè)簡(jiǎn)單的字符串健，方便之后進(jìn)行查找
    page_key = "cache:" + hashRequest(request)
    content = conn.get(page_key)
    
    # 沒有緩存的頁(yè)面，調(diào)用回調(diào)函數(shù)生成頁(yè)面，并緩存到redis中
    if not content:
        content = callback(request)
        conn.setex(page_key, content, 300)

    return content

"""
判斷頁(yè)面是否能被緩存，檢查商品是否被緩存以及頁(yè)面是否為商品頁(yè)面，根據(jù)商品排名來判斷是否需要緩存

@param {object} conn
@param {string} request

@return {boolean}
"""
def canCache(conn, request):
    # 根據(jù)請(qǐng)求的URL，得到商品ID
    item_id = extractItemId(request)
    # 檢查這個(gè)頁(yè)面能否被緩存以及這個(gè)頁(yè)面是否為商品頁(yè)面
    if not item_id or isDynamic(request):
        return False

    # 商品的瀏覽排名
    rank = conn.zrank("viewed:", item_id)
    return rank is not None and rank < 10000

"""
解析請(qǐng)求的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]

"""
判斷請(qǐng)求的頁(yè)面是否動(dòng)態(tài)頁(yè)面

@param {string} request

@return {boolean}
"""
def isDynamic(request):
    parsed = urlparse.urlparse(request)
    query = urlparse.parse_qs(parsed.query)
    return "_" in query

"""
將請(qǐng)求轉(zhuǎn)換為一個(gè)簡(jiǎn)單的字符串健，方便之后進(jìn)行查找
@param {string} request

@return {string}
"""
def hashRequest(request):
    return str(hash(request))

（6）為了讓緩存函數(shù)定期地緩存數(shù)據(jù)行，首先需要將行ID和給定的延遲值添加到延遲有序集合中，再將行ID和當(dāng)前時(shí)間的時(shí)間戳添加到調(diào)度有序集合中。如果某個(gè)數(shù)據(jù)行的延遲值不存在，那么程序?qū)⑷∠麑?duì)這個(gè)數(shù)據(jù)行的調(diào)度。如果我們想要移除某個(gè)數(shù)據(jù)行已有的緩存并且不再緩存那個(gè)數(shù)據(jù)行，只需要把那個(gè)數(shù)據(jù)行的延遲值設(shè)置為小于或等于0即可。

"""
設(shè)置數(shù)據(jù)行緩存的延遲值和調(diào)度時(shí)間

@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）嘗試讀取”數(shù)據(jù)行緩存調(diào)度有序集合“的第一個(gè)元素以及該元素的分支，如果”數(shù)據(jù)行緩存調(diào)度有序集合“沒有包含任何元素，或者分值存儲(chǔ)的時(shí)間戳所指定的時(shí)間尚未來臨，那么函數(shù)先休眠50毫秒，然后再重新進(jìn)行檢查。

當(dāng)發(fā)現(xiàn)一個(gè)需要立即進(jìn)行更新的數(shù)據(jù)行時(shí)，如果數(shù)據(jù)行的延遲值小于或者等于0，會(huì)從”數(shù)據(jù)行緩存延時(shí)有序集合“和”數(shù)據(jù)行緩存調(diào)度有序集合“移除這個(gè)數(shù)據(jù)行的ID，并從緩存里面刪除這個(gè)數(shù)據(jù)行已有的緩存，再重新進(jìn)行檢查。

對(duì)于延遲值大于0的數(shù)據(jù)行來說，從數(shù)據(jù)庫(kù)里面取出這些行，將他們編碼為json格式并存儲(chǔ)到Redis里面，然后更新這些行的調(diào)度時(shí)間。

"""
守護(hù)進(jìn)程，根據(jù)調(diào)度時(shí)間有序集合和延遲值緩存數(shù)據(jù)行

@param {object} conn

"""
def cacheRow(conn):
    while not QUIT:
        # 需要讀取”數(shù)據(jù)行緩存調(diào)度有序集合“的第一個(gè)元素，如果沒有包含任何元素，或者分值存儲(chǔ)的時(shí)間戳所指定的時(shí)間尚未來臨，那么函數(shù)先休眠50毫秒，然后再重新進(jìn)行檢查
        next = conn.zrange("schedule:", 0, 0, withscores=True)
        now = time.time()
        if not next or next[0][1] > now:
            time.sleep(.05)
            continue
        
        row_id = next[0][0]
        # 取出延遲值
        delay = conn.zscore("delay:", row_id)
        # 如果延遲值小于等于0，則不再緩存該數(shù)據(jù)行
        if delay <= 0:
            conn.zrem("schedule:", row_id)
            conn.zrem("delay:", row_id)
            conn.delete("inv:" + row_id) 
            continue;

        # 需要緩存的，更新緩存調(diào)度的有序集合，并緩存該數(shù)據(jù)行
        row = Inventory.get(row_id)
        conn.zadd("schedule:", row_id, now + delay)
        conn.set("inv:" + row_id, json.dumps(row.toDict()))
        
"""
庫(kù)存類，庫(kù)存的商品信息
"""
class Inventory(object):
    def __init__(self, id):
        self.id = id

    @classmethod
    def get(cls, id):
        return Inventory(id)
    
    def toDict(self):
        return {"id":self.id, "data":"data to cache...","cached":time.time()}

（8）我們需要在用戶瀏覽頁(yè)面時(shí)，“商品瀏覽次數(shù)有序集合”對(duì)應(yīng)的商品中需要減一，使得保持在有序集合較前的索引位置。

同時(shí)我們需要開啟一個(gè)守護(hù)進(jìn)程，每隔5分鐘，刪除所有排名在20000名之后的商品瀏覽數(shù)，并使用ZINTERSTORE將刪除之后剩余的所有商品的瀏覽次數(shù)減半。

而判斷頁(yè)面是否需要緩存，我們需要通過ZRANK取出商品的瀏覽次數(shù)排名，如果排名在10000內(nèi)，那么說明該頁(yè)面需要緩存。

"""
守護(hù)進(jìn)程，刪除所有排名在20000名之后的商品，并將刪除之后剩余的所有商品瀏覽次數(shù)減半，5分鐘執(zhí)行一次

@param {object} conn

"""
def rescaleViewed(conn):
    while not QUIT:
        conn.zremrangebyrank("viewed:", 20000, -1)
        conn.zinterstore("viewed:", {"viewed:", .5})
        time.sleep(300)

"""
測(cè)試
"""
import time
import urlparse
import uuid
import threading
import unittest
import json

class TestShoppingWebsite(unittest.TestCase):
    def setUp(self):
        import redis
        self.conn = redis.Redis(db=15)
    
    def tearDown(self):
        conn = self.conn
        to_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.conn

        global QUIT, LIMIT
        QUIT = False
        LIMIT = 10000000
        print
        print

    def testLoginCookies(self):
        conn = self.conn
        global LIMIT, QUIT
        token = str(uuid.uuid4())

        updateToken(conn, token, "username", "itemX")
        print "We just logged-in/updated token:", token
        print "For user:", "username"
        print

        print "What username do we get when we look-up that tokan?"
        r = checkToken(conn, token)
        print r
        print
        self.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 = 0
        t = threading.Thread(target = cleanFullSession, args = (conn,))
        t.setDaemon(1)
        t.start()
        time.sleep(1)
        QUIT = True
        time.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:", s
        self.assertFalse(s)

    def testShoppingCartCookies(self):
        conn = self.conn
        global LIMIT, QUIT
        token = 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:", r
        print

        self.assertTrue(len(r) >= 1)

        print "Let"s clean out our sessions an carts"
        LIMIT = 0
        t = threading.Thread(target=cleanFullSession, args=(conn,))
        t.setDaemon(1)
        t.start()
        time.sleep(1)
        QUIT = True
        time.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:", r

        self.assertFalse(r)

    def testCacheRequest(self):
        conn = self.conn
        token = str(uuid.uuid4())

        def callback(request):
            return "content for " + request

        updateToken(conn, token, "username", "itemX")
        url = "http://test.com/?item=itemX"
        print "We are going to cache a simple request against", url
        result = cacheRequest(conn, url, callback)
        print "We got initial content:", repr(result)
        print

        self.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 pprint
        conn = self.conn
        global  QUIT

        print "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)
        print
        print "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)
        print
        self.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 r
        print
        self.assertFalse(r)

        QUIT = True
        time.sleep(2)
        if t.isAlive():
            raise Exception("The database caching thread is still alive?!?")


if __name__ == "__main__":
    unittest.main()

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