第三部分redis高级

一、击穿：

原因1：缓存失效，大量用户访问相同的热点数据在缓存中没有，造成大量用户去访问数据库，数据库承受的压力过大，容易导致数据库崩溃。解决办法是使用分布式锁，大量的用户中只有一个用户能拿到这把锁，让获得锁的请求去访问数据库，其他的请求阻塞或者自旋，直到获得锁请求的用户从数据库取到数据，将数据放到缓存中。
原因2：空值数据，在数据库中更新了的数据，这个新数据在缓存中写入该值的空值key,比如
k-v ： user_id null

二、redis缓存和内存数据库的区别

redis做缓存，redis后边是有真实的=数据库的，用redis是为了降低对数据库的访问，减小对数据库的压力。
应用场景：表的主键，表的存放方式是水平切分，不同节点存的是不同的数据，不同节点怎么保证主键不重复呢？就可以使用redis 生产主键，同时把数据写到数据库中，redis的单线程保证数据肯定是不重复的。
内存数据库：后边是没有其他数据库的，这里边存储的是一些高频被访问的数据，或者生命周期不长的数据(临时数据)。

三、5大数据类型

string，set ，zset , list，hash
主要是读和写的命令：
以key为主体的通用命令：dbsize/expire/ttl/

1.string:一对一

读：get/mget
写：set/mset/setnx(分布式锁:判断加写入是原子的，判断加写入的时候不会被其他人操作)/setex、incr/incrby(计数的)

2.set ： 一对多，无序不重复，经常用于去重

读：smembers(查所有)/sismember(是否存在)/
写：sadd/sinter/sunion/sdiff

3.list:一对多，常用于队列、双端双链表

区别于kafka消息队列的特点：redis很小，轻量级
读：lrange(读范围，全读：0 -1)redis中的范围都是包括开始位置和结束位置的。
写：rpop,lpop,消息一旦弹出数据消失

4.hash:一对多，复杂类型

使用场景1：复杂对象，对象中的字段经常有单独修改变化的场景（key field value）
使用场景2：存储集合对象。(班级 学号 姓名)这是按班级查的
写：hset/hmset/incrby
读：hget/hmget/hgetall

5.zset:一对多，有序不重复

用score进行评分排序。
比如：按照时间戳排序，按照排行榜排序
读：zrange/zrevrange/zrank
写：zadd/zincrby(减法用负的就行)

四、单线程 io多路复用

(io多路复用其实是Linux的特点)redis只是使用了，原理是有一个管理器在不同的请求之间进行切换，一个请求使用cpu空闲的时候（cpu对之前请求的计算需求已经完成，后续可能进行网络传输了，就会切换到其他 的请求），会切换到其他的请求进行处理。

io多路复用使用场景

请求非常多，计算任务比较小，最大化单线程应用面对多请求的利用率，让cpu在不同的请求中切换。
补充：多线程结构复杂，利用率有限（主要指的是对cpu的使用率）

五、持久化

redis数据在磁盘和内存中都有的话 ，以内存数据为准。磁盘数据存放的位置默认为启动redis时命令行所在的目录下。

redis提供了两种不同的持久化方式：

①rdb

在指定的时间间隔内将内存中的数据集快照写入磁盘，也就是Snapshot快照，它恢复时是将快照文件直接读到内存里

rdb原理：

Redis会单独创建（fork）一个子进程来进行持久化，会先将数据写入到一个临时文件中，待持久化过程都结束了，再用这个临时文件替换上次持久化好的文件。整个过程中，主进程是不进行任何IO操作的，这就确保了极高的性能如果需要进行大规模数据的恢复，且对于数据恢复的完整性不是非常敏感，那RDB方式要比AOF方式更加的高效。RDB的缺点是最后一次持久化后的数据可能丢失。（是周期性的备份，可能会丢失数据，因为是替换上次的文件）。

关于fork：

在Linux程序中，fork()会产生一个和父进程完全相同的子进程，但子进程在此后多会exec系统调用，出于效率考虑，Linux中引入了“写时复制技术”，一般情况父进程和子进程会共用同一段物理内存，只有进程空间的各段的内容要发生变化时，才会将父进程的内容复制一份给子进程。

配置文件：

每条配置之间是and的关系，三条之间是or的关系

意思：第一条为例：900秒之后至少有一条数据被改变才会触发备份
save 900 1
save 300 10
save 60 10000

自己指定磁盘文件的位置：

dir /opt/module/redis-3.2.5
在客户端中查看配置文件的内容：
config get dir

