SortedSet类型zset

　　在集合类型的基础上，有序集合为集合中的每个元素都关联一个分数，这使得我们不仅可以完成插入、删除和判断元素是否存在集合中，还能够获得最高或最低的前N个元素、获取指定分数范围内的元素等与分苏有关的操作。

在某些方面有序集合和列表类型有些相似。

1. 二者都是有序的。
2. 二者都可以获得某一范围的元素

但是二者有着很大的区别：

1. 列表类型是通过链表实现的，后去靠近两端的数据速度极快，而当元素增多后，访问中间数据的速度会变慢。
2. 有序集合类型使用散列实现，所有即使读取位于中间部分的数据也很快。
3. 列表中不能简单的调整某个元素的位置，但是有序集合可以(通过更改分数实现)。
4. 有序集合要比列表类型更耗内存。

添加元素

　　向有序集合中加入一个元素和该元素的分数，如果该元素已经存在则会用新的分数替换原有的分数。返回值是新加入到集合中的元素个数，不不含之前已经存在的元素。

语法：ZADD key score member [score member ...]

获取排名在某个范围的元素列表

按照元素分数从小到大的顺序返回索引从start到stop之间的所有元素(包含两端的元素)

语法：ZRANGE key start stop [WITHSCORES]

如果需要获取元素的分数的可以在命令尾部加上WITHSCORES参数

获取元素的分数

语法：ZSCORE key member

删除元素

移除有序集key中的一个或多个成员，不存在的成员将被忽略。

当key存在但不是有序集类型时，返回错误。

语法：ZREM key member [member ...]

获取指定分数范围的元素

语法：ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count]

增加某个元素的分数#

返回值是更改后的分数

语法：ZINCRBY key increment member

获取集合中元素的数量

语法：ZCARD key

获得指定分数范围内的元素个数

语法：ZCOUNT key min max

按照排名范围删除元素

语法：ZREMRANGEBYRANK key start stop

按照分数范围删除元素

语法：ZREMRANGEBYSCORE key min max

获取元素的排名

从小到大

语法：ZRANK key member

从大到小

语法：ZREVRANK key member

应用之商品销售排行榜

需求：根据商品销售对商品进行排序显示

思路：定义商品销售排行榜(sorted set集合)，key为items:sellsort，分数为商品小数量。

写入商品销售量：

>商品编号1001的销量是9，商品编号1002的销量是10

>商品编号1001销量家1

>商品销量前10名

通用命令

keys

语法：keys pattern

del

语法：DEL key

exists

作用：确认一个key是否存在

语法：exists key

expire

　　Redis在实际使用过程中更多的用作缓存，然后缓存的数据一般都是需要设置生存时间的，即：到期后数据销毁。

EXPIRE key seconds             设置key的生存时间（单位：秒）key在多少秒后会自动删除

TTL key                     查看key生于的生存时间

PERSIST key                清除生存时间 

PEXPIRE key milliseconds    生存时间设置单位为：毫秒

例子：
192.168.101.3:7002> set test 1        设置test的值为1
OK
192.168.101.3:7002> get test            获取test的值
"1"
192.168.101.3:7002> EXPIRE test 5    设置test的生存时间为5秒
(integer) 1
192.168.101.3:7002> TTL test            查看test的生于生成时间还有1秒删除
(integer) 1
192.168.101.3:7002> TTL test
(integer) -2
192.168.101.3:7002> get test            获取test的值，已经删除
(nil)

rename

作用：重命名key

语法：rename oldkey newkey

type

作用：显示指定key的数据类型

语法：type key

Redis事务

事务介绍

• Redis的事务是通过MULTI，EXEC，DISCARD和WATCH这四个命令来完成。
• Redis的单个命令都是原子性的，所以这里确保事务性的对象是命令集合。
• Redis将命令集合序列化并确保处于一事务的命令集合连续且不被打断的执行。
• Redis不支持回滚的操作。

相关命令

• MULTI

　　　　注：用于标记事务块的开始。

　　　　Redis会将后续的命令逐个放入队列中，然后使用EXEC命令原子化地执行这个命令序列。

　　　　语法：MULTI

• EXEC

　　　　在一个事务中执行所有先前放入队列的命令，然后恢复正常的连接状态。

　　　　语法：EXEC

