上一篇文章：
Python–Redis实战：第五章：使用Redis构建支持程序：第1节：使用Redis来记录日志

下一篇文章：
Python–Redis实战：第五章：使用Redis构建支持程序：第3节：查找IP所属城市以及国家

正如第三章所述，通过记录各个页面的被访问次数，我们可以根据基本的访问计数信息来决定如何缓存页面。但是第三章只是一个非常简单的例子，现实情况很多时候并非是如此简单的，特别是涉及实际网站的时候，尤为如此。

知道我们的网站在最近5分钟内获得了10 000次点击，或者数据库在最近5秒内处理了200次写入和600次读取，是非常有用的。通过在一段时间内持续地记录这些信息，我们可以注意到流量的突增和渐增情况，预测何时需要对服务器进行升级，从而预防系统因为负载超载而下线。

这一节将分别介绍使用Redis来实现计数器的方法以及使用Redis来进行数据统计的方法，并在最后讨论如何简化示例中的数据统计操作。本节展示的例子都是由实际的用例和需求驱动的。首先，让我们来看看，如何使用Redis来实现时间序列计数器，以及如何使用这些计数器来记录和监视应用程序的行文。

将计数器存储到Redis里面

在监控应用程式的同时，持续的收集信息是一件非常重要的事情。那些影响网站响应速度以及网站所能服务的页面数量的代码改动、新的广告营销活动或者是刚刚接触系统的新用户，都有可能会彻底地改变网站载入页面的数量，并因此而影响网站的各项性能指标。但如果我们平时不记录任何指标数据的话，我们就不可能知道指标发生了变化，也就不知道网站的性能是在提高还是在下降。

为了收集指标数据并进行监视和分析，我们将构建一个能够持续创建并维护计数器的工具，这个工具创建的每个计数器都有自己的名字（名字带有网站点击量、销量或者数据库查询字样的计数器都是比较重要的计数器）。这些计数器会以不同的精度（如1秒、5秒、1分钟等）存储最新的120个数据样本，用户也可以根据自己的需要，对取样的数量和精度进行修改。

实现计数器首先需要考虑的就是如何存储计数器的信息，接下来将说明我们是如何将计数器信息存储在Redis里面

对计数器进行更新

为了对计数器进行更新，我们需要存储实际的计数器信息，对于每个计数器以及每种精度，如网站点击量计数器/5秒，我们将使用一个散列来存储网站在每个5秒时间片之内获得的点击量，其中，散列的每个键都是某个时间片的开始时间，而键对应的值则存储了网站在该时间片之内获得的点击量。下表展示了一个点击量计数器存储的其中一部分数据，这个计数器以每5秒为一个时间片记录着网站的点击量：

为了能够清理计数器包含的旧数据，我们需要在使用计数器的同时，对被使用的计数器进行记录。为了做到这一点，我们需要一个有序序列，这个序列不能包含任何重复元素，并且能够让我们一个接一个地遍历序列中包含的所有元素。虽然同时使用列表和集合可以实现这种序列，但同时使用两种数据结构需要编写更多代码，并且增加客户端和Redis之间的通信往返次数。实际上，实现有序序列更好的方法时使用有序集合，有序集合的各个成员分别由计数器的精度以及计数器的名字组成，而所有成员的分值都是0.因为所有成员的分值都被设置为0，所以Redis在尝试按分值对有序集合进行排序的时候，就会发现这一点，并改为使用成员名进行排序，这使得一组给定的成员总是具有固定的排列顺序，从而可以方便地对这些成员进行顺序性的扫描。下表展示了一个有序集合，这个有序集合记录了正在使用的计数器。

既然我们已经知道应该使用什么结构来记录并表示计数器了，现在是时候来考虑一下如何使用和更新这些计数器了。

下面代码展示了程序更新计数器的方法，对于每种时间片精度，程序都会将计数器 的精度和名字作为引用信息添加都记录已有计数器的有序集合里面，并增加散列计数器在指定时间片内的计数值。

#以秒为单位的计数器精度，分别为1秒/5秒/1分钟/5分钟/1小时/5小时/1天
#用户可以按需调整这些精度
import time

