Redis大杀器，深入剖析，从简入深

//思路1)点击参与抽奖加入集合SADD key {userID}2)查看参与抽奖所有用户SMEMBERS key3)抽count名中奖者如果希望中奖用户依然保留在抽奖池里面SRANDMENMBER key [count]SRANDMENMBER 从集合key中选出count个元素,元素不从key中删除如果不希望中奖用户依然保留在抽奖池里面SPOP key [count]SPOP从集合key中选出count个元素,元素从key中删除

//思路1)抖音大V视频点赞量SADD like:{messageId}  {userId}2)取消点赞SREM like:{messageId} {userId}移除集合 key 中的一个或多个 member 元素，不存在的 member 元素会被忽略3)检查用户是否点过赞SISMEMBER like:{messageId} {userId}4)获取点赞的用户列表SISMEMBER like:{messageId}5)获取点赞用户数返回集合 key 的基数(集合中元素的数量)SCARD like:{messageId}

//思路共同关注的人://两个集合的交集SINTER simaSet yanguoSet我关注的人也关注他//我关注的人在他的集合里面,可以运用SISMENBERSISMENBER simaSet yanguo我可能认识的人//他关注的人,不在我的集合里面,但是在我的朋友集合里面SDIFF simaSet yanguoSet

//思路1)热搜//新闻点击量ZINCRBY HOTNEWS:20191115 1 李小璐离婚为有序集合key中元素member的分值加上increment2)展示当日排行版前十ZREVRANGE hotNews:20191115 0 10 WITHSCORESZREVRANGE倒序获取有序集合key中start下标到stop下标的元素3)七日搜索榜单计算ZUNIONSTORE hotNews:20190722 7并集计算4)展示七日排行前十ZREVRANGE hotNews:20190716-20190722 0 10 WITHSCORES交集计算

//思路微博、微信、默默<附近的人>滴滴打车、哈啰单车、美团单车<附近的车>美团和饿了吗<附近的餐馆>
基于地理位置比如GEORADIUSredis> GEOADD Sicily 13.361389 38.115556 "Palermo" 15.087269 37.502669 "Catania"(integer) 2
redis> GEORADIUS Sicily 15 37 200 km WITHDIST1) 1) "Palermo"   2) "190.4424"2) 1) "Catania"   2) "56.4413"
redis> GEORADIUS Sicily 15 37 200 km WITHCOORD1) 1) "Palermo"   2) 1) "13.361389338970184"      2) "38.115556395496299"2) 1) "Catania"   2) 1) "15.087267458438873"      2) "37.50266842333162"
redis> GEORADIUS Sicily 15 37 200 km WITHDIST WITHCOORD1) 1) "Palermo"   2) "190.4424"   3) 1) "13.361389338970184"      2) "38.115556395496299"2) 1) "Catania"   2) "56.4413"   3) 1) "15.087267458438873"      2) "37.50266842333162"

Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:5379").setPassword("123456").setDatabase(0);
RedissonClient redissonClient = Redisson.create(config);
RLock rLock = redissonClient.getLock(lockKey);try {         * 4.尝试获取锁     * waitTimeout 尝试获取锁的最大等待时间，超过这个值，则认为获取锁失败     * leaseTime   锁的持有时间,超过这个时间锁会自动失效（值应设置为大于业务处理的时间，确保在锁有效期内业务能处理完）     */    boolean res = rLock.tryLock((long)waitTimeout, (long)leaseTime, TimeUnit.SECONDS);    if (res) {            }} catch (Exception e) {    throw new RuntimeException("aquire lock fail");}finally{        rLock.unlock();}

-- 若锁不存在：则新增锁，并设置锁重入计数为1、设置锁过期时间if (redis.call('exists', KEYS[1]) == 0) then    redis.call('hset', KEYS[1], ARGV[2], 1);    redis.call('pexpire', KEYS[1], ARGV[1]);    return nil;end; -- 若锁存在，且唯一标识也匹配：则表明当前加锁请求为锁重入请求，故锁重入计数+1，并再次设置锁过期时间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]);

