使用redis
使用redis之前,先对redis的五大对象及其底层实现了解,这样更能方便我们去选用某种对象来实现我们的需求,当然还有redis的哨兵模式,集群模式,持久化等内容也会在本篇文章中介绍。
如果对英文官网看的不舒服的话,可以使用访问https://www.redis.net.cn,中文网站。
1、redis的五大对象及其底层实现
1.1、string
字符串可以是字符串(简单字符串,复杂字符串(json xml)),数字(整数,浮点数),甚至是二进制(图片,音频,视频),但是最大值不能超过512M。
常用命令
#添加元素 //ex:秒级过期时间,nx:键不存在时才能设置成功,xx键存在时才能设置成功
set key value [ex seconds] [px milliseconds] [nx|xx]
# 获取元素 基于key
get key
# 批量设置值
mset key value key value
#批量获取值
mget key1 key2
#Redis Incr 如果 key 不存在,那么 key 的值会先被初始化为 0 ,然后再执行 INCR 操作。如果值包含错误的类型,或字符串类型的值不能表示为数字,那么返回一个错误
incr key
# edis Incrby 命令将 key 中储存的数字加上指定的增量值。
incrby key 30
#Redis Decr 命令将 key 中储存的数字值减一
decr key
#Redis Decrby 命令将 key 所储存的值减去指定的减量值。
decrby key 20
#对值得追加
append key value
#获取字符串的长度
strlen key
使用场景说明
A、作为缓存
缓存一个数据库的it_user表,有两种方式
1、缓存关键字段 ,key设计技巧:
set it_user:userid:1:name name_value
2、缓存整行,表+id作为key
set it_user:1 it_user的序列化
#java实现
// 定义键
userRedisKey = "it_user:" + id;
// 从 Redis 获取值
value = redis.get(userRedisKey);
if (value != null) {
// 将值进行反序列化为 UserInfo 并返回结果
userInfo = deserialize(value);
return userInfo;
}
B、计数 点赞 播放量等相关功能
incr命令
#java实现
long incrVideoCounter(long id) {
key = "video:playCount:" + id;
return redis.incr(key);
}
1.2、hash
Redis hash 是一个 string 类型的 field 和 value 的映射表,hash 特别适合用于存储对象。Redis 中每个 hash 可以存储 232 - 1 键值对(40多亿)。hash的底层实现是压缩表或者hashtables(字典)。
常用命令
#设置值
hset key field value
#获取值
hget key field
#删除field
hdel key field1 field2
#计算field的个数
hlen key
#批量设置
hmset key field1 value1 field2 value2
hmget key field1 field2
#判断field是否存在
hexists key field
#获取所有的field
hkeys key
#获取所有的values
hvals key
1.3、list
列表,用来存储多个有序的字符串,列表中的每一个字符串称之为元素,一个列表最多可以存储2 32-1个元素。底层使用压缩表或者双端链表实现。常用命令
#从右边插入
rpush key value
#从左边插入
lpush key value
获取指定范围内的元素列表
lrange key start end
#获取列表中指定索引下表的元素
lindex key index
#获取列表的长度
llen key
#从左侧弹出
lpop key
#从右侧弹出
rpop key
#阻塞弹出
blpop key1 key2 timeout
brpop key1 key2 timeout
使用场景
消息队列可以使用lpush + brpop实现。
1.4、set
用来保存多个的字符串元素,但和列表类型不一样的是,集合中不允许有重复元素,并且集合中的元素是无序的,不能通过索引下标获取元素。
底层使用的是整数集合或者字典实现的集合对象。
常用命令
#添加元素
sadd key elements
sadd key values1 values2
#删除元素
srem key elements
#计算元素的个数
scard key
#判断元素是否在集合中
simember key values
#随机从集合中返回指定个数元素
srandmember key count
#从集合中随机弹出元素
spop key
#列出所有的元素
smember key
#求元素的交接
sinter key1 key2
# 多个集合求并集
suinon key1 key2
# 多个集合的差集
sdiff key1 key2
使用场景
1、给用户增加标签
2、抽奖,将人员信息录入,随机pop出来一个实现抽奖功能
3、社交类型的需求,比如共同好友,好友推荐等
1.5、zset
有序集合(zset)*不能有重复的元素,而且还可以排序,它和列表使用索引下标作为排序依据不同的是,它给每个元素设置一个分数(score)作为排序的依据,底层使用chang用了压缩表,跳跃表和字典实现。
常用命令
#添加用户
zadd key score member
#计算成员个数
zcard key
#计算某一个成员的分数,
zscore key member
#列出成员的排名
zrank key member
zrevrank key member
#删除成员
zrem key member
#增加某个成员的分数
zincrby key increment member
#返回指定排名范围内成员
zrange key start end
zrevrange key start end (从高到低)
#返回指定分数范围内的成员个数
zcount key min max
2、redis的高可用
2.1、主从复制
redis的主从模式,使用异步复制,slave节点异步从master节点复制数据,master节点提供读写服务,slave节点只提供读服务(这个是默认配置,可以通过修改配置文件 slave-read-only 控制)。master节点可以有多个从节点。配置一个slave节点只需要在redis.conf文件中指定 slaveof master-ip master-port 即可。
2.2、哨兵模式
哨兵模式是一种特殊的模式,首先Redis提供了哨兵的命令,哨兵是一个独立的进程,作为进程,它会独立运行。其原理是哨兵通过发送命令,等待Redis服务器响应,从而监控运行的多个Redis实例。
这里的哨兵有两个作用
- 通过发送命令,让Redis服务器返回监控其运行状态,包括主服务器和从服务器。
- 当哨兵监测到master宕机,会自动将slave切换成master,然后通过发布订阅模式通知其他的从服务器,修改配置文件,让它们切换主机。
然而一个哨兵进程对Redis服务器进行监控,可能会出现问题,为此,我们可以使用多个哨兵进行监控。各个哨兵之间还会进行监控,这样就形成了多哨兵模式。
用文字描述一下故障切换(failover)的过程。假设主服务器宕机,哨兵1先检测到这个结果,系统并不会马上进行failover过程,仅仅是哨兵1主观的认为主服务器不可用,这个现象成为主观下线。当后面的哨兵也检测到主服务器不可用,并且数量达到一定值时,那么哨兵之间就会进行一次投票,投票的结果由一个哨兵发起,进行failover操作。切换成功后,就会通过发布订阅模式,让各个哨兵把自己监控的从服务器实现切换主机,这个过程称为客观下线。这样对于客户端而言,一切都是透明的。
2.3、集群模式
Redis3.0版本之前,可以通过Redis Sentinel(哨兵)来实现高可用 ( HA ),从3.0版本之后,官方推出了Redis Cluster,它的主要用途是实现数据分片(Data Sharding),不过同样可以实现HA,是官方当前推荐的方案。
Redis cluster在设计的时候,就考虑到了去中心化,去中间件,也就是说,集群中的每个节点都是平等的关系,都是对等的,每个节点都保存各自的数据和整个集群的状态。每个节点都和其他所有节点连接,而且这些连接保持活跃,这样就保证了我们只需要连接集群中的任意一个节点,就可以获取到其他节点的数据。
那么Redis 是如何合理分配这些节点和数据的呢?
Redis 集群没有并使用传统的一致性哈希来分配数据,而是采用另外一种叫做哈希槽 (hash slot)
的方式来分配的。redis cluster 默认分配了 16384 个slot,当我们set一个key 时,会用CRC16
算法来取模得到所属的slot
,然后将这个key 分到哈希槽区间的节点上,具体算法就是:CRC16(key) % 16384
。
3、redis的持久化
3.1、RDB
RDB持久化方式是通过快照(snapshotting)完成的,当符合一定条件时,redis会自动将内存中所有数据以二进制方式生成一份副本并存储在硬盘上。当redis重启时,并且AOF持久化未开启时,redis会读取RDB持久化生成的二进制文件(默认名称dump.rdb,可通过设置dbfilename修改)进行数据恢复,对于持久化信息可以用过命令“info Persistence”查看。
我们可以通过命令查看RDB快照文件位置
config get dbfilename
config get dir
RDB触发条件
- 客户端执行命令save和bgsave会生成快照;
- 根据配置文件save m n规则进行自动快照;
- 主从复制时,从库全量复制同步主库数据,此时主库会执行bgsave命令进行快照;
- 客户端执行数据库清空命令FLUSHALL时候,触发快照;
- 客户端执行shutdown关闭redis时,触发快照;
save命令触发
客户端执行save命令,该命令强制redis执行快照,这时候redis处于阻塞状态,不会响应任何其他客户端发来的请求,直到RDB快照文件执行完毕,所以请慎用
bgsave命令触发
bgsave命令可以理解为background save即:“后台保存”。当执行bgsave命令时,redis会fork出一个子进程来执行快照生成操作,需要注意的redis是在fork子进程这个简短的时间redis是阻塞的(此段时间不会响应客户端请求,),当子进程创建完成以后redis响应客户端请求。其实redis自动快照也是使用bgsave来完成的。
save m n 规则触发
save m n规则说明:在指定的m秒内,redis中有n个键发生改变,则自动触发bgsave。
flushall触发
flushall命令用于清空数据库,当我们使用了则表明我们需要对数据进行清空,那redis当然需要对快照文件也进行清空,所以会触发bgsave。
shutdown触发
shutdown命令触发就不用说了,redis在关闭前处于安全角度将所有数据全部保存下来,以便下次启动会恢复。
主从复制
在redis主从复制中,从节点执行全量复制操作,主节点会执行bgsave命令,并将rdb文件发送给从节点。
3.2、AOF持久化
当redis存储非临时数据时,为了降低redis故障而引起的数据丢失,redis提供了AOF(Append Only File)持久化,从单词意思讲,将命令追加到文件。AOF可以将Redis执行的每一条写命令追加到磁盘文件(appendonly.aof)中,在redis启动时候优先选择从AOF文件恢复数据。由于每一次的写操作,redis都会记录到文件中,所以开启AOF持久化会对性能有一定的影响,但是大部分情况下这个影响是可以接受的,我们可以使用读写速率高的硬盘提高AOF性能。与RDB持久化相比,AOF持久化数据丢失更少,其消耗内存更少(RDB方式执行bgsve会有内存拷贝)。
开启AOF
config get appendonly
config set appendonly yes
# 同步到配置文件 不用重启redis AOF就可以直接生效
config rewrite
redisAOF持久化过程可分为以下阶段:
1.追加写入
redis将每一条写命令以redis通讯协议添加至缓冲区aof_buf,这样的好处在于在大量写请求情况下,采用缓冲区暂存一部分命令随后根据策略一次性写入磁盘,这样可以减少磁盘的I/O次数,提高性能。
2.同步命令到硬盘
当写命令写入aof_buf缓冲区后,redis会将缓冲区的命令写入到文件,redis提供了三种同步策略,由配置参数appendfsync决定,下面是每个策略所对应的含义:
- no:不使用fsync方法同步,而是交给操作系统write函数去执行同步操作,在linux操作系统中大约每30秒刷一次缓冲。这种情况下,缓冲区数据同步不可控,并且在大量的写操作下,aof_buf缓冲区会堆积会越来越严重,一旦redis出现故障,数据丢失严重。
- always:表示每次有写操作都调用fsync方法强制内核将数据写入到aof文件。这种情况下由于每次写命令都写到了文件中, 虽然数据比较安全,但是因为每次写操作都会同步到AOF文件中,所以在性能上会有影响,同时由于频繁的IO操作,硬盘的使用寿命会降低。
- everysec:数据将使用调用操作系统write写入文件,并使用fsync每秒一次从内核刷新到磁盘。 这是折中的方案,兼顾性能和数据安全,所以redis默认推荐使用该配置。
3.文件重写(bgrewriteaof)
当开启的AOF时,随着时间推移,AOF文件会越来越大,当然redis也对AOF文件进行了优化,即触发AOF文件重写条件(后续会说明)时候,redis将使用bgrewriteaof对AOF文件进行重写。这样的好处在于减少AOF文件大小,同时有利于数据的恢复。
为什么重写?比如先后执行了“set foo bar1 set foo bar2 set foo bar3” 此时AOF文件会记录三条命令,这显然不合理,因为文件中应只保留“set foo bar3”这个最后设置的值,前面的set命令都是多余的,下面是一些重写时候策略:
- 重复或无效的命令不写入文件
- 过期的数据不再写入文件
- 多条命令合并写入(当多个命令能合并一条命令时候会对其优化合并作为一个命令写入,例如“RPUSH list1 a RPUSH list1 b” 合并为“RPUSH list1 a b” )
重写的触发条件
AOF文件触发条件可分为手动触发和自动触发:
手动触发:客户端执行bgrewriteaof命令。
自动触发:自动触发通过以下两个配置协作生效:
- auto-aof-rewrite-min-size: AOF文件最小重写大小,只有当AOF文件大小大于该值时候才可能重写,4.0默认配置64mb。
- auto-aof-rewrite-percentage:当前AOF文件大小和最后一次重写后的大小之间的比率等于或者等于指定的增长百分比,如100代表当前AOF文件是上次重写的两倍时候才重写。
redis开启在AOF功能开启的情况下,会维持以下三个变量
- 记录当前AOF文件大小的变量aof_current_size。
- 记录最后一次AOF重写之后,AOF文件大小的变量aof_rewrite_base_size。
- 增长百分比变量aof_rewrite_perc。
每次当serverCron(服务器周期性操作函数)函数执行时,它会检查以下条件是否全部满足,如果全部满足的话,就触发自动的AOF重写操作:
- 没有BGSAVE命令(RDB持久化)/AOF持久化在执行;
- 没有BGREWRITEAOF在进行;
- 当前AOF文件大小要大于server.aof_rewrite_min_size的值;
- 当前AOF文件大小和最后一次重写后的大小之间的比率等于或者大于指定的增长百分比(auto-aof-rewrite-percentage参数)
AOF的本质
AOF实现本质是基于redis通讯协议,将命令以纯文本的方式写入到文件中。
首先Redis是以行来划分,每行以\r\n行结束。每一行都有一个消息头,消息头共分为5种分别如下:
- (+) 表示一个正确的状态信息,具体信息是当前行+后面的字符。
- (-) 表示一个错误信息,具体信息是当前行-后面的字符。
- () 表示消息体总共有多少行,不包括当前行,后面是具体的行数。
- ($) 表示下一行数据长度,不包括换行符长度\r\n,$后面则是对应的长度的数据。
- (:) 表示返回一个数值,:后面是相应的数字节符。
3.3、RDB 和AOF的优缺点总结
RDB
优点:
- RDB 是一个非常紧凑(compact)的文件,体积小,因此在传输速度上比较快,因此适合灾难恢复。
- RDB 可以最大化 Redis 的性能:父进程在保存 RDB 文件时唯一要做的就是
fork
出一个子进程,然后这个子进程就会处理接下来的所有保存工作,父进程无须执行任何磁盘 I/O 操作。 - RDB 在恢复大数据集时的速度比 AOF 的恢复速度要快。
缺点:
- RDB是一个快照过程,无法完整的保存所以数据,尤其在数据量比较大时候,一旦出现故障丢失的数据将更多。
- 当redis中数据集比较大时候,RDB由于RDB方式需要对数据进行完成拷贝并生成快照文件,fork的子进程会耗CPU,并且数据越大,RDB快照生成会越耗时。
- RDB文件是特定的格式,阅读性差,由于格式固定,可能存在不兼容情况。
AOF
优点:
- 数据更完整,秒级数据丢失(取决于设置fsync策略)。
- 兼容性较高,由于是基于redis通讯协议而形成的命令追加方式,无论何种版本的redis都兼容,再者aof文件是明文的,可阅读性较好。
缺点:
- 数据文件体积较大,即使有重写机制,但是在相同的数据集情况下,AOF文件通常比RDB文件大。
- 相对RDB方式,AOF速度慢于RDB,并且在数据量大时候,恢复速度AOF速度也是慢于RDB。
- 由于频繁地将命令同步到文件中,AOF持久化对性能的影响相对RDB较大,但是对于我们来说是可以接受的。
4、java使用redis
4.1、jedis单次连接
建立连接
import redis.clients.jedis.Jedis;
public class RedisJava {
public static void main(String[] args) {
//连接本地的 Redis 服务
Jedis jedis = new Jedis("localhost");
System.out.println("连接成功");
//查看服务是否运行
System.out.println("服务正在运行: "+jedis.ping());
}
}
操作字符串
import redis.clients.jedis.Jedis;
public class RedisStringJava {
public static void main(String[] args) {
//连接本地的 Redis 服务
Jedis jedis = new Jedis("localhost");
System.out.println("连接成功");
//设置 redis 字符串数据
jedis.set("runoobkey", "www.runoob.com");
// 获取存储的数据并输出
System.out.println("redis 存储的字符串为: "+ jedis.get("runoobkey"));
}
}
操作list
import java.util.List;
import redis.clients.jedis.Jedis;
public class RedisListJava {
public static void main(String[] args) {
//连接本地的 Redis 服务
Jedis jedis = new Jedis("localhost");
System.out.println("连接成功");
//存储数据到列表中
jedis.lpush("site-list", "Runoob");
jedis.lpush("site-list", "Google");
jedis.lpush("site-list", "Taobao");
// 获取存储的数据并输出
List<String> list = jedis.lrange("site-list", 0 ,2);
for(int i=0; i<list.size(); i++) {
System.out.println("列表项为: "+list.get(i));
}
}
}
4.2、使用连接池
public class RedisUtil {
//服务器IP地址
private static String ADDR = "ip";
//端口
private static int PORT = 6379;
//密码
private static String AUTH = "123456";
//连接实例的最大连接数
private static int MAX_ACTIVE = 1024;
//控制一个pool最多有多少个状态为idle(空闲的)的jedis实例,默认值也是8。
private static int MAX_IDLE = 200;
//等待可用连接的最大时间,单位毫秒,默认值为-1,表示永不超时。如果超过等待时间,则直接抛出JedisConnectionException
private static int MAX_WAIT = 10000;
//连接超时的时间
private static int TIMEOUT = 10000;
// 在borrow一个jedis实例时,是否提前进行validate操作;如果为true,则得到的jedis实例均是可用的;
private static boolean TEST_ON_BORROW = true;
private static JedisPool jedisPool = null;
//数据库模式是16个数据库 0~15
public static final int DEFAULT_DATABASE = 0;
/**
* 初始化Redis连接池
*/
static {
try {
JedisPoolConfig config = new JedisPoolConfig();
config.setMaxTotal(MAX_ACTIVE);
config.setMaxIdle(MAX_IDLE);
config.setMaxWaitMillis(MAX_WAIT);
config.setTestOnBorrow(TEST_ON_BORROW);
jedisPool = new JedisPool(config, ADDR, PORT, TIMEOUT,AUTH,DEFAULT_DATABASE);
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 获取Jedis实例
*/
public synchronized static Jedis getJedis() {
try {
if (jedisPool != null) {
Jedis resource = jedisPool.getResource();
System.out.println("redis--服务正在运行: "+resource.ping());
return resource;
} else {
return null;
}
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/***
*
* 释放资源
*/
public static void returnResource(final Jedis jedis) {
if(jedis != null) {
jedisPool.returnResource(jedis);
}
}
}
4.3、springboot项目使用redis
maven坐标:
<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-data-redis -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
<version>2.1.8.RELEASE</version>
</dependency>
在配置文件中配置相关redis的信息
#redis配置
#Redis服务器地址
spring.redis.host=127.0.0.1
#Redis服务器连接端口
spring.redis.port=6379
#Redis数据库索引(默认为0)
spring.redis.database=0
#连接池最大连接数(使用负值表示没有限制)
spring.redis.jedis.pool.max-active=50
#连接池最大阻塞等待时间(使用负值表示没有限制)
spring.redis.jedis.pool.max-wait=3000
#连接池中的最大空闲连接
spring.redis.jedis.pool.max-idle=20
#连接池中的最小空闲连接
spring.redis.jedis.pool.min-idle=2
#连接超时时间(毫秒)
spring.redis.timeout=5000
使用redisTemplate操作redis,不过可以封装成一个工具类方便实用
package com.xcbeyond.springboot.redis;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
@Component
public class RedisUtils {
@Autowired
private RedisTemplate<String, String> redisTemplate;
/**
* 读取缓存
*
* @param key
* @return
*/
public String get(final String key) {
return redisTemplate.opsForValue().get(key);
}
/**
* 写入缓存
*/
public boolean set(final String key, String value) {
boolean result = false;
try {
redisTemplate.opsForValue().set(key, value);
result = true;
} catch (Exception e) {
e.printStackTrace();
}
return result;
}
/**
* 更新缓存
*/
public boolean getAndSet(final String key, String value) {
boolean result = false;
try {
redisTemplate.opsForValue().getAndSet(key, value);
result = true;
} catch (Exception e) {
e.printStackTrace();
}
return result;
}
/**
* 删除缓存
*/
public boolean delete(final String key) {
boolean result = false;
try {
redisTemplate.delete(key);
result = true;
} catch (Exception e) {
e.printStackTrace();
}
return result;
}
}