redis学习(五)
使用redis记录日志
常见日志
使用一个列表来存储日志，并且使用ltrim限制日志的条数，使用pipe原子执行。
//name,log_rate组合成日志的列表名，log_rate是日志的级别
function (name,message string,log_rate int,pipe Pipeline){
String key="recent:%s,%d".format(name,log_rate)
pipe.lpush(key,message)
pipe.ltrim(key,0,99)
pipe.execute()
}
记录特定消息出现的频率，并根据出现频率的高低来决定消息的排列顺序，从而帮助我们找出最重要的消息。
在上述的函数中，在存入列表前，根据日志的某些特征，找出日志的重复次数，然后将日志记录到一个有序集合里。
计数器和统计数据
构建一个计数器
构建一个能够持续创建并维护计数器的工具，这个工具创建的每个计数器都有自己的名字（名字里带有网站点击
量、销量或者数据库查询字样的计数器都是比较重要的计数器）。这些计数器会以不同的时间精度（如1秒、5秒、1分钟等）存储最新的120个数据样本。
对于每个计数器以及每种精度，如网站点击量计数器和5秒，我们将使用一个散列来存储网站在每个5秒时间片（time slice）之内获得的点击量，其中，散列的每个键都是某个时间片的开始时间，而键对应的值则存储了网站在该时间片之内获得的点击量。
如 hset weibo count:5:hits 12
为了能够清理计数器包含的旧数据，我们需要在使用计数器的同时，对被使用的计数器进行记录。为了做到这一点，我们需要一个有序序列（orderedsequence），这个序列不能包含任何重复元素，并且能够让我们一个接一个地遍历序列中包含的所有元素。
实现有序序列更好的办法是使用有序集合，有序集合的各个成员分别由计数器的精度以及计数器的名字组成，而所有成员的分值都为0。因为所有成员的分值都被设置成了0，所以Redis在尝试按分值对有序集合进行排序的时候，就会发现这一点，并改为使用成员名进行排序，这使得一组给定的成员总是具有固定的排列顺序，从而可以方便地对这些成员进行顺序性的扫描。
function updateCount(conn,name,count=1,now=null){
now=Nowtime()
pipe=conn.pipe()
pnow=当前时间片
hash=计数器的名字
pipe.zadd('known',hash,0)
pipe.hincrby('count'+hash,pnow,count)
pipe.execute()
}
清理旧数据，即清理旧集合，遍历有序集合的成员，获取他们的过期时间和当前时间比较，如果过期，使用zrem移除成员，然后在哈希中也移除相应的成员，确保有序集合的成员数保证在一定的范围之内。
这个函数以守护线程的方式运行，并且每隔一段时间内运行。
分析访问IP地址的信息
依据IP地址的相关信息，把IP存储在一个有序集合里面。
使用redis实现自动补全
在大多数情况下，我们使用有序集合是为了快速地判断某个元素是否存在于有序集合里面、查看某个成员在有序集合中的位置或索引，以及从有序集合的某个地方快速地按范围取出多个元素。
然而这一次，我们将把有序集合里面的所有分值都设置为0——这种做法使得我们可以使用有序集合的另一个特性：当所有成员的分值都相同时，有序集合将根据成员的名字来进行排序；而当所有成员的分值都是0的时候，成员将按照字符串的二进制顺序进行排序。
为了执行自动补全操作，程序会以小写字母的方式插入联系人的名字，并且为了方便起见，程序规定用户的名字只能包含英文字母，这样的话就不需要考虑如何处理数字或者符号了。
那么我们该如何实现这个自动补全功能呢？首先，如果我们将用户的名字看作是类似abc，abca，abcd，…，abd这样的有序字符串序列，那么查找带有abc前缀的单词实际上就是查找介于abbz...之后和abd之前的字符串。如果我们知道第一个排在abbz之前的元素的排名以及第一个排在abd之后的元素的排名，那么就可以用一个ZRANGE调用来取得所有介于abbz...和abd之间的元素，而问题就在于我们并不知道那两个元素的具体排名。为了解决这个问题，我们需要向有序集合分别插入两个元素，一个元素排在abbz...的后面，而另一个元素则排在abd的前面，接着根据这两个元素的排名来调用ZRANGE命令，最后移除被插入的两个元素。
分布式锁
先获取锁，然后执行操作，最后释放锁”的动作非常常见。
Redis使用WATCH命令来代替对数据进行加锁，因为WATCH只会在数据被其他客户端抢先修改了的情况下通知执行了这个命令的客户端，而不会阻止其他客户端对数据进行修改，所以这个命令被称为乐观锁。
分布式锁也有类似的“首先获取锁，然后执行操作，最后释放锁”动作，但这种锁既不是给同一个进程中的多个线程使用，也不是给同一台机器上的多个进程使用，而是由不同机器上的不同Redis客户端进行获取和释放的。
为了对Redis存储的数据进行排他性访问，客户端需要访问一个锁，这个锁必须定义在一个可以让所有客户端都看得见的范围之内，而这个范围就是Redis本身，因此我们需要把锁构建在Redis里面。
另一方面，虽然Redis提供的SETNX命令确实具有基本的加锁功能，但它的功能并不完整，并且也不具备分布式锁常见的一些高级特性。
SETNX命令天生就适合用来实现锁的获取功能，这个命令只会在键不存在的情况下为键设置值，而锁要
做的就是将一个随机生成的128位UUID设置为键的值，并使用这个值来防止锁被其他进程取得。
如果程序在尝试获取锁的时候失败，那么它将不断地进行重试，直到成功地取得锁或者超过给定的时限为止。
function acquire(con,lockname,acquire_time=60s){
identifier=uuid //随机生成
end=nowTime+acquire_time
while nowtime<end{ //给的的时间内获取锁
if conn.setnx("lock:"+lockname,identifer) //成功会返回1
return identifier
time.sleep(0.001) //每隔0.001秒尝试获取锁
return null
}
}
释放锁
dunction release(conn,lockname,identifier){
pipe=conn.pipe
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
}
throw exception 其他进程尝试修改锁，异常
return false //进程丢失了锁
}
}
目前的锁实现在持有者崩溃的时候不会自动被释放，这将导致锁一直处于已被获取的状态。为了解决这个问题，在这一节中，我们将为锁加上超时功能。
为了给锁加上超时限制特性，程序将在取得锁之后，调用EXPIRE命令来为锁设置过期时间，使得Redis可以自动删除超时的锁。
为了确保锁在客户端已经崩溃（客户端在执行介于SETNX和EXPIRE之间的时候崩溃是最糟糕的）的情况下仍然能够自动被释放，客户端会在尝试获取锁失败之后，检查锁的超时时间，并为未设置超时时间的锁设置超时时间。因此锁总会带有超时时间，并最终因为超时而自动被释放，使得其他客户端可以继续尝试获取已被释放的锁。
if conn.setnx("lock:"+lockname,identifer) //成功会返回1
conn.expire("lock:"+lockname,60s)
return identifier
elif not conn.ttl("loak:"+lockname)
conn.expire("lock:"+lockname,60s)
time.sleep(0.001) //每隔0.001秒尝试获取锁
计数信号量
对于同一个资源，允许n个进程同时访问，对于超过n的进程拒绝访问。
常见的有对于同一个账户只允许登录5个设备。
构建计数信号量时要考虑的事情和构建其他类型的锁时要考虑的事情大部分都是相同的，比如判断是哪个客户端取得了锁，如何处理客户端在获得锁之后崩溃的情况，以及如何处理锁超时的问题
使用Redis来实现超时限制特性通常有两种方法可选。一种是像之前构建分布式锁那样，使用EXPIRE命令，而另一种则是使用有序集合。为了将多个信号量持有者的信息都存储到同一个结构里面。
程序将为每个尝试获取信号量的进程生成一个唯一标识符，并将这个标识符用作有序集合的成员，而成员对应的分值则是进程尝试获取信号量时的Unix时间戳。
进程在尝试获取信号量时会生成一个标识符，并使用当前时间戳作为分值，将标识符添加到有序集合里面。
接着进程会检查自己的标识符在有序集合中的排名。
如果排名低于可获取的信号量总数（成员的排名从0开始计算），那么表示进程成功地取得了信号量。
反之，则表示进程未能取得信号量，它必须从有序集合里面移除自己的标识符。
为了处理过期的信号量，程序在将标识符添加到有序集合之前，会先清理有序集合中所有时间戳大于超时数值（timeout number value）的标识符。
function acquire_sem(conn,semname,limit,timeout=10s){
identifier=uuid
now=nowtime
pipe=conn.pipeline()
pipe.zremrangebyscore(semname,min= max=now-timeout) 移除过期的键 max之前的键都已经过期了
pipe.zadd(semname,identifier,now) 尝试获取
pipe.zrank(semname,identifier)
if pipe.execute()[-1]<limt
return identifier
conn.zrem(semname,identifier) //信号量获取失败，移除键
return none
}
释放键非常简单，直接使用conn.zrem(semname,identifier)
但这个信号量实现也存在一些问题：它在获取信号量的时候，会假设每个进程访问到的系统时间都是相同的，而这一假设在多主机环境下可能并不成立。举个例子，对于系统A和B来说，如果A的系统时间要比B的系统时间快10毫秒，那么当A取得了最后一个信号量的时候，B只需要在10毫秒内尝试获取信号量，就可以在A不知情的情况下，“偷走”A已经取得的信号量。对于一部分应用程序来说这并不是一个大问题，但对于另外一部分应用程序来说却并不是如此。
每当锁或者信号量因为系统时钟的细微不同而导致锁的获取结果出现剧烈变化时，这个锁或者信号量就是不公平的（unfair）。不公平的锁和信号量可能会导致客户端永远也无法取得它原本应该得到的锁或信号量。
公平信号量
为了尽可能地减少系统时间不一致带来的问题，我们需要给信号量实现添加一个计数器以及一个有序集合。
其中，计数器通过持续地执行自增操作，创建出一种类似于计时器（timer）的机制，确保最先对计数器执行自增操作的客户端能够获得信号量。
另外，为了满足“最先对计数器执行自增操作的客户端能够获得信号量”这一要求，程序会将计数器生成的值用作分值，存储到一个“信号量拥有者”有序集合里面，然后通过检查客户端生成的标识符在有序集合中的排名来判断客户端是否取得了信号量。
计数器是一个键是字符串类型，值是整形
function auquire_fair_semaphore(conn,semname,limit,timeout){
indentifier=uuid
czset=semname+":owner"
cstr=semname+":counter" //计数器
now=nowtime
pipe=conn.pipeline
pipe.zremrangerbyscore(semname,min,max=now-timeout)
pipe.zinterstore(czset,{czset:1,semname:0})
pipe.incr(zstr) 计数器增加
counter=pipe.execute
pipe.zadd(semname,identifier,now)
pipe.zadd(czset,identifier,counter)
pipe.zrank(czset,indentifer) //获取计数集合中的排名
if (pipe.exe[-1]<limit)
return identifier
pipe.zrem(semname,identifer)
pipe.zrem(czset,indentifer)
pipe.execute
return null
}
刷新信号量
前面介绍的信号量实现默认只能设置10秒的超时时间，它主要用于实现超时限制特性并掩盖自身包含的潜在缺陷，但是短短的10秒连接时间对于流API的使用者来说是远远不够的，因此我们需要想办法对信号量进行刷新，防止其过期。
因为公平信号量区分开了超时有序集合和信号量拥有者有序集合，所以程序只需要对超时有序集合进行更新，就可以立即刷新信号量的超时时间了。
function refresh(conn,semname,identifier){
if conn.zadd(semname,identifier,nowtime){ //如果有序集合中键还在，返回0，否则返回1
conn.zrem(semname,identifier) //返回1，代表进程的信号量被释放了，也能拥有信号量，释放
return false
}
return true
}
消除竞争条件
将系统时钟用作获取锁的手段提高了这类竞争条件出现的可能性，导致信号量持有者的数量比预期的还要多，多出的信号量数量与各个系统时钟之间的差异有关——差异越大，出现额外信号量持有者的可能性也就越大。虽然引入计数器和信号量拥有者有序集合可以移除系统时钟这一不确定因素，并降低竞争条件出现的几率，但由于执行信号量获取操作需要客户端和服务器进行多次通信，所以竞争条件还是有可能会发生。
为了消除信号量实现中所有可能出现的竞争条件，构建一个正确的计数器信号量实现，我们需要用到前面在构建的带有超时功能的分布式锁。
总的来说，当程序想要获取信号量的时候，它会先尝试获取一个带有短暂超时时间的锁。如果程序成功取得了锁，那么它就会接着执行正常的信号量获取操作。如果程序未能取得锁，那么信号量获取操作也宣告失败
任务队列
比如先进先出（FIFO）队列、后进先出（LIFO）队列和优先级（priority）队列。
任务队列来记录邮件的收信人以及发送邮件的原因，并构建一个可以在邮件发送服务器运行变得缓慢的时候，以并行方式一次发送多封邮件的工作进程。
邮件队列将使用RPUSH命令来将待发送的邮件推入列表的右端，并且因为工作进程除了发送邮件之外不需要执行其他工作，所以它将使用阻塞版本的弹出命令BLPOP从队列中弹出待发送的邮件，而命令的最大阻塞时限为30秒。
在一些情况下，为每种任务单独使用一个队列的做法并不少见，但是在另外一些情况下，如果一个队列能够处理多种不同类型的任务，那么事情就会方便很多。
工作进程会监视用户提供的多个队列，并从多个已知的已注册回调函数里面，选出一个函数来处理JSON编码的函数调用。
队列的优先级
BLPOP命令和BRPOP命令都允许用户给定多个列表作为弹出操作的执行对象：其中BLPOP命令将弹出第一个非空列表的第一个元素，而BRPOP命令则会弹出第一个非空列表的最后一个元素。
例如
blpop 优先级高队列，优先级中队列，优先级地队列，timeout
使用blpop,brpop命令的意图是谁最先能取出来，谁就先执行。
延迟队列
有几种不同的方法可以为队列中的任务添加延迟性质，以下是其中3种最直截了当的方法。
- 在任务信息中包含任务的执行时间，如果工作进程发现任务的执行时间尚未来临，那么它将在短暂等待之后，把任务重新推入队列里面。
- 工作进程使用一个本地的等待列表来记录所有需要在未来执行的任务，并在每次进行while循环的时候，检查等待列表并执行那些已经到期的任务。
- 把所有需要在未来执行的任务都添加到有序集合里面，并将任务的执行时间设置为分值，另外再使用一个进程来查找有序集合里面是否存在可以立即被执行的任务，如果有的话，就从有序集合里面移除那个任务，并将它添加到适当的任务队列里面。
第三种方法是最简单和直接的。
消息拉取
利用redis的订阅模式可以接受最新的消息，但是如何获取历史消息，可以为每个消息创建一个ID，或者消息创建的时间戳，创建一个有序集合，键为消息ID，时间戳为分值，根据时间戳，用户可以回顾历史消息。