Redis的常用场景

[TOC]

★ Redis分布式锁

示例代码, 其实该分布式锁的实现是存在很多问题.此处仅为帮助理解分布式锁的思想

import com.alibaba.fastjson.JSONObject;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.params.SetParams;

import java.util.HashMap;
import java.util.Map;

/**
 * 1. 分布式锁
 *
 * set, setnx(set if not exists), expire, del, get
 *
 * <p>实现原理</p>
 * 1. 加锁：执行setnx，若成功再执行expire添加过期时间
 * 2. 解锁：执行delete命令
 *
 * @author chenh
 * @date 2020-05
 */
public class RedisWithReentrantLock {

    private static final int EXPIRE_TIME = 5;

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

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

    /**
     * 加锁
     * 
     * Redis2.6.12以上版本为set命令增加了可选参数
     * jedis.setnx(key, "");
     * jedis.expire(key, 5);
     */
    private boolean _lock(String key) {
        String setnx = jedis.set(key, "", SetParams.setParams().nx().ex(EXPIRE_TIME));
        return setnx != null;
    }

    /**
     * 释放锁
     *
     * @param key
     */
    private void _unlock(String key) {
        jedis.del(key);
    }

    private Map<String, Integer> currentLockers() {
        Map<String, Integer> refs = lockers.get();
        System.out.println("get=" + JSONObject.toJSONString(refs));
        if (refs != null) {
            return refs;
        }
        lockers.set(new HashMap<>());
        System.out.println("set");
        return lockers.get();
    }

    public boolean lock(String key) {
        Map<String, Integer> refs = this.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 = this.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;
    }
}

对比 setnx,expire 与set (set命令增加可选参数)

jedis.setnx(key, "");
jedis.expire(key, 5);

该方案有一个致命问题，由于setnx和expire是两条Redis命令，不具备原子性，如果一个线程在执行完setnx()之后突然崩溃，导致锁没有设置过期时间，那么将会发生死锁。

优点
1. 实现简单，相比数据库和分布式系统的实现，该方案最轻，性能最好

缺点
1. setnx和expire分2步执行，非原子操作；若setnx执行成功，但expire执行失败，就可能出现死锁
2. delete命令存在误删除非当前线程持有的锁的可能
3. 不支持阻塞等待、不可重入

jedis.set(key, "", SetParams.setParams().nx().ex(EXPIRE_TIME));

我们对锁设置了过期时间,即使锁的持有者后续发生崩溃而没有解锁,锁也会因为到了过期时间而自动解锁(即key被删除),不会发生死锁。

因为判断和删除不是一个原子性操作。在并发的时候很可能发生解除了别的客户端加的锁。具体场景有：客户端A加锁，一段时间之后客户端A进行解锁操作时，在执行jedis.del()之前，锁突然过期了，此时客户端B尝试加锁成功，然后客户端A再执行del方法，则客户端A将客户端B的锁给解除了。从而不也不满足加锁和解锁必须是同一个客户端特性。解决思路就是需要保证GET和DEL操作在一个事务中进行，保证其原子性。

★ Redisson分布式锁的实现(扩展1)

import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

import java.util.concurrent.TimeUnit;

/**
 * Redisson 分布式重入锁用法
 * Redisson 支持单点模式、主从模式、哨兵模式、集群模式，这里以单点模式为例
 * Redisson 这个框架重度依赖了Lua脚本和Netty
 *
 * 官方文档说明
 * @link https://github.com/redisson/redisson/wiki/8.-Distributed-locks-and-synchronizers
 *
 * @author huichen
 * @date 2020-06
 */
public class RedissonWithLock {

    /**
     * RedissonClient lock&unlock的使用
     */
    public void execute() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379").setDatabase(0);
        RedissonClient redissonClient = Redisson.create(config);
        // 获取锁对象实例（无法保证是按线程的顺序获取到）
        RLock rLock = redissonClient.getLock("codehole");

        // 尝试获取锁的最大等待时间, 超过这个时间即获取锁失败
        long waitTimeout = 1L;
        // 持有锁的时间, 相当于 redis.expire()
        long leaseTime = 10L;
        try {
            // 尝试获取
            boolean bool = rLock.tryLock(waitTimeout, leaseTime, TimeUnit.SECONDS);
            if (bool) {
                // do 业务代码...
            }
        } catch (InterruptedException e) {
            throw new RuntimeException("aquire lock fail");
        } finally {
            //无论如何, 最后都要解锁
            rLock.unlock();
        }
    }
}

/**
 * 单元测试
 */
public class RedissonWithLockTest {

