你有真正了解redis吗？

你部署的redis服务器每秒执行的命令数量是多少? 支持的并发连接数量上限是多少？各种数据结构命令的操作性能是多少？

你公司的redis服务器性能你了解吗? 它现在的状态你了解吗？你知道如何去优化吗？

redis的性能测试

redis-benchmark是redis包里面的性能测试程序，这个性能测试能清晰的告诉我们redis服务器在你使用的测试机的情况下各种命令的执行性能。redis-benchmark命令参数解释如下：

//我应该把这些因为都翻译过来。并添加测试效果。redis-benchmark [-h ] [-p ] [-c ] [-n  [-k ]  -h       (redis服务器IP，默认值127.0.0.1)Server hostname (default 127.0.0.1) -p           (redis服务器端口，默认值6379)Server port (default 6379) -s         Server socket (overrides host and port) -c        (测试机与redis服务器创建的连接数量)Number of parallel connections (default 50) -n       (测试机发送的redis命令数量，默认值发送10000)Total number of requests (default 10000) -d           (测试机发送给redis的命令中数据的大小，单位字节，默认值2)Data size of SET/GET value in bytes (default 2) -k        (测试机器与redis服务器连接的模式：1表示一直连接，0表需要发送命令的时候重连，默认值1)1=keep alive 0=reconnect (default 1) -r    Use random keys for SET/GET/INCR, random values for SADD  Using this option the benchmark will get/set keys  in the form mykey_rand:000000012456 instead of constant  keys, the  argument determines the max  number of values for the random number. For instance  if set to 10 only rand:000000000000 - rand:000000000009  range will be allowed. -P         (redis管道命令数量，默认值1，表示不开启管道模式)Pipeline  requests. Default 1 (no pipeline). -q                 (只测试redis存取性能)Quiet. Just show query/sec values --csv              Output in CSV format -l                 (表示一直循环测试redis的性能)Loop. Run the tests forever -t          Only run the comma-separated list of tests. The test                    names are the same as the ones produced as output. -I                 Idle mode. Just open N idle connections and wait.

redis测试命令示例(redisIP ,redisPort 分别是你的redis服务器ip和端口)，redis测试环境：旧的个人pc机，6G内存,双核,性能比较差。

1.redis-benchmark -h redisIP -p redisPort -q -d 100

测试存取大小为100字节数据包的性能，测试结果：

2.redis-benchmark -h redisIP -p redisPort -c 40 -n 8000

40是个redis客户端并发连接，8000个请求，部分结果如下：

3.redis-benchmark -h redisIP -p redisPort set,get -n 8000-q

只测试redis的set和get命令，总共发送8000个命令，部分结果如下：

通过redis-benchmark对redis服务器的性能测试，我们可以知道我的redis服务器的性能瓶颈，在知道业务每秒的并发量和每次业务运行时执行的redis命令个数，就能大概判断我们的redis服务器是否符合业务并发需求。

如果业务的并发需求超过了redis服务器的性能瓶颈，业内有分布式缓存的解决方案，使用成百上千个redis服务器来满足业务并发要求，某博平台的数据存取就是基于大量的redis服务器，redis-benchmark是redis自带的压测工具，虽然能测试redis的性能瓶颈，但不同的业务对redis的并发要求是不同的，具体的业务还需要开发者设计压测方案。

redis实现分布式锁

在分布式系统中，不同系统相互通信协作，分布式锁有着举足轻重的作用，在分布式集群环境中，多进程多线程操作一个无锁的共享资源的时候，就会出现数据的安全问题。

单机系统锁的特性

分布式环境下需要保证共享资源的安全，同单机的多线程环境保证共享资源安全的需求是类似的。"日光之下无新事",分布式锁的体系下的各种机制和单机多线程锁体系下的各种机制原理上是不谋而合的。

单机环境下，可以通过本地锁来控制系统的多个线程或进程运行；在分布式集群环境中，单机本地锁就无法控制集群中的多个服务的并发行为。因此就需要分布式锁来进行控制。

回顾单机环境下锁的一些特性：

互斥

可重入

阻塞等待(超时等待)

隐式锁synchronized

先看jdk自带的同步锁synchronized，java代码如下所示

synchronized下面的三种用法，分别是使用在方法lock(),以及使用类对象和普通对象来同步代码块。注意：synchronized锁是没有超时等待机制的。

经过字节码反编译我们能看到对应的synchronized同步锁的相关字节码指令monitorenter，monitorexit，ACC_SYNCHRONIZED。在同步代码块的代码入口会有monitorenter指令，代码的出口是monitorexit指令，以及方法的标志位flags:ACC_SYNCHRONIZED。到这里介绍的概念就能让读者明白互斥机制的工作原理了，在同一个jvm内存里，类对象是唯一的，data对象也是唯一的，而且他们都是继承了Object，而内存中的每个Object都包含一种数据：Mark Word(Object的组成一部分) 。 那么monitorenter和monitorexit做的是什么工作与Mark word