PRECISION=[1,5,60,300,3600,18000,86400]

def update_counter(conn,name,count=1,now=None):
    #通过获取当前时间来判断应该对哪个时间片执行自增操作。
    now=now or time.time()
    #为了保证之后的清理工作可以正确的执行，这里需要创建一个事务性流水线
    pipe=conn.pipeline()
    #为我们记录的每种精度都创建一个计数器
    for prec in PRECISION:
        #取得当前时间片的开始时间
        pnow=int(now/prec)*prec
        #创建负责存储计数信息的散列
        hash='%s:%s'%(prec,name)
        # 将计数器的引用信息添加到有序集合里面，并将其分值设为0，以便在之后执行清理操作
        pipe.zadd('known:',hash,0)
        #对给定名字和精度的计数器进行更新
        pipe.hincrby('count:'+hash,pnow,count)
    pipe.execute()

更新计数器信息的过程并不复杂，程序只需要为每种时间片精度执行zadd命令和hincrby命令就可以了。于此类似，从指定精度和名字的计数器里面获取计数数据也是一件非常容易地事情。下面代码展示了用于执行这一操作的代码：程序首先使用hgetall命令来获取整个散列，接着将命令返回的时间片和计数器的值从原来的字符串格式转换成数字格式，根据时间对数据进行排序，最后返回排序后的数据：

def get_counter(conn,name,precision):
    #取得存储计数器数据的键的名字
    hash='%s:%s'%(precision,name)
    #从Redis里面取出计数器数据
    data=conn.hgetall('count:'+hash)
    to_return=[]
    #将计数器数据转换成指定的格式
    for key,value in data.iteritems():
        to_return.append((int(key),int(value)))
    #对数据进行排序，把旧的数据样本排在前面
    to_return.sort()
    return to_return

get_counter()函数的工作方式就和之前描述的一样，它获取计数器数据并将其转换成整数，然后根据时间先后对转换后的数据进行排序。

在弄懂了获取计数器存储的数据之后，接下来我们要考虑的是如何防止这些计数器存储过多的数据。

清理旧计数器

经过前面的介绍，我们已经知道了怎样将计数器存储到Redis里面，已经怎样从计数器里面取出数据。但是，如果我们只是一味地对计数器进行更新而不执行任何清理操作的话，那么程序最终将会因为存储了过多的数据而导致内存不足。好在我们事先已将所有已知的计数器记录到了一个有序集合里面，所以对计数器进行清理只需要遍历有序集合并删除其中的旧计数器旧可以了。

为什么不使用expire？

expire命令的其中一个限制就是它只能应用整个键，而不能只对键的某一部分数据进行过期处理。并且因为我们将同一个计数器在不同精度下的所有计数器数据都存放到了同一个键里面，所以我们必须定期地对计数器进行清理。如果读者感兴趣的话，也可以试试改变计数器组织数据的方式，使用Redis的过期键功能来代替手工的清理操作。

在处理和清理旧数据的时候，有几件事情是需要我们格外留心的，其中包括以下几件：

1. 任何时候都可能会有新的计数器被添加进来
2. 同一时间可能会有多个不同的清理操作在执行
3. 对于一个每天只更新一次的计数器来说，以每分钟一次的频率尝试清理这个计数器只会浪费计算资源。
4. 如果一个计数器不包含任何数据，那么程序就不应该尝试对它进行清理。

我们接下来要构建一个守护进程函数，这个函数的工作方式和第三章中展示的守护进程函数类似，并且会严格遵守上面列出的各个注意事项。和之前展示的守护进程函数一样，这个守护进程函数会不断地重复循环知道系统终止这个进程为止。为了尽可能地降低清理操作的执行负载，守护进程会以每分钟一次的频率清理那些每分钟更新一次或者每分钟更新多次的计数器，而对于那些更新频率低于每分钟一次的计数器，守护进程则会根据计数器自身的更新频率来决定对他们进行清理的频率。比如说，对于每秒更新一次或者每5秒更新一次的计数器，守护进程将以每分钟一次的频率清理这些计数器；而对于每5分钟更新一次的计数器，守护进程将以每5分钟一次的频率清理这些计数器。

