Redis

• Redis
  • 一、Redis概述
  • 二、Redis使用方式
    • 1. 启动和关闭
    • 2. 常用操作命令
  • 三、Redis常见数据类型
    • 1. String
    • 2. List
    • 3. Set
    • 4. Hash
    • 5. Zset(Sorted Set)
  • 四、Redis持久化机制
    • 1. RDB(Redis Database)
    • 2. AOF(Append only File)
  • 五、事务控制
  • 六、发布/订阅模式
  • 七、主从复制(Redis集群的策略)
    • 1. 主从复制的相关特性
    • 2. 主从复制的实现原理
    • 3. 主从复制的哨兵模式(Sentinel)
  • 八、Jedis
    • 1. Jedis连接问题
    • 2. Jedis常用API
    • 3. 事务控制
    • 4. JedisPool
  • 九、分布式锁
    • 1. 单进程多线程并发
    • 2. 多进程多线程并发

一、Redis概述

• cache：位于项目中Dao层与数据库之间，主要用于数据访问量大时使用缓存技术来缓解数据库的压力

  一些频繁需要访问的数据放在关系型数据库中，每次查询开销很大，而放在cache中可以高效地被访问

• Redis是运行在内存上的NoSQL(Not Only SQL)数据库，也是一种cache

  NoSQL与传统数据库相比无需为要存储的数据建立字段，没有表的概念所以可以随时存储自定义的数据格式

  其他常见的NoSQL数据库：Memcache(与Redis作用类似，也是运行在内存上)、MongoDB(运行在硬盘上)

• 分布式数据库具备C(强一致性)、A(高可用性)、P(分区容错性)

  Consistency：所有节点在同一时间的数据完全一致

  Availability：服务一直可用

  Partition tolerance：某节点或网络分区故障时，仍然能够对外提供满足一致性或可用性的服务

  对于分布式数据库来说分区不可避免(CA表示非分布式但同时满足一致性和可用性，这种扩展性不强)，

  分区容错性的分布式数据库有两种情况：CP和AP(一致性和可用性一般不可同时满足)

  CP：满足一致性，可用性不强

  AP：满足可用性，一致性不强

二、Redis使用方式

1. 启动和关闭

• 加载配置文件(配置后台运行)的启动方式

  /usr/local/bin/redis-server /opt/redis-5.0.4/redis.conf
  

• 查看运行状态：检测6379端口是否在监听

  netstat -lntp | grep 6379
  

• 查看运行状态：查看后台进程是否存在

  ps -ef|grep redis
  

• 单实例关闭(/usr/local/bin/下操作)

  redis-cli shutdown
  

• 多实例关闭(/usr/local/bin/下操作)

  redis-cli -p 6379 shutdown
  

• 进入客户端(/usr/local/bin/下操作)

  redis-cli
  

• 进入客户端后关闭Redis

  shutdown
  

2. 常用操作命令

• 测试连接(正常返回PONG)

  ping
  

• 切换数据库(共16个库，索引范围0-15，默认是0不显示库号)

  select xxx
  

• 查看当前库的所有键数

  dbsize
  

• 查看当前库的键(支持模糊查询：*一次占用多位，?一次占用一位，[xxx]指定与xxx中的每个字符都匹配一次并且占用一位)

  查看所有的键

  keys *
  

  keys xxx*
  

  keys *xxx
  

  查看包含xxx的键

  keys *xxx*
  

• 清空当前库

  flushdb
  

• 清空所有(16)库

  flushall
  

• 判断键是否存在

  exists xxx
  

• 移动指定的键到指定的库中

  move 键名 库索引号
  

• 查看过期时间(-1永不过期，-2已过期)

  ttl 键名
  

• 设置过期时间(过期后相当于从当前库中自动移除了指定的键，执行ttl返回-2)

  expire 键名 时间(s)
  

• 查看指定键的数据类型

  type 键名
  

三、Redis常见数据类型

1. String

• 特点：不存在重复的键，键对应的值是一个字符串，可向其中追加其他字符串