• DISCARD

　　　　清楚所有先前在一个事务中放入队列的命令，然后恢复正常的连接状态。

　　　　语法：DISCARD

• WATCH

　　　　当某个事务需要按条件执行时，就要使用这个命令将给定的键设置为受监控的状态。

　　　　语法：WATCH key [key ....]

　　　　注：该命令可以实现redis的乐观锁

• UNWATCH

　　　　清除所有先前为一个事务监控的键。

　　　　语法：UNWATCH

事务失败处理

• Redis语法错误(编译器错误)

• Redis类型错误(运行期错误)

为什么redis不支持事务回滚？

1. 大多数事务失败是因为语法错误或者类型错误，这两种错误，再开发阶段都是可以避免的
2. Redis为了性能方面就忽略了事务回滚

Redis实现分布式锁

锁的处理

　　单应用中使用锁：单线程多线程

　　　　synchronize、Lock

　　分布式应用中使用锁：多进程

分布式锁的实现方式

1. 数据库的乐观锁
2. 给予zookeeper的分布式锁
3. 给予redis的分布式锁

分布式锁的注意事项

1. 互斥性：在任意时刻，只有一个客户端能持有锁
2. 同一性：加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了。
3. 避免死锁：即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。

实现分布式锁

获取锁#

方式一(使用set命令实现)

方式二(使用setnx命令实现)

package com.cyb.redis.utils;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

public class jedisUtils {
    private static String ip = "192.168.31.200";
    private static int port = 6379;
    private static JedisPool pool;
    static {
        pool = new JedisPool(ip, port);
    }
    public static Jedis getJedis() {
        return pool.getResource();
    }
    public static boolean getLock(String lockKey, String requestId, int timeout) {
        //获取jedis对象，负责和远程redis服务器进行连接
        Jedis je=getJedis();
        //参数3：NX和XX
        //参数4：EX和PX
        String result = je.set(lockKey, requestId, "NX", "EX", timeout);
        if (result=="ok") {
            return true;
        }
        return false;
    }

    public static synchronized boolean getLock2(String lockKey, String requestId, int timeout) {
        //获取jedis对象，负责和远程redis服务器进行连接
        Jedis je=getJedis();
        //参数3：NX和XX
        //参数4：EX和PX
        Long result = je.setnx(lockKey, requestId);
        if (result==1) {
            je.expire(lockKey, timeout); //设置有效期
            return true;
        }
        return false;
    }
}

释放锁

package com.cyb.redis.utils;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

public class jedisUtils {
    private static String ip = "192.168.31.200";
    private static int port = 6379;
    private static JedisPool pool;
    static {
        pool = new JedisPool(ip, port);
    }
    public static Jedis getJedis() {
        return pool.getResource();
    }
    /**
     * 释放分布式锁
     * @param lockKey
     * @param requestId
     */
    public static void releaseLock(String lockKey, String requestId) {
        Jedis je=getJedis();
        if (requestId.equals(je.get(lockKey))) {
            je.del(lockKey);
        }
    }
}

Redis持久化方案#

导读

　　Redis是一个内存数据库，为了保证数据的持久性，它提供了两种持久化方案。

1. RDB方式(默认)
2. AOF方式

RDB方式

　　RDB是Redis默认采用的持久化方式。

　　RDB方式是通过快照(snapshotting)完成的，当符合一定条件时Redis会自动将内存中的数据进行快照并持久化到硬盘。

RDB触发条件

1. 符合自定义配置的快照规则
2. 执行save或者bgsave命令
3. 执行flushall命令
4. 执行主从复制操作

在redis.conf中设置自定义快照规则

1、RDB持久化条件

　　格式：save <seconds> <changes>

示例:

　　save 900 1：表示15分钟(900秒)内至少1个键更改则进行快照。

　　save 300 10：表示5分钟(300秒)内至少10个键被更改则进行快照。

　　save 60 10000：表示1分钟内至少10000个键被更改则进行快照。

2、配置dir指定rdb快照文件的位置

# Note that you must specify a directory here, not a file name.
dir ./

3、配置dbfilename指定rdb快照文件的名称

# The filename where to dump the DB
dbfilename dump.rdb

说明