有什么联系？。在Mark Word中存放了两种数据，分别是锁标识位和线程id，monitorenter指令判断锁标识为空则设置不为空，即获取了锁，同时也会把线程id设置为成功获取锁的线程id(线程在同一个JVM中也是唯一的，唯一性机制在这里不在阐述)。

锁标识位和monitorenter，monitorenter指令实现了互斥机制。

在判断锁已经被获取的情况下，取出锁指向的线程id判断是否为当前的线程id，是则再次成功获取锁，否则锁获取失败，线程进入后续操作。

线程id和monitorenter指令实现了锁的可重入机制。

反编译获取字节码的命令：javap -verbose Synchronized.class

当然synchronized锁机制很复杂，还有自旋锁，轻量锁等很多机制，有兴趣的读者可以深究。

单机锁ReentrantLock

上面我们分析的是jdk内置的synchronized隐式锁，其实jdk还模仿synchronized实现了显示锁Lock，显示锁的核心类是AbstractQueuedSynchronizer。以ReentrantLock为例分析。

作为隐式锁synchronized的模仿者Lock，实现的互斥与可重入机制的设计思路是和隐式锁一样的。

state字段和lock方法实现了互斥机制

execlusieOwnerThread字段和lock方法实现了可重入机制

对比synchronized，锁标志位和monitorenter，monitorexit实现互斥；Mark Word中的线程id和monitorenter，monitorexit实现可重入。

ReentrantLock使用unsafe类的原子方法compareAndSwapInt设置state的值，方法逻辑：如果state值为0，则将state的值设置为1，线程成功获取锁；如果state值不是0，则线程获取锁失败。这样就实现了互斥机制。

protected final boolean compareAndSetState(int expect, int update) {        // See below for intrinsics setup to support this  return unsafe.compareAndSwapInt(this, stateOffset, expect, update);    }

ReentrantLock类的字段exclusiveOwnerThread会在线程成功获取锁的时候设置该线程的引用，标识当前获取锁的线程。代码逻辑：当线程获取锁失败，执行的方法链 acquire->tryAcquire->

nonfairTryAcquire，nonfairTryAcquire方法先尝试通过互斥机制去竞争锁，失败后通过可重入机制判断execlusieOwnerThread线程引用是否为当前线程，是则成功获取锁并将state增1操作，失败则获取锁失败。到此实现了锁的可重入机制

final boolean nonfairTryAcquire(int acquires) {    final Thread current = Thread.currentThread();    int c = getState();    //重新进行尝试获取锁，互斥机制获取锁    if (c == 0) {        if (compareAndSetState(0, acquires)) {            setExclusiveOwnerThread(current);            return true;        }    }  //获取锁失败则判断锁的获取的锁线程引用是否为当前的线程，如果是则获取，可重入逻辑    else if (current == getExclusiveOwnerThread()) {        int nextc = c + acquires;        if (nextc < 0) // overflow            throw new Error("Maximum lock count exceeded");        setState(nextc);        return true;    }    return false;}

ReentrantLock类有锁获取的超时机制，提供的方法tryLock(long timeout, TimeUnit unit)，这个方法的核心操作是通过unsafe.park(false, nanos)使得线程挂起一段nanos时间，锁的等待机制的核心代码逻辑是：传入的等待时间nanos/1000如果大于1毫秒，则调用park挂起线程nanos/1000毫秒，然后唤起线程循环竞争锁，执行nanos%1000纳秒并返回失败，如果nanos/1000小于1毫秒，则线程直接循环竞争锁，执行nanos%1000纳秒并返回，最终线程等待了nanos纳秒时间。这就是锁的获取超时机制

核心代码如下

for (;;) {    final Node p = node.predecessor();    if (p == head && tryAcquire(arg)) {        setHead(node);        p.next = null; // help GC        failed = false;        return true;    }    if (nanosTimeout <= 0)        return false;    if (shouldParkAfterFailedAcquire(p, node) &&        nanosTimeout > spinForTimeoutThreshold)        LockSupport.parkNanos(this, nanosTimeout);    long now = System.nanoTime();    nanosTimeout -= now - lastTime;    lastTime = now;    if (Thread.interrupted())        throw new InterruptedException();}

到这里我们可以总结下锁的三种特性：互斥机制中的锁状态的变更需要通过原子操作来实现；可重入机制需要锁的一个线程字段属性来标记成功获取到锁的线程，每次通过互斥机制去竞争锁，失败后就判断是否能重入；超时机制就是循环里面每次线程挂起一段时间，再重新去竞争锁，成功则返回，失败则继续挂起线程循环继续，直到总共耗的时间超过给定时间线程就获取锁失败并返回。

redis分布式锁

那么如果我们想要实现分布式锁，且让分别是锁也拥有互斥和可重入的机制，如何基于redis进行设计？互斥 = 状态数据 + 原子操作 ，可重入 = 线程标记 + 可重入判断。

对redis了解的同学应该知道redis的命令都是串行执行的，因此redis就天生支持原子操作，那我们就能借助redis来实现多个进程之间的互斥机制，即分布式锁，对于可重入特性，可以选择适合的redis数据结构来存储进程标记，这种方式就能实现了可重入机制。