• 相关命令

  • set：添加指定的键和值，示例set 键名 值

    添加已存在的键时，会替换原来的键值

  • get：查询指定的键对应的值，示例get 键名

  • del：删除指定的键，示例del 键名

  • append：向指定键对应的值追加字符串，示例append 键名 xxx

  • strlen：返回指定键对应值的长度，示例strlen 键名

  • incr：键值自动增长1(值必须为数字)，示例incr 键名

  • decr：键值自动减少1(值必须为数字)，示例decr 键名

  • incrby：键值按指定值增加(值必须为数字)，示例incrby 键名 值

  • decrby：键值按指定值减少(值必须为数字)，示例decrby 键名 值

  • getrange：查询指定键的指定索引范围的值，返回字符串，示例getrange 键名 索引开始 索引结束

    索引范围是0到-1时，查询全部的键值

  • setrange：替换指定键的指定索引范围的值(从指定索引开始替换，替换长度为指定字符串长度)，

    示例setrange 键名 索引 xxx

  • setex：添加键时同时设定过期时间，示例setex 键名 时间(s) 值

  • setnx：添加键时判断该键是否已经存在(只添加不存在的键，防止替换已存在的键值)，

    示例setnx 键名 值

    添加已存在的键时，会返回0，表示添加失败

  • mset：一次添加多组键值，示例mset 键名1 值1 键名2 值2 键名3 值3...

  • mget：一次查询指定的多个键对应的值，示例mget 键名1 键名2 键名3...

  • msetnx：一次添加多组键值，若有已存在的键则返回0，示例msetnx 键名1 值1 键名2 值2 键名3 值3...

  • getset：先查询当前的键值(返回当前键值)，再设置值为指定值，示例getset 键名 值

2. List

• 特点：不存在重复的键，键对应的值是元素可重复集合

• 相关命令

  • lpush：从左开始压栈，示例：lpush list01 1 2 3 4 5

  • rpush：从右开始压栈，示例：rpush list02 1 2 3 4 5

  • lrange：从栈顶开始向下读取List，示例：lrange list02 0 -1

    0表示开始，-1表示结尾

  • lpop：从栈顶移除一个元素，示例lpop list02

  • rpop：从栈底移除一个元素，示例rpop list02

  • lindex：从栈顶开始读取指定索引的元素，示例lindex list01 2

    索引从0开始

  • llen：返回集合的总元素数，示例llen list01

  • lrem：从集合中移除指定个数的指定元素，示例lrem list01 个数 元素

  • ltrim：截取集合中指定索引的元素其余元素被移除，示例ltrim list01 起始索引 结束索引

  • rpoplpush：将集合1从栈底移除一个元素后集合2从栈顶压入这个元素，示例rpoplpush list01 list02

  • lset：修改指定索引的元素，示例lset list01 索引 元素

  • linsert：在指定元素的前或后插入元素，示例 linsert list01 before/after 集合中的元素 插入元素

    从左边进入

• List类似链表，头尾操作效率高，中间操作效率低

3. Set

• 特点：不存在重复的键，键对应的值是元素不重复的集合

• 相关命令

  • sadd：添加不重复的集合元素，示例sadd set01 1 2 2 3 3 3

    sadd自动去重后添加

  • smembers：查看集合中的元素，示例smembers set01

  • sismemeber：查看指定的元素是否为集合中的元素，示例sismember set01 元素

  • scard：获取集合中的元素个数，示例scard set01

  • srem：移除指定元素，示例srem set01 元素

  • srandmember：获取指定个数的随机集合元素，示例srandmember set01 元素个数

  • spop：随机移除集合中的一个元素，示例spop set01

  • smove：将元素从集合1移动到集合2中，示例smove set01 set02 元素

  • sinter：返回两个集合中元素的交集，示例sinter set01 set02

  • sunion：返回两个集合中元素的并集，示例sunion set01 set02

  • sdiff：返回两个集合中元素的差集，示例sdiff set01 set02

    sdiff set01 set02：返回在set01中存在，在set02中不存在的元素

4. Hash

• 特点：不存在重复的键，键对应的值是一个/多个键值对集合，键值对的键是不重复的，键值对的值可重复