其他配置

配置文件：

stop-writes-on-bgsave-error yes:

当磁盘满的时候或者是磁盘备份出现问题了，往内存的写要不要停呢？设置这个参数是内存的写也停掉，这样磁盘不能备份的时候，内存也不再写，不会丢失数据，先去解决问题吧！

rdbcompression yes:

进行rdb保存时开启压缩，减少备份对磁盘的占用

rdbchecksum yes:校验

校验备份出来的数据是对的还是错的，万一错的会重新备份或报错，磁盘备份一定要恢复演练，如果数据库倒了能从备份恢复，能保证备份的数据是完整的。
在存储快照后，还可以让Redis使用CRC64算法来进行数据校验，但是这样做会增加大约10%的性能消耗，如果希望获取到最大的性能提升，可以关闭此功能

会触发存盘的时机：bgsave后台独立进程备份

flushall:会触发磁盘，但是没有数据的（最小就是76kb），是执行后存盘，也就是先flushall再存盘
flushdb :不触发存盘
shutdown：会触发磁盘
kill ：会触发磁盘，是温柔的关掉
kill -9：不会存盘
意外宕机 ：不触发存盘

那么数据保存了吗？存了多少？
127.0.0.1:6379> config get save

1. “save”
2. “900 1 30 10 60 10000”
   30 10是写到第10条数据的时候就触发存盘，虽然自己添加 了13条数据，但是只在磁盘中只存了10条数据
   这就是rdb的弊端：在两次触发存盘之间如果发生了宕机，意味着在这两次之间发生新增变化的数据会丢失。
   rdb的优点是：能压缩，可节省磁盘空间，恢复速度快。

手动备份:save

在不自动存盘的情况下，对于非常重要的数据可以手动存盘：save，虽然能立刻保存数据，但是有风险，rdb配置文件中的save是在后台又启动了一个线程（是bgsave后台备份）,手动执行save是主线程去执行save，有可能会阻塞其他进程

②AOF是增量备份：

rdb加载比较容易：怎么加载？
aof加载数据会非常慢，那么aof怎么优化呢？就是重写机制，rewrite。

以日志的形式来记录每个写操作，将Redis执行过的所有写指令记录下来(读操作不记录)，只许追加文件但不可以改写文件，Redis启动之初会读取该文件重新构建数据，换言之，Redis重启的话就根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作

AOF默认是不开的，需要手动开启：aof
appendonly no 改成 yes
可以在redis.conf中配置文件名称，默认为 appendonly.aof
AOF文件的保存路径，同RDB的路径一致。

AOF文件故障恢复

如遇到AOF文件损坏，可通过
redis-check-aof  --fix  appendonly.aof   进行恢复

如果当我们启动客户端的时候，报连接不上服务器是因为aof文件保存的数据不完整，可能是aof文件写残了，redis提供了一个工具，redis-check- aof --fix appendonly.aof,问要不要处理，输入y,之后再启动服务器就可以了

注意：

当 aof 和 rdb 都开启的时候会优先加载aof,但是第一次加载AOF数据是0，是空的，而且这种情况下不能shutdown，这时关掉会把内存中的空数据覆盖rdb,完了！

正确开 aof 的姿势：

不停机开启（在有数据的情况下，在使用客户端过程中开启aof）开启了rdb的情况下，在客户端直接改配置，**config set appendonly yes,**会自动生成以当前系统内存为准的aof备份文件。再往里面写数据，aof就能备份的。在内存客户端改完之后，再修改配置文件中的参数。

aof 存数据的同步频率设置：

# appendfsync always
appendfsync everysec
# appendfsync no

解释：
always：始终同步，每次Redis的写入都会立刻记入日志
everysec:每秒同步，每秒记入日志一次，如果宕机，本秒的数据可能丢失
no:把不主动进行同步，把同步时机交给操作系统。always是先把数据放buffer再写磁盘，no 不确定什么时候会把缓冲区数据写磁盘，是Linux判断什么时候适合写，这个设置一般不用

Rewrite 重写

AOF采用文件追加方式，文件会越来越大为避免出现此种情况，新增了重写机制,当AOF文件的大小超过所设定的阈值时，Redis就会启动AOF文件的内容压缩，只保留可以恢复数据的最小指令集.可以使用命令bgrewriteaof

redis如何实现重写

