Python--Redis实战：第五章：使用Redis构建支持程序：第2节：计数器和统计数据

时间  2019-11-11
标签 python redis 实战 第五 使用 构建 支持 程序 计数器 统计数据 栏目 Python 繁體版
原文   原文链接
上一篇文章： Python--Redis实战：第五章：使用Redis构建支持程序：第1节：使用Redis来记录日志
下一篇文章： Python--Redis实战：第五章：使用Redis构建支持程序：第3节：查找IP所属城市以及国家

正如第三章所述，经过记录各个页面的被访问次数，咱们能够根据基本的访问计数信息来决定如何缓存页面。可是第三章只是一个很是简单的例子，现实状况不少时候并不是是如此简单的，特别是涉及实际网站的时候，尤其如此。前端

知道咱们的网站在最近5分钟内得到了10 000次点击，或者数据库在最近5秒内处理了200次写入和600次读取，是很是有用的。经过在一段时间内持续地记录这些信息，咱们能够注意到流量的突增和渐增状况，预测什么时候须要对服务器进行升级，从而预防系统由于负载超载而下线。web

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

将计数器存储到Redis里面

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

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

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

对计数器进行更新

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

键名：count:5:hits 类型：hash
1336376410 45
1336376405 28
1336376395 17（本行数据表示：网站在2012年5月7日早晨7:39:55到7:40：00总共得到了17次点击）
1336376400 29

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

键名：known: 类型：zset(有序集合)
1:hits 0
5:hits 0
60:hits 0

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

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

#以秒为单位的计数器精度，分别为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（访问时间）统计数据。

表名：starts:ProfilePage:AccessTime 类型：zset
min 0.035
max 4.958
sunsq 194.268
sum 258.973
count 2323

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

下面代码展现了负责更新统计数据的代码。和以前介绍过的常见日志程序同样，统计程序在写入数据以前会进行检查，确保被记录的是当前这小时的统计数据，并将不属于当前这一小时的旧数据进行归档。在此以后，程序会构建两个临时有序集合，其中一个用于保存最小值，而另外一个则用于保存最大值而后使用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所属城市以及国家
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程