1. Redis启动后会读取RDB快照文件，将数据从硬盘载入到内存
2. 根据数据量大小与结构和服务器性能不同，这个时间也不同。通常将记录1千万个字符串类型键，大小为1GB的快照文件载入到内存中需要花费20-30秒钟。

快照的实现原理

快照过程

1. redis使用fork函数复制一份当前进程的副本(子进程)
2. 父进程继续接受并处理客户端发来的命令，而子进程开始将内存中的数据写入到硬盘中的临时文件。
3. 当子进程写入完所有数据后会用该临时文件替换旧的RDB文件，至此，一次快照操作完成。

注意

1. redis在进行快照的过程中不会修改RDB文件，只有快照结束后才会将旧的文件替换成新的，也就是说任何时候RDB文件都是完整的。
2. 这就使得我们可以通过定时备份RDB文件来实现redis数据库的备份，RDB文件是经过压缩的二进制文件，占用的空间会小于内存中的数据，更加利于传输。

RDB优缺点

缺点

　　使用RDB方式实现持久化，一旦redis异常退出，就会丢失最后一次快照以后更改的所有数据。这个时候我们就需要根据具体的应用场景，通过组合设置自动快照条件的方式将可能发生的数据损失控制在能够接受范围。如果数据相对来说比较重要，希望将损失降到最小，则可以使用AOF方式进行持久化

优点

　　RDB可以最大化redis的性能：父进程在保存RDB文件时唯一要做的就是fork出一个字进程，然后这个子进程就会处理接下来的所有保存工作，父进程无需执行任何磁盘I/O操作。同时这个也是一个缺点，如果数据集比较大的时候，fork可能比较耗时，造成服务器在一段时间内停止处理客户端的请求。

AOF方式

介绍

　　默认情况下Redis没有开启AOF(append only file)方式的持久化

　　开启AOF持久化后每执行一条会更改Redis中的数据命令，Redis就会将该命令写入硬盘中的AOF文件，这一过程显示会降低Redis的性能，但大部分下这个影响是能够接受的，另外使用较快的硬盘可以提高AOF的性能。

配置redis.conf

设置appendonly参数为yes

appendonly yes

AOF文件的保存位置和RDB文件的位置相同，都是通过dir参数设置的

dir ./

默认的文件名是appendonly.aof，可以通过appendfilename参数修改

appendfilename appendonly.aof

AOF重写原理(优化AOF文件)

1. Redis可以在AOF文件体积变得过大时，自动地后台对AOF进行重写
2. 重写后的新AOF文件包含了恢复当前数据集所需的最小命令集合。
3. 整个重写操作是绝对安全的，因为Redis在创建新的AOF文件的过程中，会继续将命令追加到现有的AOF文件里面，即使重写过程中发生停机，现有的AOF文件也不会丢失。而一旦新AOF文件创建完毕，Redis就会从旧AOF文件切换到新AOF文件，并开始对新AOF文件进行追加操作。
4. AOF文件有序地保存了对数据库执行的所有写入操作，这些写入操作以Redis协议的格式保存，因此AOF文件的内容非常容易被人读懂，对文件进行分析(parse)也很轻松。

参数说明

1. #auto-aof-rewrite-percentage 100：表示当前aof文件大小超过上次aof文件大小的百分之多少的时候会进行重写。如果之前没有重写过，以启动时aof文件大小为基准。
2. #auto-aof-rewrite-min-size 64mb：表示限制允许重写最小aof文件大小，也就是文件大小小于64mb的时候，不需要进行优化

同步磁盘数据

　　Redis每次更改数据的时候，aof机制都会将命令记录到aof文件，但是实际上由于操作系统的缓存机制，数据并没有实时写入到硬盘，而是进入硬盘缓存。再通过硬盘缓存机制去刷新到保存文件中。

参数说明

1. appendfsync always：每次执行写入都会进行同步，这个是最安全但是效率比较低
2. appendfsync everysec：每一秒执行
3. appendfsync no：不主动进行同步操作，由于操作系统去执行，这个是最快但是最不安全的方式

AOF文件损坏以后如何修复

　　服务器可能在程序正在对AOF文件进行写入时停机，如果停机造成AOF文件出错(corrupt)，那么Redis在重启时会拒绝载入这个AOF文件，从而确保数据的一致性不会被破坏。

　　当发生这种情况时，可以以以下方式来修复出错的AOF文件：

