Redis实战学习笔记二

使用Redis构建支持程序

案例一：使用Redis实现日志的记录

需求： 在构建应用程序和服务的过程中，对正在运行的系统的相关信息进行挖掘变得越来越重要。这都依赖于日志。

分析： 许多日志的记录的方式都是将日志记录到文件中去，然后随着时间的流逝新建日志文件（因为日志文件大大小不可能无限的增长），因为每个服务都有相应的日志记录，并且每种服务的日志轮换机制也是不同的，这就会缺少了一种将这些日志聚合的方案。
还有一种日志记录方式是利用syslog服务来进行日志的记录。syslog服务接收各个服务的日志信息，并路由到不同的磁盘上，它还负责日志的轮换和删除工作。它要比直接写文件方便的多。

我们也可以利用Redis来实现日志的记录。

实现方式：

SEVERITY={   # 设置一个字典，将日志的安全级别映射为字符串
    loggin.DEBUG:'debug',
    loggin.INFO：'info',
    loggin.WARNING:'warning',
    loggin.ERROR:'error',
    loggin.CRITICAL:'critical'
}
SERVERITY.update((name,name) for name in SERVERITY.values())
def log_recent(conn,name,message,severity=loggin.INFO,pipe=None):
    # 将日志的安全级别映射为简单的字符串
    severity=str(SEVERITY.get(severity,severity)).lower() 
    # 创建负责存储消息的键
    destination='recent:%s:%s'%(name,severity)
    # 在消息中添加当前时间
    message=time.asctime()+' '+message
    # 使用流水线通信方式
    pipe=pipe or conn.pipeline()
    pipe=lpush(destination,message)
    # 对日志列表进行修剪，只保留100条消息
    pipe.ltrim(destination,0,99)
    # 执行两个命令
    pipe.execute()

我们还可以统计每种消息出现的频率，并根据消息频率来进行消息的排序，从而找出最重要的信息。

def log_common(conn,name,message,severity=logging.INFO,timeout=5):
    severity=ser(SEVERITY.get(severity,severity)).lower()
    destination='common:%s:%s'%(name,severity)
    start_key=destination+':start'
    pipe=conn.pipeline()
    end = time.time()+timeout
    while time.time() <end
        try:
            pipe.watch(start_key)
            now=datetime.utcnow().timetuple()
            hour_start=datetime(*now[:4]).isoformat()
            existing=pipe.get(start_key)
            pipe.multi() # 创建一个事务
            if existing and existing < hour_start:
                pipe.rename(destination,destination+':last')
                pipe.rename(start_key,destination+':pstart')
                pipe.set(start_key,hour_start)
                # 对日志出现的次数的计算器自增
            pipe.zincrby(destination,message) 
            log_recent(pipe,name,message,severity,pipe)
            return 
        except redis.exceptions.WatchError:
            # 如果因为其它客户端正在执行归档而出现监视错误，就重试
            continue

案例二：计数器和统计数据

**需求：**为了收集指标数据并进行监视和分析，我们构建一个能够持续创建并维护计数器的工具，这个工具创建的每个计数器都有自己的名字。这些技术器以指定的精度存储指定的数据样本。
分析： 为了记录点击量，我们使用了一个hash，它的名字为count:5:hit表示记录每5秒为一个时间断内的点击量，其中键为时间段，值为点击量。

PRECISION=[1,5,60,300,3600,16000,86400] # 计数器精度
'''
更新计数器
'''
def update_counter();
    now=now or time.time()
    # 创建一个事务型流水线
    pipe=conn.pipeline()
    for prec in PRECISION
        # 获取当前时间片开始的时间 
        pnow =int(npw/prec)*prec
        hash='%s:%s'%(prec,name)
        pipe.zadd('known:',hash,0)
        pipe.hincrby('count:'+hash,pnow,count)
    pipe.execute()

'''
获得指定计数器中的内容
'''
def get_counter(conn,name,precision):
    # 获得取得存储计数器的键的名字 
    hash='%s:%s'%(precision,name)
    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
    
