redis基本应用

// lock: 后是一个普通字符串，锁的标识
> setnx lock:lock-something true
OK

// 业务逻辑
... do something critical ...

// 释放锁
> del lock:lock-something
(integer) 1

> set lock:lock-something true ex 5 nx
OK
// 业务逻辑
... do something critical ...
> del lock:lock-something

public class RedisWithReentrantLock {

  private ThreadLocal<Map<String, Integer>> lockers = new ThreadLocal<>();

  private Jedis jedis;

  public RedisWithReentrantLock(Jedis jedis) {
    this.jedis = jedis;
  }

  private boolean _lock(String key) {
    return jedis.set(key, "", "nx", "ex", 5L) != null;
  }

  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 = this._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);
      this._unlock(key);
    }
    return true;
  }

  public static void main(String[] args) {
    Jedis jedis = new Jedis();
    RedisWithReentrantLock redis = new RedisWithReentrantLock(jedis);
    System.out.println(redis.lock("codehole"));
    System.out.println(redis.lock("codehole"));
    System.out.println(redis.unlock("codehole"));
    System.out.println(redis.unlock("codehole"));
  }

}

public class RedisDelayingQueue<T> {

  static class TaskItem<T> {
    public String id;
    public T msg;
  }

  // fastjson 序列化对象中存在 generic 类型时，需要使用 TypeReference
  private Type TaskType = new TypeReference<TaskItem<T>>() {
  }.getType();

  private Jedis jedis;
  private String queueKey;

  public RedisDelayingQueue(Jedis jedis, String queueKey) {
    this.jedis = jedis;
    this.queueKey = queueKey;
  }

  public void delay(T msg) {
    TaskItem<T> task = new TaskItem<T>();
    task.id = UUID.randomUUID().toString(); // 分配唯一的 uuid
    task.msg = msg;
    String s = JSON.toJSONString(task); // fastjson 序列化
    jedis.zadd(queueKey, System.currentTimeMillis() + 5000, s); // 塞入延时队列 ,5s 后再试
  }

  public void loop() {
    while (!Thread.interrupted()) {
      // 只取一条
      Set<String> values = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis(), 0, 1);
      if (values.isEmpty()) {
        try {
          Thread.sleep(500); // 歇会继续
        } catch (InterruptedException e) {
          break;
        }
        continue;
      }
      String s = values.iterator().next();
      if (jedis.zrem(queueKey, s) > 0) { // 抢到了
        TaskItem<T> task = JSON.parseObject(s, TaskType); // fastjson 反序列化
        this.handleMsg(task.msg);
      }
    }
  }

  public void handleMsg(T msg) {
    System.out.println(msg);
  }

  public static void main(String[] args) {
    Jedis jedis = new Jedis();
    RedisDelayingQueue<String> queue = new RedisDelayingQueue<>(jedis, "q-demo");
    Thread producer = new Thread() {

      public void run() {
        for (int i = 0; i < 10; i++) {
          queue.delay("codehole" + i);
        }
      }

    };
    Thread consumer = new Thread() {

      public void run() {
        queue.loop();
      }

    };
    producer.start();
    consumer.start();
    try {
      producer.join();
      Thread.sleep(6000);
      consumer.interrupt();
      consumer.join();
    } catch (InterruptedException e) {
    }
  }
}

3. bitop 对多个键进行位操作。 op值得是operation。

用法：`bitop operation destkey key1 key2 [key ...]`

- destkey: 运算结果保存当值
- operation: and,or,not,xor(异或)

4. bitcount 计算给定字符串上，位为1的个数。 返回位为1的数量

用法： `bitcount key [start] [end]`

5. bittop 获取某个键第一位被设置为 0 或 1 位的位置。

6. 魔术指令,bitfield 。一次对多个位范围进行操作。bitfield 有三个子指令，分别是 get/set/incrby。每个指令都可以对指定片段做操作

