Redis学习之旅~基础应用篇
2008 年 9 月 7 日
这一年以来,写了太多的业务代码。是时候要总结一下自己的积累了。本文是redis深度历险的读书笔记，做个记录以及分享给大家。
docker
redis
数据结构
字符串
- 字符串是一个字符数组
- 常见用途就是信息JSON序列化成为字符串之后，存入redis，取信息会经过一次反序列化
- 字符串是动态字符串，可以进行扩容；当字符串长度小于1m的时候，扩容是加倍。当大于1m，每次扩容就增加1m。字符串最大的长度为512m
- 使用字符串计数的时候，单个值的最大值为signed long，超过就会报错
-
常用命令
set name a get name exists a del a mset a 1 b 2 mget a b expire a 5 setex a 5 1 setnx a 1 incr a incrby a 10
列表
-
相当于java中的LinkedList，一个双向链表。插入和删除操作快，但是索引定位很慢。
节点删除操作 p.pre.next = p.next p.next.pre = p.pre p = null
-
列表可以作为队列，又进左出即可
rpush books java python golang kotlin 4 llen books 4 lpop books java
-
列表可以作为栈，右进右出即可。就是，调整了队列使用的方向
rpush books java python golang kotlin 4 llen books 4 rpop books kotlin
-
使用队列会存在的慢操作有get(index)方法，需要遍历整个链表，性能为O(n)
rpush books java python golang kotlin 4 lindex 1 # 时间为O(n) lindex books 1 "python" lrange books 0 -1 "java" "python" "golang" "kotlin" ltrim books 1 -1 # 设置长度为3的数组，区间外的全部被清除 OK lrange books 0 -1 # 获取所有元素 "python" "golang" "kotlin" llen books 3 ltrim books 1 0 # 删除所有元素 OK
- 快速列表：在列表元素较少的时候，会使用一块连续的内存，叫ziplist（压缩列表）。当数据量较多的时候，就要变成quicklist(快速列表),多个ziplist使用双向指针串起来使用。
// 快速列表
struct quicklist {
quicklistNode* head;
quicklistNode* tail;
long count; // 元素总数
int nodes; // ziplist 节点的个数
int compressDepth; // LZF 算法压缩深度
...
}
// 快速列表节点
struct quicklistNode {
quicklistNode* prev;
quicklistNode* next;
ziplist* zl; // 指向压缩列表
int32 size; // ziplist 的字节总数
int16 count; // ziplist 中的元素数量
int2 encoding; // 存储形式 2bit，原生字节数组还是 LZF 压缩存储
}
struct ziplist_compressed {
int32 size;
byte[] compressed_data;
}
struct ziplist {
...
}
Hash字典
- 内部实现结构就是数组+链表的方式进行实现。相同hash值的元素使用链表串联起来。java的链表数量超过6个会变成红黑树，不知道redis的会不会有这个查询优化。
- redis里面的值就只能是字符串，便于比较。不像java的hashmap可以存储对象，equals方法的编写可谓五花八门
-
java的hashmap在进行rehash的时候，需要一次性进行rehash，是个耗时操作，多线程还会出现环。redis的是渐进式rehash。rehash的时候，会保留两个hash结构，查询时候会查询两个hash结构，在后续的定时任务和hash指令操作的时候，渐进地将旧的内容迁移到新的hash结构中。当内容迁移完毕，删除旧的hash结构
hset books java "think in java" 1 hset books golang "concurrency in go" 1 hset books python "python cookbook" 1 hgetall books # 返回所有entryies，key和value间隔返回 "java" "think in java" "golang" "concurrency in go" "python" "python cookbook" hlen books 3 hget books java "think in java" hset books java "my java book" 0 hmset books java "" golang "" ok hset user-lixinkun age 25 1 hincrby user-lixinkun 1 26
set
-
相当于java的hashset，确保唯一性
sadd books java 1 sadd books java 0 sadd books python golang smembers books java golang python #顺序可能被打乱
Zset 有序列表
- 类似于java的sortedset和hashmap的结合体。具备set的唯一性原理，也可以给每个值赋一个score，代表它的排序权重。它的内部实现是一个跳跃列表。貌似java也有这个数据结构
*内部排序通过skiplist实现。zset要支持睡觉的插入和删除，不能使用数组来表示。普通的链表数据结构，有新元素插入的时候，需要进行定位，这样才能保证是有序的。通常会通过二分查找来确定位置，但是只有数组才支持快速查找。这时候就需要一个多层级的列表结构，最下面一层的元素都是串联起来的。某个元素存在于所有的层。定位插入点的时候，先在顶层进行定位，然后下潜到下一层，然后一直到最底下一层。redia采用随机策略来决定元素可以兼职到第几层。L0层的概率是100%，L1的是50%。这样去类推
-
数据结构,具体介绍链接 redis跳跃列表介绍
/* ZSETs use a specialized version of Skiplists */ typedef struct zskiplistNode { //成员对象 robj *obj; //分值 double score; //后退指针 struct zskiplistNode *backward; //层 struct zskiplistLevel { struct zskiplistNode *forward;//前进指针 unsigned int span;//跨度 } level[]; } zskiplistNode; typedef struct zskiplist { //表头节点和表尾节点 struct zskiplistNode *header, *tail; //表中节点的的数量 unsigned long length; //表中层数最大的节点层数 int level; } zskiplist;
容器的通用规则
- create if not exists.如果容器不存在，则先创建容器。
- drop if no elements.如果容器没有元素了，就立即删除该容器，释放内存。
- 容器的过期时间是针对整个对象的，hash的过期是整个进行过期，而不是单个元素
分布式锁
- 对于不是原子性的操作，就需要通过分布式锁来保证只有一个线程可以获取到锁。具体使用的是setnx和del命令。锁处理完了以后，就要被释放。中间需要解决的问题就是，如果业务逻辑执行之间，出现了异常，可能导致del无法执行，整个业务瘫痪。
- 或许，可以加一个过期时间，5s以后自动释放。但是逻辑还有问题的，如果expire的时候，命令丢失了，也会死锁。出现这种问题的原因就是setnx和expire两个命令不是原子性的。
- 在redis2.8的版本中，作者加入了set指令的扩展餐厨，setnx和expire指令可以一起执行了，两个指令是原子性的。
set lock:book true ex 5 nx del lock:book
- 超时问题是redis分布式锁不能解决的问题，如果加锁和释放锁之间的逻辑太长，超出了锁的超时时限，那么第二个线程可能获得redis分布式锁。所以，redis分布式锁不要用于较长时间的任务
- 可重入性。java里面有一个ReentrantLock。Redis分布式锁如果要支持重入，需要对客户端的set方法进行封装，使用线程的ThreadLocal遍历存储当前持有锁的计数。
-
这个是redis分布式锁的使用和错误案例介绍 redis分布式锁使用与错误示例
//错误使用就是使用setnx和expire两个指令
/**
* 尝试获取分布式锁
* @param jedis Redis客户端
* @param lockKey 锁
* @param requestId 请求标识
* @param expireTime 超期时间
* @return 是否获取成功
*/
public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {
String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
if (LOCK_SUCCESS.equals(result)) {
return true;
}
return false;
}
- java版本的可重入锁,核心就是用threadlocal去记录每个线程，某个key的锁计数
public class RedisWithReentrantLock{
//线程锁计数器8
private ThreadLocal<Map> lockers = new ThreadLocal();
private Jedis jedis;
//内部使用的lock
private boolean _lock(String key) {
return jedis.set(key, "", "nx", "ex", 5L) != null;
}
//内部使用的unlock
private void _unlock(String key) {
jedis.del(key);
}
//获取计数器
private Map currentLockers() {
Map refs = lockers.get();
if (refs != null) {
return refs;
}
lockers.set(new HashMap());
return lockers.get();
}
//可重入加锁
public boolean lock(String key) {
Map refs = currentLockers();
Integer refCnt = refs.get(key);
if (refCnt != null) {
refs.put(key, refCnt + 1);
return true;
}
boolean ok = _lock(key);
if (!ok) {
return false;
}
refs.put(key, 1);
return true;
}
//可重入释放锁
public boolean unlock(String key) {
Map refs = currentLockers();
Integer refCnt = refs.get(key);
if (refCnt == null) {
return false;
}
refCnt = -1;
if (refCnt > 0) {
refs.put(key, refCnt);
} else {
refs.remove(key);
_unlock(key);
}
return true;
}
}
redis的消息队列，只做特性介绍，实际使用我们目前用rocketmq
- 使用简单，只有一个消费者
- 没有ack保证，如果对消息可靠性有极高的要求，推荐使用rocketmq的事务消息
- 使用list作为异步消息的发送结构，rpush，rpop
延时队列的实现
- 延时队列可以使用zset实现，将消息序列化成为一个字符串，消息的到期时间为score。然后使用多个线程去轮询这个队列。使用多线程是为了保障可用性。一个线程挂了，还有其他线程可以进行处理
位图，节约存储空间
- 平时进行开发的时候，要对一些bool类型的数据进行存取，如用户一年的签到记录。如果使用key/value方式进行存取，就要记录365个，当用户量上亿的时候，需要的存储结构是非常惊人的。
- redis提供了位图数据结构，每个用户每天的签到就只占据一个位。365天大概需要46个字节，大大节约空间
- 位图就是byte数组，可以使用getbit和setbit进行存储,如果字节是不可以打印字符，redis会展示其16进制。实际来操作一波
零存整取的方式，一个个bit去设置，然后一次性拿出值 通过python获取字母`h`的hash值为0b1101000 setbit s 1 1 setbit s 2 1 setbit s 4 1 get s "h" # 这样就得到了h的值 零存零取的方式，就是对单个bit进行操作,没有设置的位就是默认为0 set bit w 1 1 get bit w 0 0 整存零取 set w h getbit w 1 1 不可打印字符 setbit x 0 1 setbit x 1 1 get x "\xc0"
-
位图的统计和查找：redis提供了对位图的基本统计算
set w hello ok bitcount w 21 bitcount w 0 0 3 bitcount w 0 1 # 统计1的数量 7 bitpos w 0 # 第一个0 0 bitpos w 1 1 1 # 第二个字符算起，第一个1位 魔术指令bitfield bitfield w get u4 0
高级数据结构
HyberLogLog,提供不精确的统计数据
- 页面的uv，不能使用一个key的incr，因为要进行去重
- 使用zset也可以做到这个，但是消耗空间太大
- HyperLogLog的标准误差是0.81%。对于不精确的统计，可以使用这个数据结构
- 提供pdadd，pfcount两个指令,pf是Philippe Flajolet教授的名字缩写，这个数据结构是他发明的。这个数据结构占用的内存是12k，使用了2^14个桶去进行计算
pfadd book java 1 pfadd book java 0 pfadd book python 1 pfadd book python pfcount book 2
布隆过滤器Bloom Filter
- 这个数据结构主要实现去重的功能
- 场景使用：广告推送的时候，如果我们看过了一条广告，下一条推送的时候，要对看过的广告进行去重。如果使用关系型数据库，每次推送都要去数据库查询，这样数据库很难抗住这个压力。况且，浏览记录会随着时间线性增长，空间也跟不上
- 布隆过滤器会有一定的误判概率。这可以比喻成为一个不精确的set结构，调用contains的时候，有可能误判。如果说这个值不存在，那就肯定不存在。如果说这个值存在，优肯是误判，可能，两者长得相像？这样的话，对于实际业务也没有影响，我们可以保证推送给用户的都是最新的内容，只有极少数用户没看过的内容会被误判。经过测试，误判率大概在1%左右。调整参数可以优化这个误判率
- 布隆过滤器作为redis的一个插件，需要版本为4.0以后，才能使用。
-
基本指令为bf.add,bf.exists.bf.madd(批量)，安装docker，然后去重新安装redis，带这个插件的版本。 centos7安装docker方法
，安装好以后再安装插件，使用docker启动redis - 用户可以通过bf.reserve命令显式制定错误率，key和初始空间。错误率越低，占用空间越大。初始值越大，空间越大，所以，在使用的时候，需要准确预估一下元素的大小。
- 布隆过滤器是一个大型的位数组。向布隆过滤器添加key的时候，会使用多个hash函数对key进行hash，算得一个整数索引。然后对位数组的长度取模运算得到一个位置，每个hash函数都得到一个不同的位置，再把这几个位置都设置为1，完成了add操作。巧妙地共享可多个key的存储空间。
- 为何会出现误判呢？就是由于判断key是否存在的时候，如果某个位是0，那就一定不存在。但是，如果各个位置都是1的话，也不能说明是存在的，有可能是其他key存在导致。如果位数组比较稀疏，出现误判的概率就低了。所以，使用的时候，不要让实际的元素大于初始值。如果大于的时候，要对布隆过滤器进行重建。
- set存储的是所有的元素，布隆过滤器是存储元素的指纹（hash）。空间节省还是比较明显的.
- docker加载插件的命令，docker run -p 6379:6379 –name redis-redisbloom redislabs/rebloom:latest
- docker exec -it redis-redisbloom bash
bf.add book java 1 bf.add book java #再次加入java，就被排出了 0 bf.exists book python 0
redis限流
- 后端屏蔽用户的过于频繁的请求，比如1秒内不能请求超过5次。这个限流存在一个滑动时间窗口，zset的score圈出一个时间窗口，我们保留这个时间窗口，对于窗口外面的数据，则可以统统砍掉。zset的value可以为毫秒时间戳。
public class SimpleRateLimiter{
private Jedis jedis;
//用户请求是否允许
//使用乐观锁
//value保证唯一性就好
public boolean isActionAllow(String userId, String actionKey, int period, int maxCount) {
String key = String.format("hist:%s:%s", userId, actionKey);
long now = System.currentTimeMillis();
Pipeline pipe = jedis.pipelined();
pipe.multi();
pipe.zadd(key, now, "", now);
//截取
pipe.zremrangeByScore(key, 0, now - period * 1000);
Response count = pipe.zcard(key);
pipe.expire(key, period + 1);
pipe.exec();
pipe.close();
return count.get() <= maxCount;
}
}
- 漏斗限流，灵感来源于漏斗，就是漏斗有一个容量，也有一个消耗速率。当灌水速率小于漏水速率，漏斗超过容量以后。灌水行为就要丢弃或者暂停
//单机版的漏斗限流
public class FunnelRateLimiter{
class Funnel{
//容量
int capacity;
//漏水速率
float leakingRate;
//剩余空间
int leftQuota;
//上次触发时间
long leakingTs;
}
//....省略构造方法
void makeSpace() {
long now = System.currentTimeMillis();
long deltaTs = nowTs - leakingTs;
int deltaQuota = (int)(deltaTs * leakingRate);
//如果时间间隔太久，可能会导致int溢出
if (deltaQuota < 0) {
//重设
this.leftQuota = capacity;
this.leakingTs = 0;
return;
}
if (deltaQuota this.capacity) {
this.leftQuota = this.capacity;
}
//是否可以接收请求
boolean watering(int quota) {
makeSpace();
if (leftQuota >= quota) {
this.leftQuota -= quota;
return true;
}
return false;
}
}
private Map funnels = new HashMap();
public boolean isActionAllow(String userId, String actionKey, int capacity, fload leakingRate) {
String key = String.format("funnels:%s:%s", userId, actionKey);
Funnel funnel = funnels.get(key);
if (funnel == null) {
funnel = new Funnel(capacity, leakingRate);
funnels.put(key, funnel);
}
return funnel.watering(1);
}
}
- 这里是redis提供的分布式限流，redis4.0提供了redis-cell。这是一个差价，需要去github下载redis-cell的源码解压以后。进入redis-cli后，执行redis的 module load命令即可加载
wget https://github.com/brandur/redis-cell/releases/download/v0.2.4/redis-cell-v0.2.4-x86_64-unknown-linux-gnu.tar.gz tar -xzvf redis-cell-v0.2.4-x86_64-unknown-linux-gnu.tar.gz 解压后得到libredis_cess.so redis-cli module load /data/libredis_cell.so 我这里是放到了data目录下面，运行redis后加载就行了 //各个参数 key 容量 计算漏斗速率 可选参数 cl.throttle [] cl.throttle book 15 30 60 1 # 这个表示容量为15,2秒一个漏水速率（30/60） 0 # 0 允许 -1 拒绝 16 # 漏斗容量 15 # 剩余空间 -1 # 当第一个参数为-1师，这里的值表示需要经过多少时间以后才能再试 2 # 多久时间以后，漏斗可以完全空出来
About The Author