　　　　1、为现有的AOF文件创建一个备份。

　　　　2、使用Redis附带的redis-check-aof程序，对原来的AOF文件进行修复。

　　　　3、重启Redis服务器，等待服务器字啊如修复后的AOF文件，并进行数据恢复。

如何选择RDB和AOF

1. 一般来说，如果对数据的安全性要求非常高的话，应该同时使用两种持久化功能。
2. 如果可以承受数分钟以内的数据丢失，那么可以只使用RDB持久化。
3. 有很多用户都只使用AOF持久化，但并不推荐这种方式：因为定时生成RDB快照(snapshot)非常便于进行数据库备份，并且RDB恢复数据集的速度也要比AOF恢复的速度要快。
4. 两种持久化策略可以同时使用，也可以使用其中一种。如果同时使用的话，那么Redis启动时，会优先使用AOF文件来还原数据。

Redis的主从复制

什么是主从复制

　　持久性保证了即使redis服务重启也不会丢失数据，因为redis服务重启后将硬盘上持久化的数据恢复到内存中，但是当redis服务器的硬盘损坏了可能导致数据丢失，不过通过redis的主从复制机制旧可以避免这种单点故障，如下图：

说明：

1. 主redis中的数据有两个副本(replication)即从redis1和从redis2，即使一台redis服务器宕机其他两台redis服务也可以继续提供服务。
2. 主redis中的数据和从redis上的数据保持实时同步，当主redis写入数据时通过主从复制机制会复制到两个从redis服务上。
3. 只有一个主redis，可以有多个从redis。
4. 主从复制不会阻塞master，在同步数据时，master可以继续处理client请求
5. 一个redis可以即是主从，如下图：

主从配置

主redis配置

　　无需特殊配置

从redis配置

　　修改从服务器上的redis.conf文件

# slaveof <masterip> <masterport>
slaveof 192.168.31.200 6379

　　上边的配置说明当前【从服务器】对应的【主服务器】的ip是192.168.31.200，端口是6379.

实现原理

1. slave第一次或者重连到master发送一个SYNC的命令。
2. master收到SYNC的时候，会做两件事执行bgsave(rdb的快照文件)master会把新收到的修改命令存入到缓冲区

缺点：没有办法对master进行动态选举

Redis Sentinel哨兵机制

简介

　　Sentinel(哨兵)进程是用于监控redis集群中Master主服务器工作的状态，在Master主服务器发生故障的时候，可以实现Master和Slave服务器的切换，保证系统的高可用，其已经被集成在redis2.6+的版本中，Redis的哨兵模式到2.8版本之后就稳定了下来。

哨兵进程的作用

1. 监控(Monitoring)：哨兵(Sentinel)会不断地检查你的Master和Slave是否运作正常。
2. 提醒(Notification)：当被监控的某个Redis节点出现问题时，哨兵(Sentinel)可以通过API向管理员或者其他应用程序发送通知。
3. 自动故障迁移(Automatic failover)：当一个Master不能正常工作时，哨兵(Sentinel)会开始一次自动故障迁移操作。它会将失效Master的其中一个Slave升级为新的Master，并让失效Master的其他Slave改为复制新的Master；当客户端视图连接失效的Master时，集群也会向客户端返回新Master的地址，使得集群可以使用现在的Master替换失效的Master。Master和Slave服务器切换后，Master的redis.conf、Slave的redis.conf和sentinel.conf的配置文件的内容都会发生相应的改变，即Master主服务器的redis.conf配置文件中会多一行Slave的配置，sentinel.conf的监控目标会随之调换。

哨兵进程的工作方式