    /**
     * 加锁和解锁过程中还巧妙地利用了redis的发布订阅功能
     *
     * @link https://github.com/redisson/redisson/blob/master/redisson/src/main/java/org/redisson/RedissonLock.java
     *
     * redis-cli 可以通过ttl 来观察key的时效
     */
    @Test
    public void test_lock_unlock() {
        RedissonWithLock lock = new RedissonWithLock();
        lock.execute();
    }
}

<img src="img/ 加锁解锁流程图.png" alt="img" style="zoom: 67%;" />

读源码Redisson分析

参考文献

[CSDN]https://blog.csdn.net/w372426096/article/details/103761286

[官方文档]https://github.com/redisson/redisson/wiki/8.-Distributed-locks-and-synchronizers

[GITHUB源码]https://github.com/redisson/redisson/blob/master/redisson/src/main/java/org/redisson/RedissonLock.java

• org.redisson.RedissonLock#tryLock(long, long, java.util.concurrent.TimeUnit) 加锁
  • org.redisson.RedissonLock#tryAcquire
    • org.redisson.RedissonLock#tryAcquireAsync
      • org.redisson.RedissonLock#tryLockInnerAsync
        • org.redisson.RedissonLock#evalWriteAsync 实际执行Lua脚本的地方 evalWriteAsync()
      • org.redisson.RedissonLock#scheduleExpirationRenewal
• org.redisson.pubsub.PublishSubscribe
  • org.redisson.RedissonLock#subscribe
  • org.redisson.RedissonLock#unsubscribe
• java.util.concurrent.locks.Lock#unlock 解锁
  • org.redisson.RedissonLock#unlock
    • org.redisson.RedissonLock#unlockAsync(long)
      • org.redisson.RedissonLock#unlockInnerAsync 实际执行Lua脚本的地方evalWriteAsync()

读源码得出的小结

加锁流程核心就3步
Step1：尝试获取锁，这一步是通过执行加锁Lua脚本来做
Step2：若第一步未获取到锁，则去订阅解锁消息，当获取锁到剩余过期时间后，调用信号量方法阻塞住，直到被唤醒或等待超时
Step3：一旦持有锁的线程释放了锁，就会广播解锁消息。于是，第二步中的解锁消息的监听器会释放信号量，获取锁被阻塞的那些线程就会被唤醒，并重新尝试获取锁

Watchdog机制(看门狗)-- 有时间再研究,先放放

暂时空白

引言说明

需要特别注意的是，RedissonLock 同样没有解决 节点挂掉的时候，存在丢失锁的风险的问题。而现实情况是有一些场景无法容忍的，所以 Redisson 提供了实现了redlock算法的 RedissonRedLock，RedissonRedLock 真正解决了单点失败的问题，代价是需要额外的为 RedissonRedLock 搭建Redis环境。所以，如果业务场景可以容忍这种小概率的错误，则推荐使用 RedissonLock， 如果无法容忍，则推荐使用 RedissonRedLock。

★ Redis分布式锁的正确姿势

要使用Redis实现分布式锁。

加锁操作的正确姿势为：

1. 使用setnx命令保证互斥性
2. 需要设置锁的过期时间，避免死锁
3. setnx和设置过期时间需要保持原子性，避免在设置setnx成功之后在设置过期时间客户端崩溃导致死锁
4. 加锁的Value 值为一个唯一标示。可以采用UUID作为唯一标示。加锁成功后需要把唯一标示返回给客户端来用来客户端进行解锁操作

解锁的正确姿势为：

1. 需要拿加锁成功的唯一标示要进行解锁，从而保证加锁和解锁的是同一个客户端
2. 解锁操作需要比较唯一标示是否相等，相等再执行删除操作。这2个操作可以采用Lua脚本方式使2个命令的原子性。

Redis分布式锁实现的正确姿势的实现代码

public interface DistributedLock {
    /**
     * 获取锁
     * @return 锁标识
     */
    String acquire();

    /**
     * 释放锁
     * @param indentifier
     * @return
     */
    boolean release(String indentifier);
}

import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.params.SetParams;

import java.util.Collections;
import java.util.UUID;

@Slf4j
public class RedisDistributedLock implements DistributedLock {

    private static final String LOCK_SUCCESS = "OK";
    private static final Long RELEASE_SUCCESS = 1L;
    private static final String SET_IF_NOT_EXIST = "NX";
    private static final String SET_WITH_EXPIRE_TIME = "PX";

    /**
     * redis 客户端
     */
    private Jedis jedis;

    /**
     * 分布式锁的键值
     */
    private String lockKey;

    /**
     * 锁的超时时间 10s
     */
    int expireTime = 10;

