Redis可能会阻塞的情况

Posted 2023-04-12

参考技术A 如果一个值的size过大，写入时开辟内存以及发送时的数据 copy 开销都会很大。
建议从业务上对大key做拆分。

对于一些数据结构的操作，时间复杂度为 O(N) ，如果不加控制，可能会引起阻塞。
例如 Keys 命令，由于没有limit参数，会全表扫描，耗时大。可以考虑用Scan替代。

尽管使用了IO多路复用技术，读写 copy 以及 User Mode 和 Kernel Mode 的切换会比较耗时。

尽管 RDB 是fork后独立进程中完成落盘工作，fork 这个 System Call 本身耗时大概在700ms级别。
建议尽量避免在高峰时期执行 Save 或 Bgsave 命令。

由于AOF需要修改磁盘中的日志文件，修改文件及其iNode需要两次随机读写IO，大约耗时在20ms级别，因此业务中几乎不会开启 always 模式。
一般都用 Everysecond模式。

由于 Redis 的删除过期键策略中有一条是主动删除：会随机抽出100个设置了过期的key，对已过期的进行删除，如果发现过期的key超过25个，就会重复这个过程。因此，如果有大量同一时间过期的key，会在主动删除触发时，不停地取key删key，造成阻塞。
建议在设置过期时间时使用 Expire 而非 Expireat，或者使用 Expireat 时自己给入一个随机量，让过期时间离散开。

当 Redis 可支配的内存空间不足时，会进行内存逐出操作。尽管可以配置策略，但是逐出时CPU会hang住。
建议对内存使用情况做监控，及时扩容或进行其他人为介入操作。

Redis如何实现分布式阻塞队列？

1. Redis分布式锁实现原理

分布式锁本质上要实现的目标就是在 Redis 里面占一个“茅坑”，当别的进程也要来占时，发现已经有人蹲在那里了，就只好放弃或者稍后再试。占坑一般是使用 setnx(set if not exists) 指令，只允许被一个客户端占坑。先来先占， 用完了，再调用 del 指令释放茅坑。
死锁问题：如果逻辑执行到中间出现异常了，可能会导致 del 指令没有被调用，这样就会陷入死锁，锁永远得不到释放， 解决这个问题我们在拿到锁之后，再给锁加上一个过期时间，比如 5s，这样即使中间出现异常也可以保证 5 秒之后锁会自动释放

2. 普通非阻塞锁实现

public class RedisLock {
    private Jedis jedis;
    public RedisLock(Jedis jedis) {
        this.jedis = jedis;
    }
    public boolean lock(String key) {
    return jedis.set(key, "", "nx", "ex", 5L) != null;
    }
    public void unlock(String key) {
        jedis.del(key);
    }
}

2.1 存在问题

• 如果某一个进程没有拿到锁得到了false的结果那么次进程是否执行当前任务？显然对于一般情况来说我们的任务都是必须执行的那么此时我们就要考虑该何时执行了，在传统的锁中我们如果没有拿到锁线程就进入了阻塞状态那么此处我们是否可以改进同样实现阻塞唤醒机制

3. 分布式阻塞锁具体实现

3.1 解决思路

1. 首先我们改造lock锁，当不能创建key时就利用当前key阻塞当前线程
2. 当某一个线程释放锁时通过redis的pub/sub发送一个消息消息内容为key
3. 所有使用锁的应用监听lock通道的消息，在收到消息时通过key唤醒对应线程

3.2具体实现

package com.hgy.common.redis;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

import java.util.HashMap;

public class RedisLock extends JedisPubSub {

    //是否已经初始化监听
    private static volatile boolean isListen = false;
    //每一个redis的key对应一个阻塞对象
    private  HashMap<String, Object> blockers = new HashMap<>();

    private Jedis jedis;

    //当前获得锁的线程
    private Thread curThread;



    public RedisLock(Jedis jedis) {
        this.jedis = jedis;

        //保证没一个应用只初始化一次监听
        if (!isListen) {
            synchronized (RedisLock.class) {
                if (!isListen) {
                    // 启动一个线程做消息监听
                    new Thread(()->{
                        new Jedis("192.168.200.128", 6379).subscribe(this, "lock");
                    }).start();
                    isListen = true;
                }
            }
        }
    }


    public void lock(String key) throws InterruptedException {
        //循环判断是否能够创建key， 不能则直接wait释放CPU执行权
        while (jedis.set(key, "", "nx", "ex", 20L) == null) {
            synchronized (key) {
                System.out.println(Thread.currentThread().getName() + "=======" + key);
                blockers.put(key, key);
                key.wait();
            }
        }
        blockers.put(key, key);
        //能够成功创建，获取锁成功记录当前获取锁线程
        curThread = Thread.currentThread();
    }

    public void unlock(String key) {
        //判断是否为加锁的线程执行解锁， 不是则直接忽略
        if( curThread == Thread.currentThread()) {
            jedis.del(key);
            //删除key之后需要notifyAll所有的应用， 所以这里采用发订阅消息给所有的应用
            jedis.publish("lock", key);
        }
    }


    /**
     * 所有应用接收到消息后在当前应用中执行对应key的notifyAll方法
     * @param channel
     * @param message
     */
    @Override
    public void onMessage(String channel, String message) {
        Object lock = blockers.get(message);
        if(lock != null) {
            synchronized (lock) {
                lock.notifyAll();
            }
        }
    }
}

4. 测试

目标： 开启两个mian线程， 在第一个中首先暂停3秒然后打印1-100然后线程休眠5秒释放锁并打印最后的毫秒数； main1在执行的同时执行main2，在2中打印开始时间；最后比对1和2的开始时间即可验证

注意： 先启动1然后启动2

• main1

package com.hgy;

import com.hgy.common.redis.RedisLock;
import redis.clients.jedis.Jedis;

public class RedisLockApp1 {

    private static RedisLock redisLock;

    public static void main(String[] args) throws InterruptedException {
        Jedis client = new Jedis("192.168.200.128", 6379);
        redisLock = new RedisLock(client);

        redisLock.lock("demo");
        Thread.sleep(3000);
        for (int i = 0; i < 100; i++) {
            System.out.println("app1" + i);
        }
        Thread.sleep(5000);
        redisLock.unlock("demo");
        System.out.println("App1==> end:" + System.currentTimeMillis());
    }
}

• main2

package com.hgy;

import com.hgy.common.redis.RedisLock;
import redis.clients.jedis.Jedis;

public class RedisLockApp2 {
    private static RedisLock redisLock;

    public static void main(String[] args) throws InterruptedException {
        Jedis client = new Jedis("192.168.200.128", 6379);
        redisLock = new RedisLock(client);

        redisLock.lock("demo");
        System.out.println("App2==> start:" + System.currentTimeMillis());
        for (int i = 0; i < 100; i++) {
            System.out.println("app2" + i);
        }
        redisLock.unlock("demo");
    }
}

注意

如果细心的小伙伴儿可能已经发现了unlock其实不是一个原子操作，可能在未发布消息但删除key之后的这段时间如果有人此时执行lock那么可以直接拿到锁；但是影响不大因为拿到锁之后其他被阻塞的线程被唤醒之后将会继续阻塞

此处unlock中有两个操作，删除key和发送消息如果在这两个操作之间机器异常并没有新的线程抢占锁那么此时被阻塞的线程将永远阻塞