• 相关命令

  • hset：为一个Hash key添加一个指定的key和value，示例hset hash01 key value

  • hget：返回Hash key中指定key的value，示例hget hash01 key

  • hmget：返回Hash key中指定多个key的多个value，示例hmget hash01 key1 key2...

    重复添加已存在的键的键值对，会将其值进行替换

  • hmset：为一个Hash key添加多个指定的key和value，示例hmset hash02 key1 value1 key2 value2 key3 value3...

  • hgetall：返回Hash key的所有key和value，示例hgetall hash02

  • hdel：删除Hash key中指定key和对应的value，示例hdel hash02 key

  • hlen：返回指定Hash key的属性个数，示例hlen hash02

  • hexists：返回指定Hash key中是否存在某个key，示例hexists hash02 key

  • hkeys：返回指定Hash key中的所有key，示例hkeys hash02

  • hvals：返回指定Hash key中的所有value，示例hvals hash02

  • hincrby：增加指定的Hash key中某个key的value，示例hincrby hash02 key 增加值

  • hincrbyfloat：增加指定的Hash key中某个key的value(小数)，示例hincrbyfloat hash02 key 增加值

  • hsetnx：为一个Hash key添加一个指定的key和value若key已存在则添加失败，示例hsetnx hash02 key value

5. Zset(Sorted Set)

• 特点：不存在重复的键，键对应值是一个/多个键值对集合(会根据键的数字自动排序，键值对中可以存在重复的键，值是不重复的)

  键值对的键一般也是内容为数字的字符串