    /**
     * 锁等待，防止线程饥饿
     */
    int acquireTimeout = 1 * 1000;

    /**
     * 获取指定键值的锁
     *
     * @param jedis   jedis Redis客户端
     * @param lockKey 锁的键值
     */
    public RedisDistributedLock(Jedis jedis, String lockKey) {
        this.jedis = jedis;
        this.lockKey = lockKey;
    }

    /**
     * 获取指定键值的锁,同时设置获取锁超时时间
     *
     * @param jedis          jedis Redis客户端
     * @param lockKey        锁的键值
     * @param acquireTimeout 获取锁超时时间
     */
    public RedisDistributedLock(Jedis jedis, String lockKey, int acquireTimeout) {
        this.jedis = jedis;
        this.lockKey = lockKey;
        this.acquireTimeout = acquireTimeout;
    }

    /**
     * 获取指定键值的锁,同时设置获取锁超时时间和锁过期时间
     *
     * @param jedis          jedis Redis客户端
     * @param lockKey        锁的键值
     * @param acquireTimeout 获取锁超时时间
     * @param expireTime     锁失效时间
     */
    public RedisDistributedLock(Jedis jedis, String lockKey, int acquireTimeout, int expireTime) {
        this.jedis = jedis;
        this.lockKey = lockKey;
        this.acquireTimeout = acquireTimeout;
        this.expireTime = expireTime;
    }

    /**
     * 获取锁
     *
     * @return
     */
    @Override
    public String acquire() {
        try {
            // 获取锁的超时时间，超过这个时间则放弃获取锁
            long end = System.currentTimeMillis() + acquireTimeout;
            // 随机生成一个value
            String requireToken = UUID.randomUUID().toString();
            while (System.currentTimeMillis() < end) {
                String result = jedis.set(lockKey, requireToken, SetParams.setParams().nx().ex(expireTime));
                if (LOCK_SUCCESS.equals(result)) {
                    return requireToken;
                }
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        } catch (Exception e) {
            log.error("acquire lock due to error", e);
        }

        return null;
    }

    /**
     * 释放锁
     *
     * @param identify
     * @return
     */
    @Override
    public boolean release(String identify) {
        if (identify == null) {
            return false;
        }

        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        Object result = new Object();
        try {
            result = jedis.eval(script, Collections.singletonList(lockKey),
                    Collections.singletonList(identify));
            if (RELEASE_SUCCESS.equals(result)) {
                log.info("release lock success, requestToken:{}", identify);
                return true;
            }
        } catch (Exception e) {
            log.error("release lock due to error", e);
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }

        log.info("release lock failed, requestToken:{}, result:{}", identify, result);
        return false;
    }
}

import redis.clients.jedis.Jedis;

/**
 * 下面就以秒杀库存数量为场景，测试下上面实现的分布式锁的效果
 * 单元测试时 测试redis分布式锁
 */
public class RedisDistributedLockTest {

    // 200个商品
    static int n = 200;
    // 秒杀 商品数量减少
    public static void secskill() {
        System.out.println(--n);
    }

    public static void main(String[] args) {
        Runnable runnable = () -> {
            RedisDistributedLock lock = null;
            String unLockIdentify = null;
            try {
                // TODO 可改为 jedisPoll
                Jedis conn = new Jedis("127.0.0.1",6379);
                lock = new RedisDistributedLock(conn, "test1");
                unLockIdentify = lock.acquire();
//                System.out.println(Thread.currentThread().getName() + "正在运行");
                secskill();
            } finally {
                if (lock != null) {
                    lock.release(unLockIdentify);
                }
            }
        };

        for (int i = 0; i < 1000; i++) {
            Thread t = new Thread(runnable);
            t.start();
        }
    }

}

★ Redis的延迟队列

通过Redis的 zset(有序列表) 来实现

/**
 * 2. redis的延迟队列
 * 通过Redis的 zset(有序列表) 来实现
 *
 * @author huichen
 * @date 2020-06
 */
public class RedisDelayingQueue<T> {

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

    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;
    }

   /**
     * 入队列
     *
     * @param msg
     */
    public void delay(T msg) {
        TaskItem task = new TaskItem<T>();
        task.id = UUID.randomUUID().toString();
        task.msg = msg;
        String s = JSON.toJSONString(task);
        jedis.zadd(queueKey, System.currentTimeMillis() + 5000, s);
    }

