Redis应用实战及注意事项

亦凉 2022-08-31 00:47 151阅读 0赞

## Redis的使用 ##

引入pom

<dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson-spring-boot-starter</artifactId>
        <version>3.16.0</version>
    </dependency>

修改application.properties

使用

@Autowired
    RedisTemplate redisTemplate;

Lettuce（默认客户端）

## Redis的Java客户端 ##

*  Jedis
 *  Lettuce
 *  Redisson

### Redis的请求通讯协议 ###

> set n v
> 
> **\*3\\r\\n$3\\r\\nset\\r\\n$1\\r\\nn\\r\\n$1\\r\\nv**
> 
> \*3表示有3个指令
> 
> $3表示指令长度为3：set
> 
> $1标识指令长度为1：n
> 
> $1标识指令长度为1：v

### 如何优化 ###

*  通信层面的优化（使用NIO）
 *  是否采用异步通信（多线程）
 *  针对key和value的序列化
 *  使用Java8特性
 *  解决方案

## 使用过程中的问题 ##

### **数据库和redis的数据一致性问题（保证最终一致性）** ###

先更新数据库，再删除缓存（相当于被动更新缓存）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70][]

先删除缓存，再更新数据库

### **缓存雪崩** ###

**出现的原因**：

*  大量热点数据同时失效
 *  或者redis出现故障
 *  总之由于redis的key过期导致

**解决方案：**

*  key过期时间错开，或者设置随机过期时间
 *  对于热点数据，没必要设置过期时间
 *  redis集群，灾备方案，保证高可用，避免redis故障

### **缓存穿透** ###

redis和mysql都不存在这样的数据的情况，有可能是恶意攻击，需要通过布隆过滤器解决缓存穿透

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 1][]

### 布隆过滤器BloomFilter ###

引用pom，创建BloomFilter，类似于HashTable，将所有需要查询的数据全部存入BloomFilter中，每次查询redis缓存之前先查询key是否存在，如果不存在则直接返回，避免去redis缓存中查询，查不到还需要查询数据库从而导致缓存穿透

<dependency>
        <groupId>com.google.guava</groupId>
        <artifactId>guava</artifactId>
        <version>19.0</version>
    </dependency>

所有数据通过key存入BloomFilter中，每个key通过多个hash函数进行计算得到一个值存入；每次查询之前先通过BloomFilter进行检查，通过多个hash函数对key进行计算，如果多个hash结果和存入时的结果完全相同，则代表可能存在，如果有一个hash计算结果不相同，则代表查询的数据肯定不存在，从而避免查询redis缓存也避免查询数据库导致缓存穿透

BloomFilter<String> bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF-8), 10000000, 0.03);

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 2][]

### **过期key的删除** ###

**被动删除**

通过get(name)从过期key存储表中拿到name的过期时间，发现过期了，返回失效

**主动删除**

随机抽取20个key，删除20个key中已经过期的key，如果发现20个key中有25%已经过期，那么再次执行随机抽取

# Redis实现分布式锁 #

实现分布式锁有很多方式：redis、zookeeper、mysql、etcd

**锁的类型**

*  排他锁：不允许多个程序（线程、进程）同时访问某个共享资源
 *  共享锁：用于不更新或不更改操作，只读操作

**分布式锁**：跨进程实现共享资源的互斥

### Redis分布式锁原理 ###

*  通过Lua脚本执行redis命令实现通信
 *  如果指令还没有执行完成，通过自动续约机制
 *  通过看门狗lockWatchdogTimeout定期检查时间

redis之所以可以实现分布式锁，是因为：（1）redis是一个独立部署的中间件，可以对资源进行存储；（2）多个客户端之间不知道是否可以对一个共享资源进行访问，这个时候需要把共享资源的状态存放在redis中，客户端访问共享资源之前先访问redis获取共享资源状态，查看是否可以访问，如果可以访问再访问共享资源，redis起到分布式协调作用；（3）redis的setnx命令可以实现共享互斥变量；（4）redis的setnx命令满足原子性；

public class RedisLockDemo {
        private static RedissonClient redissonClient;
        static {
            Config config = new Config();
            config.useSingleServer().setAddress("redis://192.168.2.101:6379");
            redissonClient = Redisson.create(config);
        }
    
        public static void main(String[] args) throws InterruptedException {
            RLock rLock = redissonClient.getLock("supportUpdate");
            if (rLock.tryLock()) { //返回true，表示获得锁成功
                System.out.println("获取锁成功");
            } else {
                System.out.println("获取锁失败");
            }
            Thread.sleep(2000);
            rLock.unlock(); //释放锁
        }
    }

抢占锁原理