AOF文件持续增长而过大时，会fork出一条新进程来将文件重写(也是先写临时文件最后再rename)，遍历新进程的内存中数据，每条记录有一条的写操作。重写aof文件的操作，并没有读取旧的aof文件，而是将整个内存中的数据库内容用命令的方式重写了一个新的aof文件，这点和快照有点类似

aof对所有的写操作都会记录下来，比如我写了
set amount 1000
incrby amount 100
incrby amount 5000
decrby amount 600

我们可以在一定时间内汇总成一个统计值
set amount 5500
把对多条相同key的操作汇总就是重写rewrite，数据定期的被压缩，恢复的时候就更容易了。
但是这个计算的过程不能影响平时的读写操作，会单独开辟一个进程只做重写任务

重写时机

重写虽然可以节约大量磁盘空间，减少恢复时间。但是每次重写还是有一定的负担的，因此设定Redis要满足一定条件才会进行重写。
通过设置这两个参数：

auto-aof-rewrite-percentage 100
auto-aof-rewrite-min-size 64mb

解释：
系统载入时或者上次重写完毕时，Redis会记录此时AOF大小，设为base_size（压缩完之后的大小）,如果Redis的AOF当前大小>= base_size +base_size*100% (默认)且当前大小>=64mb(默认)的情况下，Redis会对AOF进行重写
第一次重写触发是AOF文件64M,压缩完之后假如是25M，第二次重写是100M,假如压缩完之后是70M，再下一次压缩是140M…

使用哪种持久化机制？

总结：
单机最好两个备份机制都开。
aof:
比rdb占用磁盘更多，恢复数据也比较慢，如果每条命令都写磁盘，性能也比较低，aof的校验可能会丢部分数据。但是备份机制更稳健，丢失数据概率更低，可读的日志文本，通过操作AOF稳健，可以处理误操作。
rdb:
如果对数据不敏感，可以单独选用rdb，这种情况通常是使用redis做缓存，后边有mysql数据库，丢失的数据可补回来。

六、主从复制

就是高可用

主从复制，就是主机数据更新后根据配置和策略，自动同步到备机的master/slaver机制，Master以写为主，Slave以读为主。

对于读写分离需要自己去做，大部分数据库都是读多写少，读写分离能减少阻塞。但是这种方式有一个一致性问题，还没备份到从机的，读的内容可能是旧的数据。对redis来说读写分离不是很明显，redis是基于内存的，写阻塞的压力非常小，通常不会读写分离，可以将数据放到不同的节点上（分库），也能减轻每台节点的压力。

搭建主从复制的节点：
我是在一台节点上演示的：

1. 创建目录：
/opt/module/ha_redis
2.创建文件：
文件1:**************************************
redis_6379.conf
内容：
daemonize yes
port 6379
dir /opt/module/ha_redis
dbfilename dump6379.rdb
pidfile /var/run/redis_6379.pid
protected-mode no
文件2***************************************
redis_6380.conf
内容
daemonize yes
port 6380
dir /opt/module/ha_redis
dbfilename dump6379.rdb
pidfile /var/run/redis_6379.pid
protected-mode no
文件3***************************************
redis_6381.conf
内容
daemonize yes
port 6381
dir /opt/module/ha_redis
dbfilename dump6379.rdb
pidfile /var/run/redis_6379.pid
protected-mode no
slave-priority 10  --优先级

info replication：打印主从复制的相关信息
slaveof ip port:成为某个实例的从服务器slaveof 192.168.3.103 6379

设置主从

使用的虚拟机ip：192.168.3.104
6379设置成主机，在6380和6381端口设置成从机
salveof 192.168.3.104 6379
salveof 192.168.3.104 6379
在主机set k1 v1
在两个从机能get到k1对应的v1值

关于一些问题的讨论

1 切入点问题？slave1、slave2是从头开始复制还是从切入点开始复制?比如从k4进来，那之前的123是否也可以复制/答案从头复制
mysql主从复制对老的数据是不继承的。继承了成本太高了，redis这么做是从内存到内存，性能是比较高的，能保证完全镜像。

2 从机是否可以写？set可否？答案默认从机不能写。但是可以改配置但是一般不会改。

3 主机shutdown后情况如何？从机是上位还是原地待命？答案原地待命

4 主机又回来了后，主机新增记录，从机还能否顺利复制？答案能

5 其中一台从机down后情况如何？依照原有它能跟上大部队吗？
**答案不能，是nil
因为slaveof命令只要是宕机了，在命令行的配置一律失效

复制原则

每次从机（slaveof之后）联通后，都会给主机发送sync指令，刚开始建立主从关系的时候是发rdb，之后每次都是主机写一条，从机就去获取一条