清理程序通过对记录已知计数器的有序集合执行zrange命令来一个接一个的遍历所有已知的计数器。在对计数器执行清理操作的时候，程序会取出计数器记录的所有计数样本的开始时间，并移除那些开始时间位于指定截止时间之前的样本，清理之后的计数器最多只会保留最新的120个样本。如果一个计数器在执行清理操作之后不再包含任何样本，那么程序将从记录已知计数器的有序集合里面移除这个计数器的引用信息。以上给出的描述大致地说明了计数器清理函数的运作原理，至于程序的一些边界情况最好还是通过代码来说明，要了解该函数的所有细节，请看下面代码：

import bisect
import time

import redis

QUIT=True
SAMPLE_COUNT=1

def clean_counters(conn):
    pipe=conn.pipeline(True)
    #为了平等的处理更新频率各不相同的多个计数器，程序需要记录清理操作执行的次数
    passes=0
    #持续地对计数器进行清理，知道退出为止
    while not QUIT:
        #记录清理操作开始执行的时间，这个值将被用于计算清理操作的执行时长
        start=time.time()
        index=0
        #渐进的遍历所有已知计数器
        while index<conn.zcard('known:'):
            #取得被检查的计数器的数据
            hash=conn.zrange('known:',index,index)
            index+=1
            if not hash:
                break
            hash=hash[0]
            #取得计数器的精度
            prec=int(hash.partition(':')[0])
            #因为清理程序每60秒就会循环一次，所以这里需要根据计数器的更新频率来判断是否真的有必要对计数器进行清理
            bprec=int(prec//60) or 1
            #如果这个计数器在这次循环里不需要进行清理，那么检查下一个计数器。
            #举个例子：如果清理程序只循环了3次，而计数器的更新频率是5分钟一次，那么程序暂时还不需要对这个计数器进行清理
            if passes % bprec:
                continue
            hkey='count:'+hash
            #根据给定的精度以及需要保留的样本数量，计算出我们需要保留什么时间之前的样本。
            cutoff=time.time()-SAMPLE_COUNT*prec
            #将conn.hkeys(hkey)得到的数据都转换成int类型
            samples=map(int,conn.hkeys(hkey))
            samples.sort()
            #计算出需要移除的样本数量。
            remove=bisect.bisect_right(samples,cutoff)
            #按需要移除技术样本
            if remove:
                conn.hdel(hkey,*samples[:remove])
                #这个散列可能以及被清空
                if remove==len(samples):
                    try:
                        #在尝试修改计数器散列之前，对其进行监视
                        pipe.watch(hkey)
                        #验证计数器散列是否为空，如果是的话，那么从记录已知计数器的有序集合里面移除它。
                        if not pipe.hlen(hkey):
                            pipe.multi()
                            pipe.zrem('known:',hash)
                            pipe.execute()
                            #在删除了一个计数器的情况下，下次循环可以使用与本次循环相同的索引
                            index-=1
                        else:
                            #计数器散列并不为空，继续让它留在记录已知计数器的有序集合里面
                            pipe.unwatch()
                    except redis.exceptions.WatchError:
                        #有其他程序向这个计算器散列添加了新的数据，它已经不再是空的了，
                        # 继续让它留在记录已知计数器的有序集合里面。
                        pass
            passes+=1
            # 为了让清理操作的执行频率与计数器更新的频率保持一致
            # 对记录循环次数的变量以及记录执行时长的变量进行更新。
            duration=min(int(time.time()-start)+1,60)
            #如果这次循环未耗尽60秒，那么在余下的时间内进行休眠，如果60秒已经耗尽，那么休眠1秒以便稍作休息
            time.sleep(max(60-duration,1))

正如之前所说，clean_counters()函数会一个接一个地遍历有序集合里面记录的计数器，查找需要进行清理的计数器。程序在每次遍历时都会对计数器进行检查，确保只清理应该清理的计数器。当程序尝试清理一个计数器的时候，它会取出计数器记录的所有数据样本，并判断哪些样本是需要被删除的。如果程序在对一个计数器执行清理操作之后，然后这个计数器已经不再包含任何数据，那么程序会检查这个计数器是否已经被清空，并在确认了它已经被清空之后，将它从记录已知计数器的有序集合中移除。最后，在遍历完所有计数器之后，程序会计算此次遍历耗费的时长，如果为了执行清理操作而预留的一分钟时间没有完全耗尽，那么程序将休眠直到这一分钟过去为止，然后继续进行下次遍历。

现在我们已经知道怎样记录、获取和清理计数器数据了，接下来要做的视乎就是构建一个界面来展示这些数据了。遗憾的是，这些内容设计到前端，并不在本内容介绍范围内，如果感兴趣，可以试试jqplot、Highcharts、dygraphs已经D3，这几个JavaScript绘图库无论是个人使用还是专业使用都非常合适。

在和一个真实的网站打交道的时候，知道页面每天的点击可以帮助我们判断是否需要对页面进行缓存。但是，如果被频繁访问的页面只需要花费2毫秒来进行渲染，而其他流量只要十分之一的页面却需要花费2秒来进行渲染，那么在缓存被频繁访问的页面之前，我们可以先将注意力放到优化渲染速度较慢的页面上去。在接下来的一节中，我们将不再使用计数器来记录页面的点击量，而是通过记录聚合统计数据来更准确地判断哪些地方需要进行优化。

使用Redis存储统计数据

首先需要说明的一点是，为了统计数据存储到Redis里面，笔者曾经实现过5种不同的方法，本节介绍的方法综合了这5种方法里面的众多优点，具有非常大的灵活性和可扩展性。

本节所展示的存储统计数据的方法，在工作方式上与上节介绍的log_common()函数类似：这两者存储的数据记录的都是当前这一小时以及前一小时所产生的事情。另外，本节介绍的方法会记录最小值、最大值、平均值、标准差、样本数量以及所有被记录值之和等众多信息，以便不时之需。

对于一种给定的上下文和类型，程序将使用一个有序集合来记录这个上下文以及这个类型的最小值、最大值、样本数量、值的和、值的平方之和等信息，并通过这些信息来计算平均值以及标准差。程序将值存储在有序集合里面并非是为了按照分值对成员进行排序、而是为了对存储着统计信息的有序集合和其他有序集合进行并集计算，并通过min和max这两个聚合函数来筛选相交的元素。下表展示了一个存储统计数据的有序集合实例，它记录了ProfilePage(个人简历)上下文的AccessTime（访问时间）统计数据。

既然我们已经知道了程序要存储的是什么类型的数据，那么接下来要考虑的就是如何将这些数据写到数据结构里面了。

下面代码展示了负责更新统计数据的代码。和之前介绍过的常见日志程序一样，统计程序在写入数据之前会进行检查，确保被记录的是当前这小时的统计数据，并将不属于当前这一小时的旧数据进行归档。在此之后，程序会构建两个临时有序集合，其中一个用于保存最小值，而另一个则用于保存最大值然后使用zunionstore命令以及它的两个聚合函数min和max，分别计算两个临时有序集合与记录当前统计数据的有序集合之前的并集结果。通过使用zunionstore命令，程序可以快速的更新统计数据，而无须使用watch去监视可能会频繁进行更新的存储统计数据的键，因为这个键可能会频繁地进行更新。程序在并集计算完毕之后就会删除那些临时有序集合，并使用zincrby命令对统计数据有序集合里面的count、sum、sumsq这3个成员进更新。

import datetime
import time
import uuid

import redis


def update_status(conn,context,type,value,timeout=5):
    #负责存储统计数据的键
    destination='stats:%s:%s'%(context,type)
    #像common_log()函数一样，处理当前这一个小时的数据和上一个小时的数据
    start_key=destination+':start'
    pipe=conn.pipeline(True)
    end=time.time()+timeout
    while time.time()<=end:
        try:
            pipe.watch(start_key)
            now=datetime.utcnow().timetuple()
            # 像common_log()函数一样，处理当前这一个小时的数据和上一个小时的数据
            hour_start=datetime(*now[:4]).isoformat()

            existing=pipe.get(start_key)
            pipe.multi()
            if existing and existing<hour_start:
                # 像common_log()函数一样，处理当前这一个小时的数据和上一个小时的数据
                pipe.rename(destination,destination+':last')
                pipe.rename(start_key,destination+':pstart')
                pipe.set(start_key,hour_start)

            tkey1=str(uuid.uuid4())
            tkey2=str(uuid.uuid4())
            #将值添加到临时键里面
            pipe.zadd(tkey1,'min','value')
            pipe.zadd(tkey2,'max','value')
            #使用聚合函数min和max，对存储统计数据的键以及两个临时键进行并集计算
            pipe.zunionstore(destination,[destination,tkey1],aggregate='min')
            pipe.zunionstore(destination,[destination,tkey2],aggregate='max')

            #删除临时键
            pipe.delete(tkey1,tkey2)
            #对有序集合中的样本数量、值的和、值的平方之和3个成员进行更新。
            pipe.zincrby(destination,'count')
            pipe.zincrby(destination,'sum',value)
            pipe.zincrby(destination,'sumsq',value*value)

            #返回基本的计数信息，以便函数调用者在有需要时做进一步的处理
            return pipe.execute()[-3:]

        except redis.exceptions.WatchError:
            #如果新的一个小时已经开始，并且旧的数据已经被归档，那么进行重试
            continue

update__status()函数的前半部分代码基本上可以忽略不看，因为它们和上节介绍的log_common()函数用来轮换数据的代码几乎一模一样，而update__status()函数的后半部分则做了我们前面描述过的事情：程序首先创建两个临时有序集合，然后使用适当的聚合函数，对存储统计数据的有序集合以及两个临时有序集合分别执行zunionstore命令；最后，删除临时有序集合，并将并集计算所得的统计数据更新到存储统计数据的有序集合里面。update__status()函数展示了将统计数据存储到有序集合里面的方法，但如果想要获取统计数据的话，又应该怎么做呢？

下面代码展示了程序取出统计数据的方法：程序会从记录统计数据的有序集合里面取出所有被存储的值，并计算出平均值和标准差。其中，平均值可以通过值的和(sum)除以取样数量(count)来计算得出；而标准差的计算则更复杂一些，程序需要多做一些工作才能根据已有的统计信息计算出标注差，但是为了简洁起见，这里不会解释计算标准差时用到的数学知识。

import datetime
import time
import uuid

import redis

def get_stats(conn,context,type):
    #程序将从这个键里面取出统计数据
    key='stats:%s:%s'%(context,type)
    #获取基本的统计数据，并将它们都放到一个字典里面
    data=dict(conn.zrange(key,0,-1,withscores=True))
    #计算平均值
    data['average']=data['sum']/data['count']
    #计算标准差的第一个步骤
    numerator=data['sumsq']-data['sun']**2/data['count']
    #完成标准差的计算工作
    data['stddev']=(numerator/data['count']-1 or 1)** .5
    return data

除了用于计算标准差的代码之外，get_stats()函数并没有什么难懂的地方，如果读者愿意花些时间在网上了解什么叫标准差的话，那么读懂这些标准差的代码应该也不是什么难事。尽管有了那么多统计数据，但我们可能还不太清楚自己应该观察哪些数据，而接下来的一节就会解答这个问题。

简化统计数据的记录与发现

在将统计数据存储到Redis里面之后，接下来我们该做些什么呢？说的更详细一点，在知道了访问每个页面所需的时间之后，我们要怎样才能找到那些生成速度较慢的网页？或者说，当某个页面的生成速度变得比以往要慢的时候，我们如何才能知悉这一情况？简单的说，为了发现以上提到的这些情况，我们需要存储更多信息，而具体的方法将这一节里面介绍。

要记录页面的访问时长，程序就必须在页面被访问时进行计时。为了做到这一点，我们可以在各个不同的页面设置计时器，并添加代码来记录计时的结果，但最好的办法是直接实现一个能够进行计时并将计时结果存储起来的东西，让它将平均访问速度最慢的页面都记录到一个有序集合里面，并向我们报告哪些页面的载入时间变得比以前更长了。

为了计算和记录访问时长，我们会编写一个Python上下文管理器，并使用这个上下文管理器来包裹那些需要计算并记录访问时长的代码。

在Python里面，一个上下文管理器就是一个专门定义的函数或者类，这个函数或者类的不同部分可以在一段代码执行之前以及执行之后分别执行。上下文管理器使得用户可以很容易地实现类似【自动关闭已打开的文件】这样的功能。

下面代码展示了用于计算和记录访问时长的上下文管理器：程序首先会取得当前时间，接着执行被包裹的代码，然后计算这些代码的执行时长，并将结果记录到
Redis里面；除此之外，程序还会对记录当前上下文最大访问的时间的有序集合进行更新。

import contextlib
import time

#将这个Python生成器用作上下文管理器
@contextlib.contextmanager
def access__time(conn,context):
    #记录代码块执行前的时间
    start=time.time()
    #运行被包裹的代码块
    yield

    #计算代码块的执行时长
    data=time.time()-start
    #更新这一上下文的统计数据
    stats=update_stats(conn,context,'AccessTime',data)
    #计算页面的平局访问时长
    average=stats[1]/stats[0]

    pipe=conn.pipeline(True)
    #将页面的平均访问时长添加到记录最长访问时间的有序集合里面
    pipe.zadd('slowest:AccessTime',context,average)
    #AccessTime有序集合只会保留最慢的100条记录
    pipe.zremrangebyrank('slowessTime',0,-101)
    pipe.execute()

因为access__time()上下文管理器里面有一些没办法只用三言两语来解释的概念，所以我们最好还是直接通过使用这个管理器来了解它是如何运作的。接下来的这段代码展示了使用access__time()上下文管理器记录web页面访问时长的方法，负责处理被记录页面的是一个回调函数：

#这个视图接收一个Redis连接以及一个生成内容的回调函数作为参数
def process_view(conn,callback):
    #计算并记录访问时长的上下文管理器就是这一包裹代码块的
    with access_time(conn,request.path):
        #当上下文管理器中的yield语句被执行时，这个语句就会被执行
        return callback()

如果还不理解，看下面简单的实例：

import contextlib


@contextlib.contextmanager
def mark():
    print("1")
    yield
    print(2)

def test(callback):
    with mark():
        return callback()

def xxx():
    print('xxx')

if __name__ == '__main__':
    test(xxx)

运行结果：

1
xxx
2

在看过这个例子之后，即使读者没有学过上下文管理器的创建方法，但是至少也已经知道该如何去使用它了。这个例子使用了访问时间上下文管理器来计算生成一个页面需要花费时多长时间，此外，同样的上下文管理器还可以用于计算数据库查询花费的时长，或者用来计算渲染一个模板所需的时长。作为练习，你能否构思一些其他种类的上下文管理器，并使用它们来记录有用的统计信息呢？另外，你能否让程序在页面的访问时长比平均情况要高出两个标注差或以上时，在recent_log()函数里面记录这一情况呢？

对现实世界中的统计数据进行收集和计数

尽管本书已经花费了好几页篇幅来讲述该如何收集生产系统运作时产生的相当重要的统计信息，但是别忘了已经有很多现成的软件包可以用于收集并绘制计数器以及统计数据，我个人最喜欢的是Graphite，在时间尝试构建自己的数据绘图库之前，不妨先试试这个。

在学会了如何将应用程序相关的各种重要信息存储到Redis之后，在接下来一节中，我们将了解更多与访客有关的信息，这些信息可以帮助我们处理其他问题。

上一篇文章：
Python–Redis实战：第五章：使用Redis构建支持程序：第1节：使用Redis来记录日志

下一篇文章：
Python–Redis实战：第五章：使用Redis构建支持程序：第3节：查找IP所属城市以及国家