1. 每个Sentinel(哨兵)进程以每秒钟一次的频率向整个集群中的Master主服务器，Slave从服务器以及其他Sentinel(哨兵)进程发送一个PING命令。
2. 如果一个实例(instance)距离最后一次有效回复PING命令的时间超过down-after-milliseconds选项所指定的值，则这个实例会被Sentinel(哨兵)进程标记为主观下线(SDOWN)。
3. 如果一个Master主服务器被标记为主观下线(SDOWN)，则正在监视这个Master主服务器的所有Sentinel(哨兵)进程要以每秒一次的频率确认Master主服务器确实进入了主观下线状态。
4. 当有足够数量的Sentinel(哨兵)进程(大于等于配置文件指定的值)在指定的时间范围内确认Master主服务器进入了主观下线状态(SDOWN)，则Master主服务器会被标记为客观下线(ODOWN)。
5. 在一般情况下，每个Sentinel(哨兵)进程会以每10秒一次的频率向集群中的所有Master主服务器、Slave从服务器发送INFO命令。
6. 当Master主服务器被Sentinel(哨兵)进程标记为客观下线(ODOWN)时，Sentinel(哨兵)进程向下线的Master主服务器的所有Slave从服务器发送INFO命令的频率会从10秒一次改为每秒一次。
7. 若没有足够数量的Sentinel(哨兵)进程同意Master主服务器下线，Master主服务器的客观下线状态就会被移除。若Master主服务器重新向Sentinel(哨兵)进程发送PING命令返回有效回复，Master主服务器的主观下线状态就会被移除。

实现

修改从机的sentinel.conf

sentinel monitor mymaster  192.168.127.129 6379 1

启动哨兵服务器

redis-sentinel

Redis Cluster集群

redis-cluster架构图

架构细节

1. 所有的redis节点彼此互联(PING-PING机制)，内部使用二进制协议优化传输速度和带宽。
2. 节点的fail是通过集群中超过半数的节点检测失效时才生效。
3. 客户端与redis节点直连，不需要中间proxy层，客户端不需要连接集群所有节点，连接集群中任何一个可用节点即可。
4. redis-cluster把所有的物理节点映射到[0-16383]slot上，cluster负责维护node<->slot<->value

    Redis集群中内置了16384个哈希槽，当需要在Redis集群中放置一个key-value时，redis先对key使用crc16算法算出一个结果，然后把结果对16384求余数，这样每个key都会对应一个编号在0-16384之间的哈希槽，redis会根据节点数量大致均等的将哈希槽映射到不同节点。

redis-cluster投票：容错

安装Ruby环境

导读

　　redis集群需要使用集群管理脚本redis-trib.rb，它的执行相应依赖ruby环境。

安装

安装ruby

yum install ruby
yum install rubygems

将redis-3.2.9.gen拖近Linux系统

安装ruby和redis的接口程序redis-3.2.9.gem

gem install redis-3.2.9.gem

复制redis-3.2.9/src/redis-trib.rb 文件到/usr/local/redis目录

cp redis-3.2.9/src/redis-trib.rb /usr/local/redis/ -r

安装Redis集群(RedisCluster)

　　Redis集群最少需要三台主服务器,三台从服务器,端口号分别为7001~7006。

创建7001实例，并编辑redis.conf文件，修改port为7001。

修改redis.conf配置文件，打开Cluster-enable yes

重复以上2个步骤，完成7002~7006实例的创建，注意端口修改#

启动所有的实例

创建Redis集群

./redis-trib.rb create --replicas 1 192.168.242.129:7001 192.168.242.129:7002 192.168.242.129:7003 192.168.242.129:7004 192.168.242.129:7005  192.168.242.129:7006
>>> Creating cluster
Connecting to node 192.168.242.129:7001: OK
Connecting to node 192.168.242.129:7002: OK
Connecting to node 192.168.242.129:7003: OK
Connecting to node 192.168.242.129:7004: OK
Connecting to node 192.168.242.129:7005: OK
Connecting to node 192.168.242.129:7006: OK
>>> Performing hash slots allocation on 6 nodes...
Using 3 masters:
192.168.242.129:7001
192.168.242.129:7002
192.168.242.129:7003
Adding replica 192.168.242.129:7004 to 192.168.242.129:7001
Adding replica 192.168.242.129:7005 to 192.168.242.129:7002
Adding replica 192.168.242.129:7006 to 192.168.242.129:7003
M: d8f6a0e3192c905f0aad411946f3ef9305350420 192.168.242.129:7001
   slots:0-5460 (5461 slots) master
M: 7a12bc730ddc939c84a156f276c446c28acf798c 192.168.242.129:7002
   slots:5461-10922 (5462 slots) master
M: 93f73d2424a796657948c660928b71edd3db881f 192.168.242.129:7003
   slots:10923-16383 (5461 slots) master