public boolean tryLock() {
        return (Boolean)this.get(this.tryLockAsync());
    }
    
    public RFuture<Boolean> tryLockAsync() {
        return this.tryLockAsync(Thread.currentThread().getId());
    }
    
    public RFuture<Boolean> tryLockAsync(long threadId) {
        RPromise<Boolean> result = new RedissonPromise();
        RFuture<Long> longRFuture = this.tryAcquireAsync(-1L, (TimeUnit)null, threadId);
        longRFuture.onComplete((res, e) -> {
            if (e != null) {
                result.tryFailure(e);
            }
    
            result.trySuccess(res == null);
        });
        return result;
    }
    
    // 返回RFuture，表示这里是异步调用返回结果
    private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {
        if (leaseTime != -1L) { //如果设置了锁超时时间
            return this.tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        } else { //如果没有设置锁超时时间，默认30S
            RFuture<Long> ttlRemainingFuture = this.tryLockInnerAsync(this.internalLockLeaseTime, TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
            ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
                if (e == null) {
                    if (ttlRemaining == null) {
                        // 续约/续期机制
                        this.scheduleExpirationRenewal(threadId);
                    }
    
                }
            });
            return ttlRemainingFuture;
        }
    }
    
    // 第四个参数script string是Lua脚本，通过Lua脚本执行Redis命令
    // Lua脚本的执行是原子的，所以这一堆script是原子指令
    <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        this.internalLockLeaseTime = unit.toMillis(leaseTime);
        return this.evalWriteAsync(this.getRawName(), LongCodec.INSTANCE, command, "if (redis.call('exists', KEYS[1]) == 0) then redis.call('hincrby', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end; if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('hincrby', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end; return redis.call('pttl', KEYS[1]);", Collections.singletonList(this.getRawName()), new Object[]{this.internalLockLeaseTime, this.getLockName(threadId)});
    }
    
    protected final <V> V get(RFuture<V> future) {
        return this.commandExecutor.get(future);
    }
    
    // get方法是阻塞等待获取
    public <V> V get(RFuture<V> future) {
        if (Thread.currentThread().getName().startsWith("redisson-netty")) {
            throw new IllegalStateException("Sync methods can't be invoked from async/rx/reactive listeners");
        } else {
            try {
                future.await();
            } catch (InterruptedException var3) {
                Thread.currentThread().interrupt();
                throw new RedisException(var3);
            }
    
            if (future.isSuccess()) {
                return future.getNow();
            } else {
                throw this.convertException(future);
            }
        }
    }

释放锁原理

public void unlock() {
        try {
            this.get(this.unlockAsync(Thread.currentThread().getId()));
        } catch (RedisException var2) {
            if (var2.getCause() instanceof IllegalMonitorStateException) {
                throw (IllegalMonitorStateException)var2.getCause();
            } else {
                throw var2;
            }
        }
    }
    
    protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        return this.evalWriteAsync(this.getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if (redis.call('hexists', KEYS[1], ARGV[2]) == 0) then return nil;end; local counter = redis.call('hincrby', KEYS[1], ARGV[2], -1); if (counter > 0) then redis.call('pexpire', KEYS[1], ARGV[1]); return 0; else redis.call('del', KEYS[1]); return 1; end; return nil;", Collections.singletonList(this.getRawName()), new Object[]{this.internalLockLeaseTime, this.getLockName(threadId)});
    }

### Lua脚本 ###

类似javascript的脚本语言

可以自定义redis的相关操作指令

原子性，一个Lua脚本是一个原子性指令

复用性：可以在Redis客户端定义一个Lua脚本，并复用调用

public class LuaDemo {
        private static RedissonClient redissonClient;
        static {
            Config config = new Config();
            config.useSingleServer().setAddress("redis://192.168.2.101:6379");
            redissonClient = Redisson.create(config);
        }
    
        private static final String LUA_SCRIPT = "return redis.call('incr', KEYS[1])";
    
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            RScript rScript = redissonClient.getScript();
            List<String> keys = Arrays.asList("LIMIT:" + 1);
            RFuture<String> rFuture = rScript.evalAsync(RScript.Mode.READ_WRITE, LUA_SCRIPT, RScript.ReturnType.INTEGER, keys);
            System.out.println(rFuture.get());
        }
    }

**在Lua脚本中调用redis命令**

redis.call('set', 'key', 'value');

local value = redis.call('get', 'key');

**Lua的原子性**

如果有一个程序比如redis正在通过Lua脚本执行某个指令，其他应用程序访问redis时会提示处于繁忙状态，这样做的目的是避免其他操作干扰数据的准确性。通过script kill或者shutdown nosave可以终止这个指令。可以通过lua-time-limit参数设置lua脚本执行的时间。

**evalsha指令**

通过script load "return redis.call('get', 'name')"可以将后面的lua脚本生成一个hash值

通过evalsha指令可以直接通过hash值执行对应的lua脚本

### 时间轮机制 ###

kafka、zookeeper、dubbo、netty都用到了时间轮机制

public class HashedWheelTimerDemo {
        public static void main(String[] args) {
            HashedWheelTimer hashedWheelTimer = new HashedWheelTimer(new DefaultThreadFactory(
                    "timer-pool"), 100, TimeUnit.MILLISECONDS, 8, false);
            hashedWheelTimer.newTimeout(task -> {
                System.out.println("需要执行的任务，延迟1秒钟执行任务");
            }, 1, TimeUnit.SECONDS);
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 3][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70]: /images/20220829/c072320cc94c4d1fbfada04a7eb29761.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 1]: /images/20220829/443b0596cfcd4227afe73d8383c903c2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 2]: /images/20220829/f8420208fdf644cf9e272e5dea4bc88d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 3]: /images/20220829/4416d497a9ab46a8887cde44276ce1aa.png