用法：`bitfield key [GET type offset] [SET type offset value] [INCRBY type offset increment] [OVERFLOW WRAP|SAT|FAIL]`

返回值：返回一个数组作为回复， 数组中的每个元素就是对应操作的执行结果


#### bitmap应用场景示例

1. 统计用户上线次数


每当用户在某一天上线的时候，我们就使用 `setbit`，以用户名作为 key ，将那天所代表的网站的上线日作为 offset 参数，并将这个 offset 上的位设置为 1

然后通过`bitcount`可得到用户总共上线次数

2. 统计用户签到次数

原理同上线次数

3. 统计活跃用户

统计某天或者某连续几天，活跃用户数。

方案：若某用户上线，则以日期为KEY，以用户user_id为偏移量（若ID不为整数，则将ID hash化为唯一ID），设置位为 1。 然后bitcount 日期即可得到某天活跃用户数。

某连续两天活跃用户数示例:

### HyperLogLog统计大量数据

如果要统计网页当PV，这个需求很好实现，给每个网页增加一个redis计数器，每个请求就incrby一下就好了


如果要统计UV，同一个用户一天之内的多次访问只算做一次，所以要做去重。可以用一个set存储当天访问过某个网页的用户id，当一个请求过来时，我们使用 sadd 将用户 ID 塞进去就可以了。通过 scard 可以取出这个集合的大小，这个数字就是这个页面的 UV 数据。


但是当页面的访问量非常大，比如几千万的UV的时候，这个set集合就非常大，会占据很多内存空间。其实我们对UV只是需要一个大概的数字， 100W和101W差别其实不大，redis 提供了 HyperLogLog 数据结构就是用来解决这种统计问题的。HyperLogLog 提供不精确的去重计数方案，虽然不精确但是也不是非常不精确，标准误差是 0.81%，这样的精确度已经可以满足上面的 UV 统计需求了

#### 使用方法

HyperLogLog提供了两个指令：`pfadd/pfcount`,增加计数和获取计数。比如上述统计UV的功能，只需要将用户id,pfadd进去，最后使用pfcount来获取结果就好了。这个结果是带有去重功能的

HyperLogLog还提供了一个`pfmerge`的指令，用于将多个`pfcount`的结果合并。

值得注意的是一个HyperLogLog数据结构就占据了12K的空间，所以不太适用于统计范围较小的数据，对于大量数据，HyperLogLog就可以省去很多内存空间

### 布隆过滤器

布隆过滤器可以理解为一个不怎么精确的 set 结构，当你使用它的 contains 方法判断某个对象是否存在时，它可能会误判。但是布隆过滤器也不是特别不精确，只要参数设置的合理，它的精确度可以控制的相对足够精确，只会有小小的误判概率。当布隆过滤器说某个值存在时，这个值可能不存在；当它说不存在时，那就肯定不存在。

它的使用场景可以归纳为：判断大量数据是否在一个池子里

有如下特点：
- 空间效率和查询效率高
- 不会漏判，但是有一定的误判率（哈希表是精确匹配）

#### 布隆过滤器原理

想快速找到一个元素，以hash的方式是最快的，常用的有hashmap等结构，但是对于大量数据，使用hashmap显然会占据很多内存。

布隆过滤器优于hashmap的地方就是占用空间很小。实现原理如下：

![](/images/redis/bloomfilter.png)

1. 布隆过滤器是一个bit向量，或者说是bit数组
2. 针对一个值，使用多个hash函数，每个hash值指向的地方设置为1

所以当一个数据，通过不同的hash函数映射到的位都有值时，他可能存在。但是任何一位没有值，就代表这个数据必然不存在


#### redis中布隆过滤器的基本使用

redis可以安装布隆过滤器插件来使用布隆过滤器。

布隆过滤器有两个基本指令：`bf.add` 添加元素，`bf.exists` 查询元素是否存在

### 限流

#### 简单限流

使用 `zset` 这个数据结构的 `score` 来充当滑动窗口，保证value唯一即可，Java实现如下