S: f79802d3da6b58ef6f9f30c903db7b2f79664e61 192.168.242.129:7004
   replicates d8f6a0e3192c905f0aad411946f3ef9305350420
S: 0bc78702413eb88eb6d7982833a6e040c6af05be 192.168.242.129:7005
   replicates 7a12bc730ddc939c84a156f276c446c28acf798c
S: 4170a68ba6b7757e914056e2857bb84c5e10950e 192.168.242.129:7006
   replicates 93f73d2424a796657948c660928b71edd3db881f
Can I set the above configuration? (type 'yes' to accept): yes
>>> Nodes configuration updated
>>> Assign a different config epoch to each node
>>> Sending CLUSTER MEET messages to join the cluster
Waiting for the cluster to join....
>>> Performing Cluster Check (using node 192.168.242.129:7001)
M: d8f6a0e3192c905f0aad411946f3ef9305350420 192.168.242.129:7001
   slots:0-5460 (5461 slots) master
M: 7a12bc730ddc939c84a156f276c446c28acf798c 192.168.242.129:7002
   slots:5461-10922 (5462 slots) master
M: 93f73d2424a796657948c660928b71edd3db881f 192.168.242.129:7003
   slots:10923-16383 (5461 slots) master
M: f79802d3da6b58ef6f9f30c903db7b2f79664e61 192.168.242.129:7004
   slots: (0 slots) master
   replicates d8f6a0e3192c905f0aad411946f3ef9305350420
M: 0bc78702413eb88eb6d7982833a6e040c6af05be 192.168.242.129:7005
   slots: (0 slots) master
   replicates 7a12bc730ddc939c84a156f276c446c28acf798c
M: 4170a68ba6b7757e914056e2857bb84c5e10950e 192.168.242.129:7006
   slots: (0 slots) master
   replicates 93f73d2424a796657948c660928b71edd3db881f
[OK] All nodes agree about slots configuration.
>>> Check for open slots...
>>> Check slots coverage...
[OK] All 16384 slots covered.
[root@localhost-0723 redis]#

命令客户端连接集群

命令：

./redis-cli -h 127.0.0.1 -p 7001 -c


注：-c表示是以redis集群方式进行连接

./redis-cli -p 7006 -c
127.0.0.1:7006> set key1 123
-> Redirected to slot [9189] located at 127.0.0.1:7002
OK
127.0.0.1:7002>

查看集群的命令

查看集群状态

127.0.0.1:7003> cluster info
cluster_state:ok
cluster_slots_assigned:16384
cluster_slots_ok:16384
cluster_slots_pfail:0
cluster_slots_fail:0
cluster_known_nodes:6
cluster_size:3
cluster_current_epoch:6
cluster_my_epoch:3
cluster_stats_messages_sent:926
cluster_stats_messages_received:926

查看集群中的节点

127.0.0.1:7003> cluster nodes
7a12bc730ddc939c84a156f276c446c28acf798c 127.0.0.1:7002 master - 0 1443601739754 2 connected 5461-10922
93f73d2424a796657948c660928b71edd3db881f 127.0.0.1:7003 myself,master - 0 0 3 connected 10923-16383
d8f6a0e3192c905f0aad411946f3ef9305350420 127.0.0.1:7001 master - 0 1443601741267 1 connected 0-5460
4170a68ba6b7757e914056e2857bb84c5e10950e 127.0.0.1:7006 slave 93f73d2424a796657948c660928b71edd3db881f 0 1443601739250 6 connected
f79802d3da6b58ef6f9f30c903db7b2f79664e61 127.0.0.1:7004 slave d8f6a0e3192c905f0aad411946f3ef9305350420 0 1443601742277 4 connected
0bc78702413eb88eb6d7982833a6e040c6af05be 127.0.0.1:7005 slave 7a12bc730ddc939c84a156f276c446c28acf798c 0 1443601740259 5 connected
127.0.0.1:7003>

维护节点

　　集群创建完成后可以继续向集群中添加节点。

添加主节点

添加7007节点作为新节点

命令：./redis-trib.rb add-node 127.0.0.1:7007 127.0.0.1:7001

查看集群节点发现7007已加到集群中

给刚添加的7007节点分配槽

第一步：连上集群(连接集群中任意一个可用节点都行)

./redis-trib.rb reshard 192.168.101.3:7001