'''
清理旧的计数器
'''
def clean_counters():
    pipe=conn.pipeline(True)
    passes=0
    while not QUIT:
        start=time.time()
        index=0
        while index<conn.zcard('konw:'):
            hash=conn.zrange('know:',index,index)
            index+=1
            if not hash:
                break
            hash=hash[0]
            prec=int(hash.partition(':')[0]) #获取计数器的精度
            bprec=int(prec//60) or 1
            if passes % bprec:
                continue
            hkey='count:'+hash
            cotoff=time.time() -SAMPLE_COUNT*prec
            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):
                            pipe.multi()
                            pipe.zrem('know:',hash)
                            pipe.execute()
                            index=-1
                        else:
                            pipe.unwatch()
                    except redis.exceptions.WatchError:
                        pass
                passes+=1
                duration=min(int(time.time()-start)+1,60)
                time.sleep(max(60-duration,1))

使用Redis构建应用程序组件

案例一：自动补全最近联系人

在web领域里，自动补全是一种能够让用户不进行搜索的情况下，就能快速找到所需东西的计数。自动补全一般会根据用户已输入的字母来查找所有已经输入字母为开头的单词。
**需求：**实现一个用于记录最近练习人的自动补全程序。
**分析：**构建最近练习人自动补全列表通常需要对Redis执行三个步骤：

如果指定的联系人已经存在于最近联系人列表里面，那么从列表里面移除它。
将指定的联系人添加到最近联系人列表的最前面
将添加操作完成后，如果最近联系人列表包含的联系人数量超过了100个，那么对列表进行修剪，只保留位于列表前面的100个联系人。

实现方案：

def add_update_contact(conn,user,contact):
    ac_list='recent:'+user
    pipeline=conn.pipeline(True)
    pipeline=lrem(ac_list,contact) # 如果该联系人存在，则移除它
    pipeline.lpush(ac_list,contact) # 将联系人推入列表最前端
    pipeline.ltrim(ac_list,0,99) # 只保留前面的100个联系人
    pipeline.execute()

def fetch_autocomplete_list(conn,user,prefix):
    candidates=conn.lrange('recent:'+user,0,-1) # 自动获取补全列表
    matches=[]
    for candidate in candidates:
        if candidate.lower().startswith(prefix):
            matches.append(candidate)
        return matches 3 返回所有匹配到的联系人

案例二：分布式锁

分布式锁会执行“先获取锁，然后执行操作，最后释放锁”动作。但是这种锁石油不同的机器上的不同的Redis客户端进行获取和释放的，

**需求：**为了对Redis存储的数据进行排它性访问，客户端需要访问一个锁，这个锁必须定义在一个可以让所有客户端都看得到的范围之内，而这个范围就是Redis本身，因此我们需要将锁构建在Redis里面。

简单的锁：


'''
获取锁
'''
def acquire_lock(conn,lockname,acquire_timeout=10):
    identifier=srt(uuid.uuid4()) #128位随机标识符
    end=time.time()+acquire_timeout
    while time.time() <end:
        if conn.setnx('lock:'+lockname,identifier): # 尝试获得锁
            return identifier
        time.sleep(.001)
    return False

它使用了SETNX命令，尝试在代表锁的键不存在的情况下，为键设置一个值，以此来获取锁。

def release_lock(conn,lockname,identifier):
    pipe=conn.pipeline(True)
    lockname='lock:'+lockname
    while True:
        try:
            pipe.watch(lockname)
            # 检查进程是否仍然持有锁
            if pipe.get(lockname) == identifier:
                pipe.multi()
                pipe.delete(lockname)
                pipe.execute()
                return True
            pipe.unwatch()
            break
        # 有其它客户端修改了锁，重试
        except redis.exceptions.WatchError: 
            pass
    return False # 进程已经失去了锁

目前这种锁的实现方式，当锁的持有者崩溃后锁不会被自动释放，这回导致锁一直处于被获取的状态。