链式形式的master和lave

上一个slave可以是下一个slave的Master，slave同样可以接收其他slaves的连接和同步请求，那么该slave作为了链条中下一个的master, 可以有效减轻master的写压力,去中心化降低风险。
用 slaveof
中途变更转向:会清除之前的数据，重新建立拷贝最新的
风险是一旦某个slave宕机，后面的slave都没法备份

主机宕机之后

如果之前的主机宕机，想要自己指定哪个成为master，可以在客户端使用 slave no one 将从机变成主机，就算原来的主机在恢复，也不会成为主机了

哨兵模式

能够后台监控主机是否故障，如果故障了根据投票数自动将从机转换为主机。
只监控主机，只要主机宕机，会通过阈值判断主机起不来，如果是链式结构，从机能最直接控制其他从机，但是如果是星型结构，哨兵也会帮着选出新的主机，协调其他的从机跟着新的主机

配置哨兵

自定义的/opt/module/ha_redis目录下新建sentinel.conf文件
在配置文件中填写内容：

sentinel  monitor  mymaster  127.0.0.1  6379  1
protected-mode no

第一个参数：其中mymaster为监控对象起的服务器名称， 1 为 至少有多少个哨兵同意迁移的数量。 3个哨兵都会监控主机，如果其中一个哨兵的网线断了，会发起投票，是以规定的票数为准，1就是1个机器认为挂了，就是挂了，一般设成奇数（超过总节点的半数）。
第二个参数：路由的功能，把保护模式关了，其他的写请求是通过哨兵，哨兵把写的请求传递给master。就免去了客户端去寻找哪个节点是master的操作了。

启动哨兵

在/usr/local/bin目录下执行操作
执行redis-sentinel  /op/module/ha_redis/sentinel.conf
选用谁当主机是 由slave-priority 10一开始的设置决定的
本次设置的是6781端口优先级别高

故障恢复

七、redis集群

容量不够，redis如何进行扩容？

主从是每台机器都存有所有的数据，对单台机器的要求还是比较大的，我们可以搭建一个集群。
Redis 集群实现了对Redis的水平扩容，即启动N个redis节点，将整个数据库分布存储在这N个节点中，每个节点存储总数据的1/N。
Redis 同时在集群中也实现了 节点的主从复制。

jedis连接集群

代码实现：

private static JedisSentinelPool jedisSentinelPool=null;

public static Jedis getJedis(){
if(jedisSentinelPool==null){

JedisPoolConfig jedisPoolConfig=new JedisPoolConfig();
jedisPoolConfig.setMaxTotal(20);//最大可用连接数
jedisPoolConfig.setBlockWhenExhausted(true);//当连接耗尽时是否等待
jedisPoolConfig.setMaxWaitMillis(2000);//连接等待时间
jedisPoolConfig.setMaxIdle(5);//最大闲置连接数
jedisPoolConfig.setMinIdle(5);//最小闲置连接数
jedisPoolConfig.setTestOnBorrow(true);//获取到连接时是否判断，ping --pong

Set<String> sentinelSet= new HashSet<>();
sentinelSet.add("192.168.11.103:26379");

jedisSentinelPool= new JedisSentinelPool("mymaster",sentinelSet,jedisPoolConfig);
return  jedisSentinelPool.getResource();
}else{
return   jedisSentinelPool.getResource();
}
}

参数说明：

jedis连接集群：
setmaxTotal:最大可用连接数，
setMaxIdle:最大闲置连接数，
setminIdle:最少闲置连接数。
setBlockWhenExhausted(true)当连接耗尽是否等待，
setmaxWaitMillis:连接等待的时间
setTextOnBorrow:如果Java端的poll和shell中redis网短了，池子中的连接还在，但是连接不上，会报异常，再去请求，，可能会给一个号的连接，可能也是坏的，坏的还会报错，解决办法是我们请求连接时，取到连接时先做一下判断，测试一下。就是ping返回pang就是可以用的。
建立的这个连接是有开销的，
取出来的一定是master jedis,jedis 连接的是主机
Java客户端代码中并没有指明主机，客户端是通过哨兵，哨兵找到主机转发的。
用完之后的close，是把连接还回到连接池中，而不是关闭掉整个连接。

测试主函数

public static void main(String[] args) {

Jedis jedis = RedisUtil.getJedis();
jedis.set("k1000","v1000");
jedis.set("k2000","v2000");
System.out.println(jedis.get("k1000"));
jedis.close();
}