第二步：输入要分配的槽数量

输入500，表示要分配500个槽

第三步：输入接收槽的节点id

输入：15b809eadae88955e36bcdbb8144f61bbbaf38fb

ps：这里准备给7007分配槽，通过cluster node查看7007节点id为：

15b809eadae88955e36bcdbb8144f61bbbaf38fb

第四步：输入源节点id

输入：all

第五步：输入yes开始移动槽到目标节点id

输入：yes

添加从节点

　　添加7008从节点，将7008作为7007的从节点

命令：

./redis-trib.rb add-node --slave --master-id  主节点id   新节点的ip和端口   旧节点ip和端口

执行如下命令：

./redis-trib.rb add-node --slave --master-id cad9f7413ec6842c971dbcc2c48b4ca959eb5db4  192.168.101.3:7008 192.168.101.3:7001

cad9f7413ec6842c971dbcc2c48b4ca959eb5db4 是7007结点的id，可通过cluster nodes查看。

nodes查看

注意：如果原来该节点在集群中的配置信息已经生成到cluster-config-file指定的配置文件中(如果cluster-config-file没有指定则默认为nodes.conf)，这时可能会报错

[ERR] Node XXXXXX is not empty. Either the node already knows other nodes (check with CLUSTER NODES) or contains some key in database 0

解决办法是删除生成的配置文件nodes.conf，删除后再执行./redis-trib.rb add-node指令

查看集群中的节点，刚添加7008为7007的从节点

删除节点

命令：

./redis-trib.rb del-node 127.0.0.1:7005 4b45eb75c8b428fbd77ab979b85080146a9bc017

删除已经占用hash槽的节点会失败，报错如下

[ERR] Node 127.0.0.1:7005 is not empty! Reshard data away and try again.

需要将该节点占用的hash槽分配出去

Jedis连接集群

创建JedisCluster类连接Redis集群

@Test
public void testJedisCluster() throws Exception {
    //创建一连接，JedisCluster对象,在系统中是单例存在
    Set<HostAndPort> nodes = new HashSet<>();
    nodes.add(new HostAndPort("192.168.242.129", 7001));
    nodes.add(new HostAndPort("192.168.242.129", 7002));
    nodes.add(new HostAndPort("192.168.242.129", 7003));
    nodes.add(new HostAndPort("192.168.242.129", 7004));
    nodes.add(new HostAndPort("192.168.242.129", 7005));
    nodes.add(new HostAndPort("192.168.242.129", 7006));
    JedisCluster cluster = new JedisCluster(nodes);
    //执行JedisCluster对象中的方法，方法和redis一一对应。
    cluster.set("cluster-test", "my jedis cluster test");
    String result = cluster.get("cluster-test");
    System.out.println(result);
    //程序结束时需要关闭JedisCluster对象
    cluster.close();
}

使用Spring

配置applicationContext.xml

<!-- 连接池配置 -->
<bean id="jedisPoolConfig" class="redis.clients.jedis.JedisPoolConfig">
    <!-- 最大连接数 -->
    <property name="maxTotal" value="30" />
    <!-- 最大空闲连接数 -->
    <property name="maxIdle" value="10" />
    <!-- 每次释放连接的最大数目 -->
    <property name="numTestsPerEvictionRun" value="1024" />
    <!-- 释放连接的扫描间隔（毫秒） -->
    <property name="timeBetweenEvictionRunsMillis" value="30000" />
    <!-- 连接最小空闲时间 -->
    <property name="minEvictableIdleTimeMillis" value="1800000" />
    <!-- 连接空闲多久后释放, 当空闲时间>该值 且 空闲连接>最大空闲连接数 时直接释放 -->
    <property name="softMinEvictableIdleTimeMillis" value="10000" />
    <!-- 获取连接时的最大等待毫秒数,小于零:阻塞不确定的时间,默认-1 -->
    <property name="maxWaitMillis" value="1500" />
    <!-- 在获取连接的时候检查有效性, 默认false -->
    <property name="testOnBorrow" value="true" />
    <!-- 在空闲时检查有效性, 默认false -->
    <property name="testWhileIdle" value="true" />
    <!-- 连接耗尽时是否阻塞, false报异常,ture阻塞直到超时, 默认true -->
    <property name="blockWhenExhausted" value="false" />
