Redis学习之旅~基础应用篇

栏目: 数据库 · 发布时间: 6年前

内容简介：这一年以来,写了太多的业务代码。是时候要总结一下自己的积累了。本文是redis深度历险的读书笔记，做个记录以及分享给大家。*内部排序通过skiplist实现。zset要支持睡觉的插入和删除，不能使用数组来表示。普通的链表数据结构，有新元素插入的时候，需要进行定位，这样才能保证是有序的。通常会通过二分查找来确定位置，但是只有数组才支持快速查找。这时候就需要一个多层级的列表结构，最下面一层的元素都是串联起来的。某个元素存在于所有的层。定位插入点的时候，先在顶层进行定位，然后下潜到下一层，然后一直到最底下一层。r

这一年以来,写了太多的业务代码。是时候要总结一下自己的积累了。本文是 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<String,Integer>> 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<String, Integer> currentLockers() {
    Map<String, Integer> refs = lockers.get();
    if (refs != null) {
      return refs;
    }
    lockers.set(new HashMap<>());
    return lockers.get();
  }
  //可重入加锁
  public boolean lock(String key) {
    Map<String, Integer> 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<String, Integer> 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<Long> 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 < 1) {
      return;
    }

    this.leftQuota += leftQuota;
    this.leakingTs = nowTs;
    if (this.leftQuota > this.capacity) {
      this.leftQuota = this.capacity;
    }

    //是否可以接收请求

    boolean watering(int quota) {
      makeSpace();
      if (leftQuota >= quota) {
        this.leftQuota -= quota;
        return true;
      }
      return false;
    }
  }

  private Map<String, Funnel> 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 <key> <max_burst> <count per period> <period> [<quantity>]

cl.throttle book 15 30 60 1  # 这个表示容量为15,2秒一个漏水速率（30/60）

0          # 0 允许 -1 拒绝

16        # 漏斗容量

15        # 剩余空间

-1        # 当第一个参数为-1师，这里的值表示需要经过多少时间以后才能再试

2          # 多久时间以后，漏斗可以完全空出来

以上所述就是小编给大家介绍的《Redis学习之旅~基础应用篇》，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对码农网的支持！