-- 若锁不存在：则直接广播解锁消息，并返回1if (redis.call('exists', KEYS[1]) == 0) then    redis.call('publish', KEYS[2], ARGV[1]);    return 1; end; -- 若锁存在，但唯一标识不匹配：则表明锁被其他线程占用，当前线程不允许解锁其他线程持有的锁if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then    return nil;end;  -- 若锁存在，且唯一标识匹配：则先将锁重入计数减1local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); if (counter > 0) then     -- 锁重入计数减1后还大于0：表明当前线程持有的锁还有重入，不能进行锁删除操作，但可以友好地帮忙设置下过期时期    redis.call('pexpire', KEYS[1], ARGV[2]);     return 0; else     -- 锁重入计数已为0：间接表明锁已释放了。直接删除掉锁，并广播解锁消息，去唤醒那些争抢过锁但还处于阻塞中的线程    redis.call('del', KEYS[1]);     redis.call('publish', KEYS[2], ARGV[1]);     return 1;end; return nil;

@Override    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {        // 获取锁能容忍的最大等待时长        long time = unit.toMillis(waitTime);        long current = System.currentTimeMillis();        final long threadId = Thread.currentThread().getId();         // 【核心点1】尝试获取锁，若返回值为null，则表示已获取到锁        Long ttl = tryAcquire(leaseTime, unit, threadId);        // lock acquired        if (ttl == null) {            return true;        }         // 还可以容忍的等待时长=获取锁能容忍的最大等待时长 - 执行完上述操作流逝的时间        time -= (System.currentTimeMillis() - current);        if (time <= 0) {            acquireFailed(threadId);            return false;        }         current = System.currentTimeMillis();        // 【核心点2】订阅解锁消息，见org.redisson.pubsub.LockPubSub#onMessage        /**     * 4.订阅锁释放事件，并通过await方法阻塞等待锁释放，有效的解决了无效的锁申请浪费资源的问题：     * 基于信息量，当锁被其它资源占用时，当前线程通过 Redis 的 channel 订阅锁的释放事件，一旦锁释放会发消息通知待等待的线程进行竞争     * 当 this.await返回false，说明等待时间已经超出获取锁最大等待时间，取消订阅并返回获取锁失败     * 当 this.await返回true，进入循环尝试获取锁     */        final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);        //await 方法内部是用CountDownLatch来实现阻塞，获取subscribe异步执行的结果（应用了Netty 的 Future）        if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {            if (!subscribeFuture.cancel(false)) {                subscribeFuture.addListener(new FutureListener<RedissonLockEntry>() {                    @Override                    public void operationComplete(Future<RedissonLockEntry> future) throws Exception {                        if (subscribeFuture.isSuccess()) {                            unsubscribe(subscribeFuture, threadId);                        }                    }                });            }            acquireFailed(threadId);            return false;        }         // 订阅成功        try {            // 还可以容忍的等待时长=获取锁能容忍的最大等待时长 - 执行完上述操作流逝的时间            time -= (System.currentTimeMillis() - current);            if (time <= 0) {                // 超出可容忍的等待时长，直接返回获取锁失败                acquireFailed(threadId);                return false;            }             while (true) {                long currentTime = System.currentTimeMillis();                // 尝试获取锁；如果锁被其他线程占用，就返回锁剩余过期时间【同上】                ttl = tryAcquire(leaseTime, unit, threadId);                // lock acquired                if (ttl == null) {                    return true;                }                 time -= (System.currentTimeMillis() - currentTime);                if (time <= 0) {                    acquireFailed(threadId);                    return false;                }                 // waiting for message                currentTime = System.currentTimeMillis();                 // 【核心点3】根据锁TTL，调整阻塞等待时长；                // 注意：这里实现非常巧妙，1、latch其实是个信号量Semaphore，调用其tryAcquire方法会让当前线程阻塞一段时间，避免了在while循环中频繁请求获取锁；               //2、该Semaphore的release方法，会在订阅解锁消息的监听器消息处理方法org.redisson.pubsub.LockPubSub#onMessage调用；当其他线程释放了占用的锁，会广播解锁消息，监听器接收解锁消息，并释放信号量，最终会唤醒阻塞在这里的线程。                if (ttl >= 0 && ttl < time) {                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);                } else {                    getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);                }                 time -= (System.currentTimeMillis() - currentTime);                if (time <= 0) {                    acquireFailed(threadId);                    return false;                }            }        } finally {            // 取消解锁消息的订阅            unsubscribe(subscribeFuture, threadId);        }    }


tryAcquire的实现，真的就是执行Lua脚本！

private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {    // tryAcquireAsync异步执行Lua脚本，get方法同步获取返回结果    return get(tryAcquireAsync(leaseTime, unit, threadId));} //  见org.redisson.RedissonLock#tryAcquireAsyncprivate <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {    if (leaseTime != -1) {        // 实质是异步执行加锁Lua脚本        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);    }    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);    ttlRemainingFuture.addListener(new FutureListener<Long>() {        @Override        public void operationComplete(Future<Long> future) throws Exception {            //先判断这个异步操作有没有执行成功，如果没有成功，直接返回，如果执行成功了，就会同步获取结果            if (!future.isSuccess()) {                return;            }             Long ttlRemaining = future.getNow();            // lock acquired            //如果ttlRemaining为null，则会执行一个定时调度的方法scheduleExpirationRenewal            if (ttlRemaining == null) {                scheduleExpirationRenewal(threadId);            }        }    });    return ttlRemainingFuture;} // 见org.redisson.RedissonLock#tryLockInnerAsync<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {    internalLockLeaseTime = unit.toMillis(leaseTime);     return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,              "if (redis.call('exists', KEYS[1]) == 0) then " +                  "redis.call('hset', 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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));}

private void scheduleExpirationRenewal(final long threadId) {        if (expirationRenewalMap.containsKey(getEntryName())) {            return;        }         Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {            @Override            public void run(Timeout timeout) throws Exception {                                RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,                        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +                            "redis.call('pexpire', KEYS[1], ARGV[1]); " +                            "return 1; " +                        "end; " +                        "return 0;",                          Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));                                future.addListener(new FutureListener<Boolean>() {                    @Override                    public void operationComplete(Future<Boolean> future) throws Exception {                        expirationRenewalMap.remove(getEntryName());                        if (!future.isSuccess()) {                            log.error("Can't update lock " + getName() + " expiration", future.cause());                            return;                        }                                                if (future.getNow()) {                            // reschedule itself                            scheduleExpirationRenewal(threadId);                        }                    }                });            }        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);         if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {            task.cancel();        }    }

                RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,                        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +                            "redis.call('pexpire', KEYS[1], ARGV[1]); " +                            "return 1; " +                        "end; " +                        "return 0;",                          Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));

org.redisson.RedissonLock#unlock代码

@Override    public void unlock() {        // 执行解锁Lua脚本，这里传入线程id，是为了保证加锁和解锁是同一个线程，避免误解锁其他线程占有的锁        Boolean opStatus = get(unlockInnerAsync(Thread.currentThread().getId()));        if (opStatus == null) {            throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "                    + id + " thread-id: " + Thread.currentThread().getId());        }        if (opStatus) {            cancelExpirationRenewal();        }    } // 见org.redisson.RedissonLock#unlockInnerAsyncprotected RFuture<Boolean> unlockInnerAsync(long threadId) {    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,            "if (redis.call('exists', KEYS[1]) == 0) then " +                "redis.call('publish', KEYS[2], ARGV[1]); " +                "return 1; " +            "end;" +            "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +                "return nil;" +            "end; " +            "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +            "if (counter > 0) then " +                "redis.call('pexpire', KEYS[1], ARGV[2]); " +                "return 0; " +            "else " +                "redis.call('del', KEYS[1]); " +                "redis.call('publish', KEYS[2], ARGV[1]); " +                "return 1; "+            "end; " +            "return nil;",            Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId)); }