</bean>
<!-- redis集群 -->
<bean id="jedisCluster" class="redis.clients.jedis.JedisCluster">
    <constructor-arg index="0">
        <set>
            <bean class="redis.clients.jedis.HostAndPort">
                <constructor-arg index="0" value="192.168.101.3"></constructor-arg>
                <constructor-arg index="1" value="7001"></constructor-arg>
            </bean>
            <bean class="redis.clients.jedis.HostAndPort">
                <constructor-arg index="0" value="192.168.101.3"></constructor-arg>
                <constructor-arg index="1" value="7002"></constructor-arg>
            </bean>
            <bean class="redis.clients.jedis.HostAndPort">
                <constructor-arg index="0" value="192.168.101.3"></constructor-arg>
                <constructor-arg index="1" value="7003"></constructor-arg>
            </bean>
            <bean class="redis.clients.jedis.HostAndPort">
                <constructor-arg index="0" value="192.168.101.3"></constructor-arg>
                <constructor-arg index="1" value="7004"></constructor-arg>
            </bean>
            <bean class="redis.clients.jedis.HostAndPort">
                <constructor-arg index="0" value="192.168.101.3"></constructor-arg>
                <constructor-arg index="1" value="7005"></constructor-arg>
            </bean>
            <bean class="redis.clients.jedis.HostAndPort">
                <constructor-arg index="0" value="192.168.101.3"></constructor-arg>
                <constructor-arg index="1" value="7006"></constructor-arg>
            </bean>
        </set>
    </constructor-arg>
    <constructor-arg index="1" ref="jedisPoolConfig"></constructor-arg>
</bean>

测试代码

private ApplicationContext applicationContext;
    @Before
    public void init() {
        applicationContext = new ClassPathXmlApplicationContext(
                "classpath:applicationContext.xml");
    }

    // redis集群
    @Test
    public void testJedisCluster() {
        JedisCluster jedisCluster = (JedisCluster) applicationContext
                .getBean("jedisCluster");

        jedisCluster.set("name", "zhangsan");
        String value = jedisCluster.get("name");
        System.out.println(value);
    }

缓存穿透、缓存击穿、缓存雪崩#

缓存数据步骤

1. 查询缓存，如果没有数据，则查询数据库
2. 查询数据库，如果数据不为空，将结果写入缓存

缓存穿透

什么叫缓存穿透？

　　一般的缓存系统，都是按照key去缓存查询，如果不存在对应的value，就应该去后端系统查询。如果key对应的value是一定不存在的，并且对key并发请求量很大，就会对后端系统造成很大的压力。这就叫做缓存穿透。

如何解决？

1. 对查询结果为空的情况也进行缓存，缓存时间设置短一点，或者该key对应的数据insert了之后清楚缓存。
2. 对一定不存在的key进行过滤。可以把所有的可能存在的key放到一个大的Bitmap中，查询时通过该Bitmap过滤。(布隆表达式)

缓存雪崩

什么叫缓存雪崩？

　　当缓存服务器重启或者大量缓存集合中某一个时间段失效，这样在失效的时候，也会给后端系统带来很大压力。

如何解决？

1. 在缓存失效后，通过加锁或者队列来控制读数据库写缓存的线程数量。比如对某个key只允许一个线程查询数据和写缓存，其他线程等待。
2. 不同的key，设置不同的过期时间，让缓存失效的时间点尽量均匀。
3. 做二级缓存，A1为原始缓存，A3为拷贝缓存，A1失效时，可以访问A2，A1缓存失效时间设置为短期，A2设置为长期。

缓存击穿

什么叫缓存击穿？

　　对于一些设置了过期时间的key，如果这些key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。这个时候，需要考虑一个问题：“缓存”被击穿的问题，这个和缓存雪崩的区别在于这里针对某一key缓存，前者则是很多key。

　　缓存在某个时间点过期的时候，恰好在这个时间点对这个key有大量的并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。

如何解决？

　　使用redis的setnx互斥锁先进行判断，这样其他线程就处于等待状态，保证不会有大并发操作去操作数据库。

if(redis.setnx()==1){

//查数据库

//加入线程

}

觉得有用的话，还请转发+评论+关注哦

作者：陈彦斌

出处：https://www.cnblogs.com/chenyanbin/