    /**
     * 遍历, 出队列
     */
    public void loop() {
        while (!Thread.interrupted()) {
            // zrangeByScore 返回有序集合中指定分数区间的成员列表。有序集成员按分数值递增(从小到大)次序排列
            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();
            // 移除zset中的元素
            if (jedis.zrem(queueKey, s) > 0) {
                TaskItem<T> task = JSON.parseObject(s, TaskType);
                this.handleMsg(task.msg);
            }
        }
    }

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

/**
 * 单元测试
 * redis延迟队列
 */
public class RedisDelayingQueueTest {

    @Test
    public void testDelayQueue() {
        Jedis jedis = new Jedis();
        RedisDelayingQueue<String> queue = new RedisDelayingQueue<>(jedis, "q-demo");
        // 生产线程
        Thread producer = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++) {
                    queue.delay("codehole" + i);
                }
            }
        });

                // 消费线程
        Thread consumer = new Thread(new Runnable() {
            @Override
            public void run() {
                queue.loop();
            }
        });

        producer.start();
        consumer.start();

        try {
            producer.join();
            Thread.sleep(10 * 1000);
            // 中断
            consumer.interrupt();
            consumer.join();
        } catch (InterruptedException e) {

        }
    }
}

producer(生产线程) 循环添加元素到有序列表中(zset)

consumer(消费线程) 循环遍历zset有序集合.消费zset中的元素对象

★ Redis位图(bit)

用户签到的场景.比如用户一年的签到记录. 签到状态用0,1标识.如果用记录表示那么一个用户一年就会有365+的记录.

然后,上亿用户呢. 这个数量集一下就很大了.

Redis提供的位图数据结构便可以解决这个问题, 从而大大的节约存储空间.以一年的记录为参考. 365天就是365个bit.8个bit为1个字节. 那么,365个b 约为 46个B(向上取整就)

思考用位图来实现用户一年的签到记录, Redis位图的基本语法是setbit/getbit

• 零存整取

以 hello 为例. 我们通过ASCII码对照表可以将其转为 1101000 1100101 1101100 1101100 1101111

其中 h -> 1101000 可以看到在 1,2,4位上是1 其余为0. 我们只要在1,2,4位上设置值即可

127.0.0.1:6379> setbit str 1 1
(integer) 0
127.0.0.1:6379> setbit str 2 1
(integer) 0
127.0.0.1:6379> setbit str 4 1
(integer) 0
127.0.0.1:6379> get str
"h"
......

setbit str 1 1 为零存
get str 即为整取

• 零存零取

以hello为例.

127.0.0.1:6379> getbit str 1
(integer) 1
127.0.0.1:6379> getbit str 2
(integer) 1
127.0.0.1:6379> getbit str 3
(integer) 0
127.0.0.1:6379> getbit str 4
(integer) 1
127.0.0.1:6379>

setbit str 1 1 ## 为零存

getbit str 1 ##即为零取

• 整存零取

127.0.0.1:6379> set say hello
OK
127.0.0.1:6379> getbit say 1
(integer) 1
127.0.0.1:6379> getbit say 2
(integer) 1
127.0.0.1:6379> getbit say 3
(integer) 0
127.0.0.1:6379> getbit say 4
(integer) 1

.....

Redis位图的统计和查找

• bitcount 用来统计指定范围内 1 的个数

• bitpos 用来查找指定范围内出现的第一个 0 或者 1

魔术指令(bitfield)

???没弄太懂 这个可以干啥. 官方例子看好像是可以处理多个位图上的值. 可以通过 incrby 进行自增操作

• bitfield

★ Redis的高级数据结构

HyperLogLog

举个🌰 统计UV 和 PV

• UV(Unique Visitor):即独立访客,访问您网站的一台电脑客户端为一个访客。
• PV(Page View),即页面浏览量,或点击量;用户每1次对网站中的每个网页访问均被记录1次

PV简单页面浏览量累计求和即可.

• 给每个页面配置一个独立的 Redis 计数器就可以了，把这个计数器的 key 后缀加上当天的日期。这样每来一个请求，就执行 INCRBY 指令一次，最终就可以统计出所有的 PV 数据了。

UV要去重,同一个用户一天之内的多次访问请求只能计数一次

• 同一个用户一天之内的多次访问请求只能计数一次, 需要一个唯一 ID 来标识

HyperLogLog 是用来做基数统计的算法

特点: 每个 HyperLogLog 键只需要花费 12 KB 内存，就可以计算接近 2^64 个不同元素的基数

参考文献

[Redis官方中文网]https://www.redis.net.cn/

[神奇的HyperLoglog解决统计问题]https://www.wmyskxz.com/2020/03/02/reids-4-shen-qi-de-hyperloglog-jie-jue-tong-ji-wen-ti/

★ 布隆过滤器

Docker 命令

# my local docker
docker exec -it redis-test /bin/bash
redis-cli