带有超时限制的锁

def acquire_lock(conn,lockname,acquire_timeout=10,lock_timeout=10):
    identifier=srt(uuid.uuid4()) #128位随机标识符
    # 确保传给EXPIRE的都是整数
    lock_timeout=int(math.ceil(lock_timeout))
    end=time.time()+acquire_timeout
    while time.time() <end:
        if conn.setnx('lock:'+lockname,identifier): # 尝试获得锁
            conn.expire(lockname.lock_timeout)
            return identifier
        elseif:
            # 检查过期时间，并在有需要的时候进行更新
            conn.expire(lockname,lock_timeout)
        time.sleep(.001)
    return False

案例三：计数信号量

计数信号量也是一种锁，它可以限制一项资源同时被多个进程访问，通常用于限定能够同时使用的资源数量。

构建简单的计数信号量：

def acquire_semaphore(conn,semname,limit,timeout=10):
    identifier=str(uuid.uuid4()) # 128位随机标识符
    now =time.time()
    pipeline=conn.pipeline(True)
    # 清理过期的信号量持有者
    pipeline=zremrangebyscore(semname,'-inf',now-timeout)
    pipeline=zadd(semname,identifier,now)
    pipeline.zrank(semname,identifier)
    if pipeline.execute()[-1]<limit:
        return identifier
    # 获取信号量失败，删除之前添加的标识符
    conn.zrem(semname,identifier) 
    return None
    
def release_semphore(conn,semname,identifier):
    return conn.zrem(semname,identifier)

这种信号量的实现存在一些问题，它假设假设每个进程访问到系统的时间都是相同的，而这一假设在多主机环境下可能并不成立。
每当锁或信号量因为系统时钟的细微不同而导致锁的获取结果出现剧烈变化时，这个锁或者信号量就是不公平的。不公平的锁或信号量可能会导致客户端永远无法获取到它原本应该得到的锁或信号量。

公平信号量：
为了尽可能地减少系统时间不一致带来的问题，我们需要给信号量实现添加一个计数器以及一个有序集合。其中，计数器通过持续地执行自增操作，创建一种类似于计时器的机制，确保最先对计数器执行自增操作的客户段能够获得信号量。

def acquire_fair_semaphore(conn, semname, limit, timeout=10):
    identifier = str(uuid.uuid4())                             
    czset = semname + ':owner'
    ctr = semname + ':counter'
    now = time.time()
    pipeline = conn.pipeline(True)
    # 删除超时信号量
    pipeline.zremrangebyscore(semname, '-inf', now - timeout)  
    pipeline.zinterstore(czset, {czset: 1, semname: 0}) 
    # 对计数器实现自增
    pipeline.incr(ctr)                                         
    counter = pipeline.execute()[-1]   
    # 尝试获取信号量
    pipeline.zadd(semname, {identifier: now})                  
    pipeline.zadd(czset, {identifier: counter})                
    pipeline.zrank(czset, identifier)                          
    if pipeline.execute()[-1] < limit:                         
        return identifier                                      
    pipeline.zrem(semname, identifier)                         
    pipeline.zrem(czset, identifier)                           
    pipeline.execute()
    return None
    
'''
释放公平锁
'''
def release_fair_semaphore(conn, semname, identifier):
    pipeline = conn.pipeline(True)
    pipeline.zrem(semname, identifier)
    pipeline.zrem(semname + ':owner', identifier)
    return pipeline.execute()[0]
'''
刷新信号量
'''
def refresh_fair_semaphore(conn,semname,identifier):
    if conn.zadd(semname,identifier,time.time()):
        release_fair_semaphore(conn,semname,identifier)
        return Flase
    return True


'''
'''
acquire_semaphore_with_lock():
    identifier=acquire_lock(conn,semname,acquire_timeout=.01)
    if identifier:
        try:
            return acquire_fair_semaphore(conn,semname,limit,timeout)
        finally:
            release_lock(conn,semname,identifier)