集群方案

①客户端方案

一致性hash
（key%虚节点个数 ，虚节点可以在节点上分配，哪个节点负责哪些节点的数据）是创建指定个数比如100个的虚节点，是环形的，如果A 到B的节点挂了，C节点会负责从A到C的数据备份。如果后台机器节点变了，需要在客户端进行机器和分片规则进行变换，分片规则耦合在客户端，需要自己实现很多功能， 不提供高可用

② 第三方代理中间件模式：twemproxy、 codis

把在客户端的分片规则放到了中间件中，中间代理的问题是，第三方中间节点挂了就没了，相当于只有一个哨兵，哨兵挂了，两端连接不上；好处是可以做读写分离

③redis官方解决方案：redis-cluster

是没有中间件的，几台机器之间是互相管理的，利用hash槽，是一致性hash的演变，hash槽能知道这条数据在哪个机器上存储，不管你访问的数据在不在访问的机器上，都能得到应答，如果不在被访问的节点上，这个节点会告知数据真正存储在哪个节点上。
而且有高可用，（主从复制）

并发写操作， redis如何分摊？

1. 首先安装ruby环境

搭建集群需要用ruby环境：
yum -y install  ruby-libs ruby ruby-irb ruby-rdoc rubygems
拷贝redis-3.2.0.gem到/opt/module目录下
执行t目录下执行  gem install --local redis-3.2.0.gem

2.每个节点安装redis，写配置文件

拷贝多个redis.conf文件
开启daemonize yes
Pid文件名字
指定端口
Log文件名字
Dump.rdb名字
Appendonly 关掉或者换名字
集群之间节点是能互相监控的不用哨兵

redis.conf的配置：

cluster-enabled yes//集群开关
cluster-config-file nodes-6379.conf//这个不用管，是3节点之间互通数据负责清单的，这份清单内存也有，相当于集群的元数据。
cluster-node-timeout 15000//超时，超15S ping不通 ，认为主机挂了
port 6379
daemonize yes
dir /xxxxx/xxxxx
dbfilename dump6379.rdb
appendonly no
protected-mode no
pidfile /xx
save 300 10

hadoop102:节点：

vim cluster_redis6379.conf文件：
daemonize yes
cluster-enabled yes
cluster-config-file nodes-6379.conf
cluster-node-timeout 15000
port 6379
dir /opt/module/cluster_redis
dbfilename dump6379.rdb
pidfile /var/run/redis_6379.pid
protected-mode no
save 300 10
appendonly no
***************************************
vim cluster_redis6380.conf文件：
vim cluster_redis6380.conf文件：
daemonize yes
cluster-enabled yes
cluster-config-file nodes-6380.conf
cluster-node-timeout 15000
port 6380
dir /opt/module/cluster_redis
dbfilename dump6380.rdb
pidfile /var/run/redis_6380.pid
protected-mode no
save 300 10
appendonly no

另外两台节点类似的 创建文文件，修改文件中对应的端口，cluster-config-file nodes-6380.conf和dbfilename，pidfile /var/run/redis_6380.pid
集群的节点之间能互相监控不用设置哨兵的。

3.启动集群

组合之前，先确保所有的redis实例启动。nodes_xxxx.conf文件都正常生成。
启动集群：进入到自己安装redis的目录src文件下：
cd /opt/module/redis-3.2.5/src
目录下有ruby脚本：redis-tribe.rb
执行操作：
注意：此处ip改成自己的

./redis-trib.rb create --replicas 1 192.168.3.102:6379 192.168.3.102:6380 192.168.3.104:6381 192.168.3.104:6389 192.168.3.103:6390 192.168.3.103:6391

4.通过cluster nodes查看集群信息

能查看到集群中的master和slave的具体状态，最前边是runid都是不同的，master行最后的数据是存放数据的槽，每一个key都是通过hash槽算法确定去往哪个槽，而每台机器上有自己管理的槽的范围，就能确定key存在了哪个节点上，注意这里的槽是理论上的概念，不是物理存在的。
现在是3台master，如果再新加节点，去调节槽的分配就可以了。槽的分配是可以手动的，槽连带数据都可以重新分配，就避免了增加新主机再进行hash的操作。

5.redis集群节点分配基本原则

一个集群至少要有三个主节点。
选项 --replicas 1 表示我们希望为集群中的每个主节点创建一个从节点。
分配原则尽量保证每个主数据库运行在不同的IP地址，每一对的从库和主库不在一个IP地址上。