• 相关命令

  • zadd：为指定的键添加一个/多个值，示例zadd 键名 数字1 值1 数字2 值2...

  • zrange：查询指定键对应的所有键值对的值，示例zset 键名 0 -1 (withscores)

    添加withscores后同时会返回键值对的键(数字)

  • zrangebyscore：模糊查询指定键对应的键值对的值(按键值对的键查询)，示例zrangebyscore 键名 数字1 数字2 (limit x y)

    默认的在数字1(min)和2(max)上是闭区间，数字前添加(表示不包含指定的数字

    添加limit x y，表示跳过查询结果中的前x个并取y个

  • zrem：移除指定键对应的键值对(按值删)，示例zrem 键名 xxx

  • zcard：查询指定键对应的键值对的个数，示例zcard 键名

  • zcount：查询指定键对应的某个范围下(键值对中的键)键值对的个数，示例zcount 键名 数字1 数字2

  • zrank：查询指定键对应的某个键值对索引(按键值对中的值进行查询)，示例zrank 键名 值

    键值对索引从0开始，索引不等于键值对的键(数字)

  • zscore：查询指定键对应的某个键值对的键(按键值对中的值进行查询)，示例zscore 键名 值

  • zrevrank：同zrank，但返回结果是以逆序查找，示例zrevrank 键名 值

  • zrevrange：同zrange，但返回结果是以逆序查找，示例zrevrange 键名 0 -1 (withscores)

  • zrevrangebyscore：同zrangebyscore，但返回结果是以逆序查找，示例zrevrangebyscore 键名 数字1 数字2 (limit x y)

    默认的在数字1(max)和2(min)上是闭区间，数字前添加(表示不包含指定的数字

四、Redis持久化机制

1. RDB(Redis Database)

• 自动备份

  • 在指定的时间间隔内，将内存中数据的快照写入硬盘中(默认保存在bin目录下的dump.rdb中)

    可以修改redis.config来改变默认配置，设置自动备份的频率或者直接关闭持久化

  • 在Redis客户端中执行shutdown后会自动备份

• 手动备份

  • 在Redis客户端中执行save后执行手动备份

2. AOF(Append only File)

• 将Redis执行过程中的写指令以追加的方式全部记录在appendonly.aof中

• Redis在启动之初会读取该文件，从头到尾执行一遍实现数据恢复

  flushdb后再重启Redis，不影响aof文件中的内容

  aof文件不建议手动修改，Redis在启动时若读取aof错误将导致Redis无法正常启动

  aof错误解决方案：手动执行命令恢复aof

• RDB和AOF共存时，Redis优先载入aof文件

• Redis持久化机制中RDB和AOF各有利弊，开发中采用主从复制来实现持久化

五、事务控制

• 事务控制方式(在Redis客户端中操作)

  • 开启事务(multi)，加入队列(如添加键后自动加入，返回QUEUED)，一起执行(exec)，一起加入队列将执行操作成功

  • 开启事务(multi)，执行后放弃(discard)之前的操作，所有的键和对应的值都恢复到原来

  • 开启事务(multi)，加入队列报错(如输入不存在的指令)，一起执行(exec)，提示EXECABORT，所有的键和对应的值都恢复到原来

  • 开启事务(multi)，加入队列成功但实际操作报错(如incr一个不是数字内容的字符串)，一起执行(exec)，除了实际操作报错，其他加入队列将执行操作成功

    该操作类似编译成功，运行时报错

  • 开启事务前对键加入监控watch 键名，开启事务(multi)，加入队列(操作监控的键值)，同时开启另一个线程操作监控的键值，这时在原来的线程中执行(exec)，返回(nil)表示本次事务被打断导致失败

六、发布/订阅模式

• 发布/订阅模式是Redis的一种使用场景，即实现简单消息队列

• 进程间的一种消息通信模式：发送者(pub)发送消息，订阅者(sub)接收消息。例如：微信订阅号

• 实现方式：在Redis客户端订阅subscribe 频道名1 频道名2 ...，在另一个线程的Redis客户端中发布publish 频道名 消息内容，

  此时在订阅端会收到发布的消息

七、主从复制(Redis集群的策略)

1. 主从复制的相关特性

• 配从不配主：在从机端配置该机器的主机，主机端默认成为该从机的主机

  集群测试环境：配置三台装了Redis的服务器

  slaveof 主机IP地址 6379
  

  查看主从复制信息

  info replication
  

• 读写分离：从机只负责读取数据，无权写入数据；在主机写入数据时，所有从机将会同步数据

  从机将会同步主机成为主机之前和之后的所有数据

• 一个主机理论上可以多个从机，但这样主机负担很大，可以使用继承传递性来解决，减轻主机的负担

  主机1->主机2(主机1的从机)->主机3(主机2的从机)...

  这种情况下还是保持读写分离：只有主机1负责写入数据，后面的主机只有权读取

• 谋权篡位：1个主机，2个从机，当1个主机挂掉了，其余从机主机仍然是slave并显示它们的master已离线

  可以手动将一台slave设定为主机

  slaveof no one
  

  另一台根据需要设定为新主机的从机

  slaveof 主机IP地址 6379
  

  经过以上两步，原主机上线后会成为新的master并无其他的slave

2. 主从复制的实现原理

• 复制的主要流程

  1. 从服务器连接主机服务器，并发送同步请求
  2. 主机服务器启动后台的存盘进程，同时会收集所有写的命令集
  3. 主机服务器向从服务器发送快照，从服务器载入快照
  4. 主机服务器再次向从服务器发送缓存写命令，从服务器执行命令

• 相关概念

  • 全量复制：slave初始化阶段，这时slave需要将master上的所有数据都复制一份slave接收到数据文件后，存盘，并加载到内存中

    涉及复制的主要流程中的1，2，3

    只要是slave重新连接master，一次性（全量复制）同步将自动执行

  • 增量复制：slave初始化阶段后，开始正常工作时主服务器发生的写操作同步到从服务器的过程

    涉及复制的主要流程中的4

  • Redis主从同步策略：主从刚刚连接的时候，进行全量同步；全同步结束后，进行增量同步

    如果有需要，slave在任何时候都可以发起全量同步；

    Redis策略是，无论如何，首先会尝试进行增量同步，如不成功，要求从机进行全量同步

3. 主从复制的哨兵模式(Sentinel)

• 哨兵模式(Sentinel)：

  由一个或多个Sentinel实例组成的Sentinel系统可以监视任意多个主服务器，以及所有从服务器，并在被监视的主服务器进入下线状态时，自动将下线主服务器属下的某个从服务器升级为新的主服务器，然后由新的主服务器代替已下线的主服务器继续处理命令请求

• Sentinel是Redis的高可用性解决方案，可以看作是自动版的谋权篡位

  设定一个哨兵监控master当master关闭后，自动从slave中分配新的master，维持了主从复制的读写分离特性，保证了系统的高可用性

• 实现方式

  1. 在/usr/local/bin下创建sentinel.conf，同时添加以下配置（每台服务器上的Redis分别给自身投一票）

     集群测试环境：预先设定好128为master，129和130为slave

     sentinel monitor 被监控主机名（自定义） ip port 票数
     

     

  2. 重启三台服务器的Redis，并复制128，129，130窗口，在新的复制中执行以下命令启动Sentinel

     redis-sentinel sentinel.conf
     

  3. 测试将原来的master128服务器Redis下线，会自动发起投票，在slave129和130中选出新的master

• Redis Sentinal的缺点

  • 数据同步较慢

    由于所有的写操作都是在master上完成的，然后再同步到slave上，所以两台机器之间通信会有延迟；

    当系统很繁忙的时候，延迟问题会加重；

    slave机器数量增加，问题也会加重

八、Jedis

1. Jedis连接问题

• 解决方式：

  1. 关闭服务器

     redis-cli shutdown
     

  2. 关闭防火墙

     systemctl stop firewalld.service
     

  3. 重启服务器

     redis-server /opt/redis-5.0.4/redis.conf
     

2. Jedis常用API

• String

  @Test
  public void testString() {
      Jedis jedis = new Jedis("192.168.197.128", 6379);
      // 添加指定的键和值
      jedis.set("k1", "v1");
      jedis.set("k2", "v2");
      jedis.set("k3", "v3");
      jedis.set("k33", "v3");
      // 查看所有的键
      Set<String> keys = jedis.keys("*");
      // 遍历 Set 集合 ，每次查询指定的键对应的值
      for (String key : keys) {
          System.out.println(key + "->" + jedis.get(key));
      }
      // 查看指定键是否存在
      Boolean k2 = jedis.exists("k2");
      System.out.println("k2 exists = " + k2);
      // 查看指定键的过期时间
      System.out.println(jedis.ttl("k1"));
      // 清空当前数据库
      jedis.flushDB();
      // 一次添加多组键值
      jedis.mset("k4", "v4", "k5", "v5");
      // 一次查询指定的多个键对应的值
      List<String> mGet = jedis.mget("k4", "k5");
      System.out.println(mGet);
  }
  /*
  k3->v3
  k4->v4
  k5->v5
  k33->v3
  k1->v1
  k2->v2
  k2 exists = true
  -1
  [v4, v5]
  */
  

• List

  @Test
  public void testList() {
      Jedis jedis = new Jedis("192.168.197.128", 6379);
      // 清空当前数据库
      jedis.flushDB();
      // 从左开始压栈
      jedis.lpush("list01", "l1", "l2", "l3", "l4", "l5");
      // 从栈顶开始向下读取 List 集合
      List<String> list01 = jedis.lrange("list01", 0, -1);
      System.out.println(list01);
  }
  /*
  [l5, l4, l3, l2, l1]
  */
  

• Set

  @Test
  public void testSet() {
      Jedis jedis = new Jedis("192.168.197.128", 6379);
      // 清空当前数据库
      jedis.flushDB();
      // 添加不重复的键值
      jedis.sadd("order", "o1");
      jedis.sadd("order", "o2");
      jedis.sadd("order", "o3");
      // 查看集合中的元素
      Set<String> order = jedis.smembers("order");
      System.out.println(order);
      // 移除指定的元素后查看当前键值的元素个数
      jedis.srem("order", "o2");
      System.out.println(jedis.smembers("order").size());
  }
  /*
  [o3, o2, o1]
  2
  */
  

• Hash

  @Test
  public void testHash() {
      Jedis jedis = new Jedis("192.168.197.128", 6379);
      // 清空当前数据库
      jedis.flushDB();
      // 为一个 Hash key 添加一个指定的 key 和 value
      jedis.hset("hash01", "username", "Jeff");
      // 返回 Hash key 中指定 key 的 value
      String hGet = jedis.hget("hash01", "username");
      System.out.println(hGet);
      // 准备一个 map
      HashMap<String, String> map = new HashMap<>();
      map.put("gender", "male");
      map.put("address", "Beijing");
      map.put("phoneNumber", "123456");
      // 通过传入 map 为一个 Hash key 添加多个指定的 key 和 value
      jedis.hmset("hash01", map);
      // 返回 Hash key 的所有 key 和 value
      Map<String, String> hash01 = jedis.hgetAll("hash01");
      System.out.println(hash01);
  }
  /*
  Jeff
  {gender=male, username=Jeff, address=Beijing, phoneNumber=123456}
  */
  

• Zset

  @Test
  public void testZset() {
      Jedis jedis = new Jedis("192.168.197.128", 6379);
      // 清空当前数据库
      jedis.flushDB();
      // 为指定的键添加一个/多个值
      jedis.zadd("zset01", 60d, "zs1");
      jedis.zadd("zset01", 60d, "zs2");
      jedis.zadd("zset01", 60d, "zs3");
      jedis.zadd("zset01", 70d, "zs4");
      jedis.zadd("zset01", 80d, "zs5");
      jedis.zadd("zset01", 90d, "zs6");
      jedis.zadd("zset01", 50d, "zs7");
      // 查询指定键对应的所有键值对的值
      Set<String> zSet01 = jedis.zrange("zset01", 0, -1);
      System.out.println(zSet01);
  }
  /*
  [zs7, zs1, zs2, zs3, zs4, zs5, zs6]
  */
  

3. 事务控制

• 事务控制前，建议先将指定的键监控起来

  public class JedisTransaction {
      @Test
      public void resetBalanceAndExpense() {
          Jedis jedis = new Jedis("192.168.197.128", 6379);
          // 清空当前数据库
          jedis.flushDB();
          // 设定余额和支出的初始值
          jedis.set("balance", "100");
          jedis.set("expense", "0");
      }
      
      @Test
      public void testTransaction() {
          Jedis jedis = new Jedis("192.168.197.128", 6379);
          // 获取当前的余额
          int balance = Integer.parseInt(jedis.get("balance"));
          // 指定每次消费金额
          int expense = 10;
          // 将余额监控起来
          jedis.watch("balance");
  
          if (balance < expense) {
              // 解除监控
              jedis.unwatch();
              System.out.println("balance is not enough");
          } else {
              // 开启事务
              Transaction transaction = jedis.multi();
              // 余额减少
              transaction.decrBy("balance", expense);
              // 累计消费增加
              transaction.incrBy("expense", expense);
              transaction.exec();
              System.out.println("balance:" + jedis.get("balance"));
              System.out.println("expense:" + jedis.get("expense"));
          }
      }
  }
  

4. JedisPool

• 单例模式优化Jedis连接池

  • 双层检查锁定(Double check locked)

    这个例子在单线程环境可以正常运行，但是在多线程环境就有可能会抛出空指针异常。为了防止这种情况，我们需要在该方法上使用 synchronized。这样该方法在多线程环境就是安全的，但是这么做就会导致每次方法调用都需要获取与释放锁，开销很大

    private JedisPool jedisPool;
    
    public JedisPool getJedisPool() {
        if (jedisPool == null) {
            jedisPool = new JedisPool();
        }
        return jedisPool;
    }
    

    分析可以得知只有在初始化的变量需要真正加锁，一旦初始化之后，直接返回对象即可。

    所以可以将该方法改造以下的样子

    这个方法首先判断变量是否被初始化，没有被初始化，再去获取锁。获取锁之后，再次判断变量是否被初始化。第二次判断目的在于有可能其他线程获取过锁，已经初始化该变量。第二次检查还未通过，才会真正初始化变量。

    这个方法检查判定两次，并使用锁，所以称为双重检查锁定模式

    private JedisPool jedisPool;
    
    public JedisPool getJedisPool() {
        if (jedisPool == null) {
            synchronized (this) {
                if (jedisPool == null) {
                    jedisPool = new JedisPool();
                }
            }
        }
        return jedisPool;
    }
    

  • 双层检查锁定的问题

    • 指令重排在单线程下可用，在多线程下会发生异常

      创建一个对象(JedisPool jedisPool = new JedisPool())的过程：

      1. 分配对象内存

      2. 调用构造器方法，执行初始化

      3. 将对象引用赋值给变量

      JVM运行时，指令1会先执行，指令2，3会在不影响整体结果的情况下，进行重新排序来提高程序的执行性能

      根据上一个双层检查锁定的示例代码，假设有两个线程，线程1先执行了指令1和3完成后此时线程2判断到对象不为null就会访问该对象但是该对象还未被初始化，触发异常

    • 使用 volatile可以禁止指令的重排序，保证多线程环境内的系统安全

      volatile关键字的作用：

      1. 保证可见性：使用 volatile定义的变量，将会保证对所有线程的可见性。
      2. 禁止指令重排序优化：由于 volatile禁止对象创建时指令之间重排序，所以其他线程不会访问到一个未初始化的对象，从而保证安全性。

  • 单例模式优化Jedis连接池的工具类

    定义工具类

    public class JedisPoolUtil {
        private JedisPoolUtil() {
        }
    
        private volatile static JedisPool jedisPool = null;
        private volatile static Jedis jedis = null;
    
        // 返回一个连接池对象
        private static JedisPool getInstance() {
            // 双层检查锁定
            if (jedisPool == null) {
                synchronized (JedisPoolUtil.class) {
                    if (jedisPool == null) {
                        // 设定连接池参数
                        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
                        jedisPoolConfig.setMaxTotal(1000);
                        jedisPoolConfig.setMaxIdle(30);
                        jedisPoolConfig.setMaxWaitMillis(60*1000);
                        jedisPoolConfig.setTestOnBorrow(true);
                        // 创建连接池对象
                        jedisPool = new JedisPool(jedisPoolConfig, "192.168.197.128", 6379);
                    }
                }
            }
            return jedisPool;
        }
    
        // 返回 Jedis 对象
        public static Jedis getJedis() {
            if (jedis == null) {
                jedis = getInstance().getResource();
            }
            return jedis;
        }
    }
    

    测试工具类

    public class TestJedisPool {
        @Test
        public void testGetJedis() {
            Jedis jedis01 = JedisPoolUtil.getJedis();
            Jedis jedis02 = JedisPoolUtil.getJedis();
            System.out.println(jedis01 == jedis02);
        }
    }
    /*
    true
    */
    

九、分布式锁

1. 单进程多线程并发

• 使用Synchronized锁解决

  使用Jmeter进行并发请求测试，1秒100个线程发送请求

  该方案仅适用于单进程(一个Tomcat)的并发问题，如果是分布式环境，多个进程并发，这种方案就会失效

  @Controller
  public class TestKill {
  
      @Autowired
      private StringRedisTemplate stringRedisTemplate;
  
      @RequestMapping("kill")
      @ResponseBody
      public synchronized String kill () {
          // 从 redis 中获取库存
          int phoneCount = Integer.parseInt(stringRedisTemplate.opsForValue().get("phone"));
          // 判断数量是否够秒杀
          if (phoneCount > 0) {
              phoneCount--;
              // 库存减少后，再将库存的值保存回 redis
              stringRedisTemplate.opsForValue().set("phone", phoneCount+"");
              System.out.println("库存-1，剩余：" + phoneCount);
          } else {
              System.out.println("库存不足");
          }
          return "over!";
      }
  }
  

2. 多进程多线程并发

• 分布式锁实现思路：因为Redis是单线程的，所以命令也就具备原子性，使用setnx命令实现锁，保存k-v

  Redisson是用于在Java程序中操作Redis的库，在java.util中常用接口的基础上，提供了一系列具有分布式特性的工具类

• 使用Redis分布式锁解决多进程多线程并发(Redisson)

  运行环境：使用nginx作为代理服务器接收请求转发给两个Tomcat，每个Tomcat部署以下程序，使用JMeter发送多线程请求

  @Controller
  public class TestKill {
  
      @Autowired
      private Redisson redisson;
      @Autowired
      private StringRedisTemplate stringRedisTemplate;
  
      @RequestMapping("kill")
      @ResponseBody
      public synchronized String kill () {
          // 定义商品 id
          String productKey = "phone01";
          // 通过 redisson 获取锁
          // 底层源码集成了 setnx ，过期时间等操作
          RLock lock = redisson.getLock(productKey);
          // 上锁
          lock.lock(30, TimeUnit.SECONDS);
  
          try {
              // 从 redis 中获取库存
              int phoneCount = Integer.parseInt(stringRedisTemplate.opsForValue().get("phone"));
              // 判断数量是否够秒杀
              if (phoneCount > 0) {
                  phoneCount--;
                  // 库存减少后，再将库存的值保存回 redis
                  stringRedisTemplate.opsForValue().set("phone", phoneCount+"");
                  System.out.println("库存-1，剩余：" + phoneCount);
              } else {
                  System.out.println("库存不足");
              }
          } catch (NumberFormatException e) {
              e.printStackTrace();
          } finally {
              // 释放锁
              lock.unlock();
          }
          return "over!";
      }
  
      @Bean
      public Redisson redisson() {
          Config config = new Config();
          // 使用单个redis服务器
          config.useSingleServer().setAddress("redis://192.168.197.128:6379").setDatabase(0);
          // 使用集群redis
          // config.useClusterServers().setScanInterval(2000).addNodeAddress("", "", "")
          return (Redisson)Redisson.create(config);
      }
  }