>>> Check slots coverage...
[OK] All 16384 slots covered.
集群中一共有16384个hash槽

一个 Redis 集群包含 16384 个插槽（hash slot）， 数据库中的每个键都属于这 16384 个插槽的其中一个， 集群使用公式 CRC16(key) % 16384 来计算键 key 属于哪个槽， 其中 CRC16(key) 语句用于计算键 key 的 CRC16 校验和 。

6.集群中录入值

redis-cli客户端提供了 -c 参数实现自动重定向。
如 redis-cli -c -p 6379 登入后，再录入、查询键值对可以自动重定向。
不在一个slot下的键值，是不能使用mget,mset等多键操作。因为命令的原子性，集群中这种批处理有一定限制，但是可以通过{}定义组的概念。

192.168.3.104:6379> mset k3{g2} v3 k4{g2} v4 k5{g2} v5
OK
192.168.3.104:6379> mget k3{g2} k4{g2} k5{g2}
1) "v3"
2) "v4"
3) "v5"
192.168.3.104:6379>

通过{}来定义组的概念，使用{}中的值做hash槽的计算，但是这些数据都放到了一台节点上，很可能会造成热点问题。

7.查询集群中的值

获取指定key的值：

192.168.3.104:6379> cluster keyslot k3{g2}
(integer) 1196
192.168.3.104:6379> cluster keyslot k4{g2}
(integer) 1196
192.168.3.104:6379> cluster keyslot k5{g2}
(integer) 1196
192.168.3.104:6379>

查看当前节点上总共有多少数据

kesy * //只是当前节点的，不是集群总共的

查看某个槽中有多少数据：

192.168.3.104:6379> cluster countkeysinslot 1196
(integer) 3
192.168.3.104:6379

查看槽中指定个数的键

192.168.3.104:6379> cluster getkeysinslot 1196 3
1) "k3{g2}"
2) "k4{g2}"
3) "k5{g2}"

计算键key应该被放置在哪个槽上

cluster keyslot key10

可以通过以上的命令，获知槽中有多少数据，进而可以手动的进行调整，合理的分配数据.

8.故障恢复

1.如果主节点下线？从节点能否自动升为主节点？
答案：能
2.主节点恢复后，主从关系会如何？
答案：主节点恢复会跟随新的主机，变成从机。实际生产中最好就是一个节点放主和另一个主的从，不然太浪费机器的cpu
3.如果所有某一段插槽的主从节点都当掉，redis服务是否还能继续?
答案：可以通过参数设置：

redis.conf中的参数
cluster-require-full-coverage

默认是yes ，如果选no那么即使某一部分的slot完全下线(包括从机)，集群也会继续以现存的数据提供服务。（yes是在某个节点的主从都挂掉的情况下）必须要求所有的槽都在线，少一个槽就整个集群下线 。

9.集群的idea开发

//JedisCluster:本身是维护着连接池的
private static JedisCluster jedisCluster=null;
public static JedisCluster getJedisCluster(){if(jedisCluster==null){
Set<HostAndPort>  hostAndPortSet=new HashSet<>();
hostAndPortSet.add(new HostAndPort("hadoop102",6379));
hostAndPortSet.add(new HostAndPort("hadoop103",6380));
hostAndPortSet.add(new HostAndPort("hadoop104",6381));

JedisPoolConfig jedisPoolConfig=new JedisPoolConfig();
jedisPoolConfig.setMaxTotal(20);
jedisPoolConfig.setBlockWhenExhausted(true);
jedisPoolConfig.setMaxWaitMillis(2000);
jedisPoolConfig.setMaxIdle(5);
jedisPoolConfig.setMinIdle(5);
jedisPoolConfig.setTestOnBorrow(true);

jedisCluster= new JedisCluster(hostAndPortSet,jedisPoolConfig);
return jedisCluster;
}else{
return jedisCluster;
}
}

这里不写主机而是写从机，我们即使是只写其中一个也能找到集群的所有机器。我们写了多个，是防止万一我们写的 一个那个主机宕机了，可以通过其他的节点找到集群。redis集群是去中心化的。
测试：

public static void main(String[] args) {

JedisCluster jedisCluster = RedisUtil.getJedisCluster();
jedisCluster.set("k111","v111");
jedisCluster.set("k222","v222");
jedisCluster.set("k333","v333");
//集群的线程池自己不要关，集群自己会关
// jedisCluster.close();
}

注意JedisCluster 内部会自己维护连接的保持和释放，不用手动close