redis入门

本文大部份内容出自狂神说的教学视频，地址：https://www.bilibili.com/video/BV1S54y1R7SB

问题出现的原因

1、数据类型太多

2、数据源太多

3、数据要改造，大面积改造？

阻塞I/O、非阻塞I/O何多路复用I/O

1、阻塞I/O就是一直等待数据返回的I/O

2、非阻塞I/O就是通过不断地轮询去查看数据是否已经成功返回，如果没有返回就会得到一个错误，提示你过会再来。

3、多路复用I/O就是一个线程同时监听多个文件是否有数据到来，如果有数据就通知相应的线程来处理。多路复用是指使用一个线程来检查多个文件描述符（Socket）的就绪状态，比如调用select和poll函数，传入多个文件描述符（FileDescription，简称FD），如果有一个文件描述符（FileDescription）就绪，则返回，否则阻塞直到超时。得到就绪状态后进行真正的操作可以在同一个线程里执行，也可以启动线程执行（比如使用线程池）

NoSQL的四大分类

KV键值对

新浪：Redis
美团：Redis+Tair
阿里、百度：Redis+memecache

文档型数据库（bson格式和json一样）

MongoDB
- MongoDB是一个基于分布式文件存储的数据库，C++编写，主要用来处理大量的文档
- MongoDB是一个介于关系型数据库和非关系型数据库中间的产品！MongoDB是非关系数据库中功能最丰富，最像关系型数据库。

列储存数据库

HBase
分布式文件系统

图数据库

朋友圈社交网站，广告推荐 (Neo4j)

Redis能干嘛？（www.redis.cn）

内存存储、持久化、内存中是断电即失、所以说持久化很重要
效率高、可以用于高速缓存
发布订阅系统
地图信息分析
计数器、计时器

特性

多样的数据类型
持久化
集群
事务
…….

Linux安装

1、下载安装解压

1 2	wget http://download.redis.io/releases/redis-6.0.6.tar.gz tar zxvf

进入解压后的文件，可以看到配置文件

2、基本的环境安装：安装gcc，

安装：yum install gcc-C++

升级gcc: 
sudo yum install centos-release-scl
sudo yum install devtoolset-7-gcc*
scl enable devtoolset-7 bash

3、进入redis文件夹，使用make编译

4、开启redis

5、关闭redis

1	查看ip：ps -ef\| grep redis

1 2	shutdown quit

测试redis-benchmark

1	redis-benchmark -h localhost -p 6379 -c 100 -n 100000

Redis基础

redis默认16和数据库，默认使用第0个，可以使用select切换。

#切换数据库
[root@VM-0-6-centos bin]# redis-cli -p 6379
127.0.0.1:6379> ping
PONG
127.0.0.1:6379> select 16
(error) ERR DB index is out of range
127.0.0.1:6379> select 15
OK
127.0.0.1:6379[15]>

keys * #查看所有的key
EXISTS name #判断当前的key是否存在
EXPIRE name 10 #设置key的过期时间，单位为秒
ttl name #查看当前key的剩余时间
set name hth
APPEND name "hello" 在name后面追加hello
get name 
type name #查看name是什么类型

⭐redis基础命令：http://www.redis.cn/commands.html

redis是一个开源的，内存中的数据结构存储系统，它可以用作数据库，缓存和消息中间件。

五大基础类型

1、String（字符串）

2、List

所有的list命令都是以L开头的

![](2021-03-26-redis入门/json.png)127.0.0.1:6379> Lpush list one
(integer) 1
127.0.0.1:6379> keys *
1) "list"
127.0.0.1:6379> lpush list two #头部插入
(integer) 2
127.0.0.1:6379> lpush list three
(integer) 3
127.0.0.1:6379> keys *
1) "list"
127.0.0.1:6379> get list
(error) WRONGTYPE Operation against a key holding the wrong kind of value
127.0.0.1:6379> lget list
(error) ERR unknown command `lget`, with args beginning with: `list`, 
127.0.0.1:6379> lRANGE list 0 1
1) "three"
2) "two"
127.0.0.1:6379> lRANGE list 0 -1
1) "three"
2) "two"
3) "one"
127.0.0.1:6379> RPUSH list right  #尾部插入
(integer) 4
127.0.0.1:6379> lRANGE list 0 -1 
1) "three"
2) "two"
3) "one"
4) "right"
127.0.0.1:6379> RPUSH list righr
(integer) 5
127.0.0.1:6379> lRANGE list 0 -1
1) "three"
2) "two"
3) "one"
4) "right"
5) "righr"

127.0.0.1:6379> LRANGE list 0 -1
1) "three"
2) "two"
3) "one"
4) "right"
5) "righr"
127.0.0.1:6379> LPOP list 
"three"
127.0.0.1:6379> LRANGE list 0 -1
1) "two"
2) "one"
3) "right"
4) "righr"
127.0.0.1:6379> RPOP list 
"righr"
127.0.0.1:6379> LRANGE list 0 -1
1) "two"
2) "one"
3) "right"
127.0.0.1:6379> LINDEX list 1
"one"
127.0.0.1:6379>

127.0.0.1:6379> LRANGE list 0 -1
1) "three"
2) "two"
3) "one"
127.0.0.1:6379> LREM list two
(error) ERR wrong number of arguments for 'lrem' command
127.0.0.1:6379> LREM list 1 two         #####################LREM 卸载list列表1个值为two的元素
(integer) 1
127.0.0.1:6379> LRANGE list 0 -1
1) "three"
2) "one"
127.0.0.1:6379> LPUSH list two
(integer) 3
127.0.0.1:6379> LRANGE list 0 -1
1) "two"
2) "three"
3) "one"
127.0.0.1:6379> LPUSH list two
(integer) 4
127.0.0.1:6379> LRANGE list 0 -1
1) "two"
2) "two"
3) "three"
4) "one"
127.0.0.1:6379> LREM list 2 two
(integer) 2
127.0.0.1:6379> LRANGE list 0 -1
1) "three"
2) "one"
127.0.0.1:6379>

####   trim:截断
127.0.0.1:6379> RPUSH list "hello"
(integer) 1
127.0.0.1:6379> RPUSH list "fdkjkdfd"
(integer) 2
127.0.0.1:6379> RPUSH list "sdk2ee2"
(integer) 3
127.0.0.1:6379> LPUSH list "0093"
(integer) 4
127.0.0.1:6379> LRANGE list 0 -1
1) "0093"
2) "hello"
3) "fdkjkdfd"
4) "sdk2ee2"
127.0.0.1:6379> Ltrim list 0 1
OK
127.0.0.1:6379> LRANGE list 0 -1
1) "0093"
2) "hello"
127.0.0.1:6379>

### rpoplpush #移除列表的最后一个元素并将它移动到新的列表中
127.0.0.1:6379> RPUSH list 0
(integer) 1
127.0.0.1:6379> RPUSH list 1
(integer) 2
127.0.0.1:6379> RPUSH list 2
(integer) 3
127.0.0.1:6379> LRANGE list 0 -1
1) "0"
2) "1"
3) "2"
127.0.0.1:6379> rpoplpush list otherlist
"2"
127.0.0.1:6379> LRANGE list 0 -1
1) "0"
2) "1"
127.0.0.1:6379> LRANGE otherlist 0 -1
1) "2"
127.0.0.1:6379> lset list 0 erjer   #### 将列表中制定下标的值替换为另外一个值
127.0.0.1:6379> LINSERT list before 000 1212131231 ### 将某个具体的value插入到列表中某个元素的前面或者后面

==list可以实现消息队列，栈==

3、SET

==set是无序、值唯一的==

127.0.0.1:6379> sadd myset "mam"   ### 添加
(integer) 1
127.0.0.1:6379> smembers myset
1) "love"
2) "mam"
3) "hello"
127.0.0.1:6379> sadd myset "mam"  
(integer) 0
127.0.0.1:6379> smembers myset    ### 获取set所有值
1) "love"
2) "mam"
3) "hello"
127.0.0.1:6379> sismember myset mam   ### 查询某个值是否存在
(integer) 1
127.0.0.1:6379> sismember myset ma    ### 查询某个值是否存在
(integer) 0
127.0.0.1:6379> srandmember myset    ### 随机获取set集合中的某个元素
"hello"
127.0.0.1:6379> srandmember myset
"love"
127.0.0.1:6379> srandmember myset
"mam"
127.0.0.1:6379> srandmember myset
"mam"
127.0.0.1:6379> srandmember myset 2   ### 随机获取set集合中的两个元素
1) "hello"
2) "mam"

127.0.0.1:6379> smembers myset   
1) "love"
2) "mam"
3) "hello"
127.0.0.1:6379> smembers myset2
1) "love"
2) "mam"
3) "we"
127.0.0.1:6379> Sdiff myset myset2   ## 差集
1) "hello" 
127.0.0.1:6379> Sinter myset myset2  ##交集 共同好友可以用交集实现
1) "love"
2) "mam"
127.0.0.1:6379> Sunion myset myset2  ##并集
1) "love"
2) "we"
3) "hello"
4) "mam"

==set集合可以实现共同关注，推荐好友之类的功能。==

4、Hash（哈希）集合

==map集合，<key,map> - - <key,<key,map>>==

==hash本质和String没有太大的区别，还是一个简单的key-value。hash更加适合对象的存储，String更加适合字符串存储==

######## 设置myhash ###########
127.0.0.1:6379> hget myhash field1
"gp"
127.0.0.1:6379> hset myhash field1 "mam"
(integer) 0
127.0.0.1:6379> hget myhash field1
"mam"

####### mset 和 mget ：多个值的同时设置与获取  ######
127.0.0.1:6379> hmset myhash field1 "mam" field2 "hth" field3 "sjj"  
OK
127.0.0.1:6379> hmget myhash field1 field2 field3  
1) "mam"
2) "hth"
3) "sjj"
127.0.0.1:6379> hmget myhash field1 field2 field4
1) "mam"
2) "hth"
3) (nil)

#### 删除指定的字段key，对应的value值也会消失
127.0.0.1:6379> hdel myhash field1     
(integer) 1
127.0.0.1:6379> hmget myhash field1 field2 field3
1) (nil)
2) "hth"
3) "sjj"
127.0.0.1:6379> 

###
127.0.0.1:6379> hlen myhash   ##获取hash的字段数量（即hash的长度）
(integer) 2
127.0.0.1:6379> hgetall myhash  ##获取hash的所有元素
1) "field2"
2) "hth"
3) "field3"
4) "sjj"
127.0.0.1:6379> hkeys myhash  ##获取所有的字段
1) "field2"
2) "field3"
127.0.0.1:6379> hvals myhash  ##获取所有的value值
1) "hth"
2) "sjj"
127.0.0.1:6379> HEXISTS myhash field2  ##判断hash中某个字段是否存在
(integer) 1

### incr做增减
127.0.0.1:6379> hset myhash filed5 4
(integer) 1 
127.0.0.1:6379> hincrby myhash filed5 2
(integer) 6
127.0.0.1:6379> hget myhash filed5
"6"
127.0.0.1:6379> hincrby myhash  filed5 -2
(integer) 4
127.0.0.1:6379> hget myhash filed5
"4"

127.0.0.1:6379> hsetnx myhash field2 tt
(integer) 0
127.0.0.1:6379> hget myhash field2
"hth"
127.0.0.1:6379> hsetnx myhash field223 tt
(integer) 1
127.0.0.1:6379> hget myhash field223
"tt"

5、Zset（有序集合）

（排行？？？）

==在set的基础上增加了一个排序值。==

127.0.0.1:6379> zadd myset 1 one
(integer) 1
127.0.0.1:6379> zadd myset 2 two 3 three
(integer) 2
127.0.0.1:6379> zRange myset 0 -1
1) "one"
2) "two"
3) "three"
  
####排序如何实现的？？？ 
127.0.0.1:6379> Zrangebyscore myset 0 2
1) "one"
2) "two"
127.0.0.1:6379> Zrangebyscore myset 2 0
(empty array)
127.0.0.1:6379> zrevrangebyscore myset 2 0
1) "two"
2) "one"

127.0.0.1:6379> Zrangebyscore myset 0 2 withscores  ###升序排列同时显示内容
1) "one"
2) "1"
3) "two"
4) "2"

127.0.0.1:6379> zrange myset 0 -1
1) "one"
2) "two"
3) "three"
127.0.0.1:6379> zrem myset two   ##移除有序集合中的指定元素
(integer) 1
127.0.0.1:6379> zrange myset 0 -1
1) "one"
2) "three"

127.0.0.1:6379> zcard myset   ## 获取有序集合中的个数
(integer) 2

127.0.0.1:6379> zcount myset 2 3    ## 获取区间[2，3]之间的数据个数
(integer) 1
127.0.0.1:6379> zcount myset 2 5    ## 获取区间[2，5]之间的数据个数
(integer) 1

三种特殊的数据类型

geospatital地理位置 http://www.redis.cn/commands/geoadd.html

geoadd – 添加地理位置（经度、维度、名称）** geoadd china:city 120.16 30.24235 hangzhou

geodist

geohash

geopos

georadius

georadiusbymember

Hyperlog log

什么是基数？– 数据集里不重复的元素

Hyperloglog

优点：占用的内存是固定的，2^64元素的基数只需要占用12kb的内存。

127.0.0.1:6379> PFADD mysey as sdas sd asd w fg ty sd sd
(integer) 1
127.0.0.1:6379> PFCOUNT mysey
(integer) 7
127.0.0.1:6379> PFADD mykey1 as as asf
(integer) 1
127.0.0.1:6379> PFADD mykey2 asd2s 2as asf
(integer) 1
127.0.0.1:6379> PFMERGE mysey mykey1 mykey2   ###合并集合mykey1 和 mykey2 到mysey
OK
127.0.0.1:6379> PFCOUNT mysey
(integer) 10
127.0.0.1:6379> PFCOUNT mykey1
(integer) 2
127.0.0.1:6379> PFCOUNT mykey2
(integer) 3

Bitmaps（位存储）

统计数量之类的

各种各样的实时分析

存储与对象id关联的存储高效且高性能的布尔信息

例如，假设您想知道网站用户按天计的最长的(连续)访问记录。从零开始计算天数，这是您将web站点公开的日子，并在每次用户访问web站点时设置SETBIT。至于比特位索引，只需取当前unix时间戳，减去初始偏移量，然后除以3600*24。通过这种方式，每个用户都有一个包含每天访问信息的小字符串。使用BITCOUNT可以很容易地获得指定用户访问网站的天数，而使用一些BITPOS调用，或者简单地获取和分析bitmap，可以很容易地计算最长的连续字符。https://www.jianshu.com/p/4cbfbed312bc

###设置打卡，记录一周打卡的次数；或者查看某天是否打卡之类的
127.0.0.1:6379> setbit sign 0 1
(integer) 0
127.0.0.1:6379> setbit sign 1 1
(integer) 0
127.0.0.1:6379> setbit sign 2 0
(integer) 0
127.0.0.1:6379> setbit sign 3 1
(integer) 0
127.0.0.1:6379> setbit sign 4 1
(integer) 0
127.0.0.1:6379> setbit sign 5 1
(integer) 0
127.0.0.1:6379> setbit sign 6 1
(integer) 0
127.0.0.1:6379> setbit sign 7 0
(integer) 0
127.0.0.1:6379>

Redis事务

简介

redis一个事务中所有的命令都会被序列化，在事务的执行过程中，会按照顺序执行。

一次性、顺序性、排他性

redis事务不保证原子性，也没有隔离级别的概念。

所有的命令在事务中们并没有直接执行，只有发起执行命令的时候才会被执行。Exec

redis的事务

开启事务

命令入队

执行事务

正常执行事务（multi – exec）

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> set k2 v2
QUEUED
127.0.0.1:6379> exec
1) OK
2) OK
127.0.0.1:6379> mget k1 k2
1) "v1"
2) "v2"

放弃事务 (discard)

127.0.0.1:6379> mget k1 k2
1) "v1"
2) "v2"
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k3 v3 
QUEUED
127.0.0.1:6379> set k4 v4
QUEUED
127.0.0.1:6379> discard
OK
127.0.0.1:6379> mget k3 k4
1) (nil)
2) (nil)

编译时异常（代码有问题，命令有错，事务中所有命令都不会被执行）

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> set k2 v2
QUEUED
127.0.0.1:6379> set k3 v3
QUEUED
127.0.0.1:6379> getset key3       ####错误的命令
(error) ERR wrong number of arguments for 'getset' command
127.0.0.1:6379> set k4 v4
QUEUED
127.0.0.1:6379> set k5 v5
QUEUED
127.0.0.1:6379> exec             #### 执行事务抱错
(error) EXECABORT Transaction discarded because of previous errors.
127.0.0.1:6379> get k5
(nil)
127.0.0.1:6379> keys *           #### 所有的命令都不会执行
(empty array)

运行时异常（代码正确，逻辑有错，那么除开有问题的命令，其余命令都会被执行，即保证单条原子性）

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> incr k1   ### 错误的指令
QUEUED
127.0.0.1:6379> set k2 v2
QUEUED
127.0.0.1:6379> exec
1) OK
2) (error) ERR value is not an integer or out of range
3) OK
127.0.0.1:6379> keys *   ##所以redis不保证原子性，正确的语句依然可以执行
1) "k2"
2) "k1"

监控（redis可以通过watch的方式来实现乐观锁）

悲观锁：无论做什么都加锁，极其悲观

乐观锁：无论做什么都加锁，极其乐观（233333）

1、获取version

2、更新的时候比较version

127.0.0.1:6379> set cu 900
OK
127.0.0.1:6379> watch cu
OK
127.0.0.1:6379> decrby cu 20
(integer) 880

使用watch可以在redis中实现乐观锁操作。

通过unwatch解锁

上面两张图说明了watch监视的key会被锁住，可以通过unwatch解锁

SpringBoot集成Redis

前言

SpringBoot操作数据：通过spring-data实现

说明：在springboot2.x之后，原来使用的jedis替换成了lettuce

jedis：采用直连，多个线程操作，是不安全的，如果想要避免不安全，需要使用jedis pool连接池！面向BIO；

lettuce：采用netty，实例可以在多个线程中共享，不存在线程不安全的情况，面向NIO。

@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RedisOperations.class)
@EnableConfigurationProperties(RedisProperties.class)
@Import({ LettuceConnectionConfiguration.class, JedisConnectionConfiguration.class })
public class RedisAutoConfiguration {

	@Bean
	@ConditionalOnMissingBean(name = "redisTemplate") // 我们可以自定义RedisTemplate来替换这个默认的Template
	public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory)
			throws UnknownHostException {
        // 默认的RedisTemplate没有过多的设置，redis对象都需要序列化
        // 两个泛型都是Object的类型，使用时需要强制转换，类似<String,Object>
		RedisTemplate<Object, Object> template = new RedisTemplate<>();
		template.setConnectionFactory(redisConnectionFactory);
		return template;
	}

	@Bean
	@ConditionalOnMissingBean
    //大部分情况我们都使用String类型，所以单独实现了StringRedisTemplate
	public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory)
			throws UnknownHostException {
		StringRedisTemplate template = new StringRedisTemplate();
		template.setConnectionFactory(redisConnectionFactory);
		return template;
	}

}

SpringBoot整合Redis流程：

1、导入依赖

2、配置连接

## springboot所有的配置类，都有一个自动配置类 ：RedisAutoCOnfiguration
## 自动配置类都会绑定一个properties配置文件：RedisProperties

#配置redis
spring.redis.host = 42.192.138.89
spring.redis.port = 6379

3、测试！

1
2
3

//注意需要修改redis.cong
// 1、注释bind:127:0:0:1    ⭐bind限制redis只能本地连接，修改后就可以远程连接
// 2、修改peotected mode为no

序列化

默认使用jdk的序列化，可以根据自己的需求自己配置，例如使用 Jackson序列化方式

⭐⭐编写自己的RedisTemplate

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;


/**
 * 固定模板，可以直接使用
 * 
 * @author hth
 * @date 2021-04-01-17:20
 */

@Configuration
public class RedisConfig {

    //编写自己的 RedisTemplate
    @Bean
    public RedisTemplate<String,Object> redisTemplate(RedisConnectionFactory redisConnectionFactory){
        RedisTemplate<String,Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);

        //Jackson序列化配置
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer =  new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);

        //String的序列化
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();

        //key采用string的序列化方式
        template.setKeySerializer(stringRedisSerializer);
        //hash序列化的key也采用String的序列化方式
        template.setHashKeySerializer(stringRedisSerializer);
        //Value采用Jackson序列化方式
        template.setValueSerializer(jackson2JsonRedisSerializer);
        //hash的value采用jackson
        template.setHashKeySerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet();

        return template;
    }
}

工具类

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

@Component
public final class RedisUtils {

    @Autowired
    @Qualifier("redisTemplate")
    private RedisTemplate<String, Object> redisTemplate;

    // =============================common============================
    /**
     * 指定缓存失效时间
     * @param key  键
     * @param time 时间(秒)
     */
    public boolean expire(String key, long time) {
        try {
            if (time > 0) {
                redisTemplate.expire(key, time, TimeUnit.SECONDS);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 根据key 获取过期时间
     * @param key 键 不能为null
     * @return 时间(秒) 返回0代表为永久有效
     */
    public long getExpire(String key) {
        return redisTemplate.getExpire(key, TimeUnit.SECONDS);
    }


    /**
     * 判断key是否存在
     * @param key 键
     * @return true 存在 false不存在
     */
    public boolean hasKey(String key) {
        try {
            return redisTemplate.hasKey(key);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 删除缓存
     * @param key 可以传一个值 或多个
     */
    public long del(Collection<String> key) {
        Long ret = redisTemplate.delete(key);
        return ret == null ? 0 : ret;
    }


    // ============================String=============================

    /**
     * 普通缓存获取
     * @param key 键
     * @return 值
     */
    public Object get(String key) {
        return key == null ? null : redisTemplate.opsForValue().get(key);
    }

    /**
     * 普通缓存放入
     * @param key   键
     * @param value 值
     * @return true成功 false失败
     */

    public boolean set(String key, Object value) {
        try {
            redisTemplate.opsForValue().set(key, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 普通缓存放入并设置时间
     * @param key   键
     * @param value 值
     * @param time  时间(秒) time要大于0 如果time小于等于0 将设置无限期
     * @return true成功 false 失败
     */

    public boolean set(String key, Object value, long time) {
        try {
            if (time > 0) {
                redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
            } else {
                set(key, value);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 递增
     * @param key   键
     * @param delta 要增加几(大于0)
     */
    public long incr(String key, long delta) {
        if (delta < 0) {
            throw new RuntimeException("递增因子必须大于0");
        }
        return redisTemplate.opsForValue().increment(key, delta);
    }


    /**
     * 递减
     * @param key   键
     * @param delta 要减少几(小于0)
     */
    public long decr(String key, long delta) {
        if (delta < 0) {
            throw new RuntimeException("递减因子必须大于0");
        }
        return redisTemplate.opsForValue().increment(key, -delta);
    }


    // ================================Map=================================

    /**
     * HashGet
     * @param key  键 不能为null
     * @param item 项 不能为null
     */
    public Object hget(String key, String item) {
        return redisTemplate.opsForHash().get(key, item);
    }

    /**
     * 获取hashKey对应的所有键值
     * @param key 键
     * @return 对应的多个键值
     */
    public Map<Object, Object> hmget(String key) {
        return redisTemplate.opsForHash().entries(key);
    }

    /**
     * HashSet
     * @param key 键
     * @param map 对应多个键值
     */
    public boolean hmset(String key, Map<String, Object> map) {
        try {
            redisTemplate.opsForHash().putAll(key, map);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * HashSet 并设置时间
     * @param key  键
     * @param map  对应多个键值
     * @param time 时间(秒)
     * @return true成功 false失败
     */
    public boolean hmset(String key, Map<String, Object> map, long time) {
        try {
            redisTemplate.opsForHash().putAll(key, map);
            if (time > 0) {
                expire(key, time);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 向一张hash表中放入数据,如果不存在将创建
     *
     * @param key   键
     * @param item  项
     * @param value 值
     * @return true 成功 false失败
     */
    public boolean hset(String key, String item, Object value) {
        try {
            redisTemplate.opsForHash().put(key, item, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 向一张hash表中放入数据,如果不存在将创建
     *
     * @param key   键
     * @param item  项
     * @param value 值
     * @param time  时间(秒) 注意:如果已存在的hash表有时间,这里将会替换原有的时间
     * @return true 成功 false失败
     */
    public boolean hset(String key, String item, Object value, long time) {
        try {
            redisTemplate.opsForHash().put(key, item, value);
            if (time > 0) {
                expire(key, time);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 删除hash表中的值
     *
     * @param key  键 不能为null
     * @param item 项 可以使多个 不能为null
     */
    public void hdel(String key, Object... item) {
        redisTemplate.opsForHash().delete(key, item);
    }


    /**
     * 判断hash表中是否有该项的值
     *
     * @param key  键 不能为null
     * @param item 项 不能为null
     * @return true 存在 false不存在
     */
    public boolean hHasKey(String key, String item) {
        return redisTemplate.opsForHash().hasKey(key, item);
    }


    /**
     * hash递增 如果不存在,就会创建一个 并把新增后的值返回
     *
     * @param key  键
     * @param item 项
     * @param by   要增加几(大于0)
     */
    public double hincr(String key, String item, double by) {
        return redisTemplate.opsForHash().increment(key, item, by);
    }


    /**
     * hash递减
     *
     * @param key  键
     * @param item 项
     * @param by   要减少记(小于0)
     */
    public double hdecr(String key, String item, double by) {
        return redisTemplate.opsForHash().increment(key, item, -by);
    }


    // ============================set=============================

    /**
     * 根据key获取Set中的所有值
     * @param key 键
     */
    public Set<Object> sGet(String key) {
        try {
            return redisTemplate.opsForSet().members(key);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    /**
     * 根据value从一个set中查询,是否存在
     *
     * @param key   键
     * @param value 值
     * @return true 存在 false不存在
     */
    public boolean sHasKey(String key, Object value) {
        try {
            return redisTemplate.opsForSet().isMember(key, value);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 将数据放入set缓存
     *
     * @param key    键
     * @param values 值 可以是多个
     * @return 成功个数
     */
    public long sSet(String key, Object... values) {
        try {
            return redisTemplate.opsForSet().add(key, values);
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }


    /**
     * 将set数据放入缓存
     *
     * @param key    键
     * @param time   时间(秒)
     * @param values 值 可以是多个
     * @return 成功个数
     */
    public long sSetAndTime(String key, long time, Object... values) {
        try {
            Long count = redisTemplate.opsForSet().add(key, values);
            if (time > 0)
                expire(key, time);
            return count;
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }


    /**
     * 获取set缓存的长度
     *
     * @param key 键
     */
    public long sGetSetSize(String key) {
        try {
            return redisTemplate.opsForSet().size(key);
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }


    /**
     * 移除值为value的
     *
     * @param key    键
     * @param values 值 可以是多个
     * @return 移除的个数
     */

    public long setRemove(String key, Object... values) {
        try {
            Long count = redisTemplate.opsForSet().remove(key, values);
            return count;
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }

    // ===============================list=================================

    /**
     * 获取list缓存的内容
     *
     * @param key   键
     * @param start 开始
     * @param end   结束 0 到 -1代表所有值
     */
    public List<Object> lGet(String key, long start, long end) {
        try {
            return redisTemplate.opsForList().range(key, start, end);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    /**
     * 获取list缓存的长度
     *
     * @param key 键
     */
    public long lGetListSize(String key) {
        try {
            return redisTemplate.opsForList().size(key);
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }


    /**
     * 通过索引 获取list中的值
     *
     * @param key   键
     * @param index 索引 index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推
     */
    public Object lGetIndex(String key, long index) {
        try {
            return redisTemplate.opsForList().index(key, index);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    /**
     * 将list放入缓存
     *
     * @param key   键
     * @param value 值
     */
    public boolean lSet(String key, Object value) {
        try {
            redisTemplate.opsForList().rightPush(key, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 将list放入缓存
     * @param key   键
     * @param value 值
     * @param time  时间(秒)
     */
    public boolean lSet(String key, Object value, long time) {
        try {
            redisTemplate.opsForList().rightPush(key, value);
            if (time > 0)
                expire(key, time);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }

    }


    /**
     * 将list放入缓存
     *
     * @param key   键
     * @param value 值
     * @return
     */
    public boolean lSet(String key, List<Object> value) {
        try {
            redisTemplate.opsForList().rightPushAll(key, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }

    }


    /**
     * 将list放入缓存
     *
     * @param key   键
     * @param value 值
     * @param time  时间(秒)
     * @return
     */
    public boolean lSet(String key, List<Object> value, long time) {
        try {
            redisTemplate.opsForList().rightPushAll(key, value);
            if (time > 0)
                expire(key, time);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 根据索引修改list中的某条数据
     *
     * @param key   键
     * @param index 索引
     * @param value 值
     * @return
     */

    public boolean lUpdateIndex(String key, long index, Object value) {
        try {
            redisTemplate.opsForList().set(key, index, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 移除N个值为value
     *
     * @param key   键
     * @param count 移除多少个
     * @param value 值
     * @return 移除的个数
     */

    public long lRemove(String key, long count, Object value) {
        try {
            Long remove = redisTemplate.opsForList().remove(key, count, value);
            return remove;
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }

    }

}

redis.conf详解

redis持久化

redis是内存数据库，如果不将内存中的数据库保存到磁盘，就会断电即失。

RDB（Redis DataBase）

在指定的时间间隔内将内存中的数据集快照写入磁盘，也就是snapShot快照，恢复则是将快照文件读到内存中。

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

==RDB保存的文件是dump.rdb==

触发机制：

1、满足save规则；

2、执行flushall命令也会触发；

3、退出redis也会产生rdb文件。

恢复机制：

1、只需要将rdb文件放在redis启动目录即可，redis启动会自动检查dump.rgb文件去恢复数据；在生产环境中，通常需要将dump.rdb文件进行备份。

2、查看需要存在的位置

1
2
3

127.0.0.1:6379> config get dir
1) "dir"
2) "/usr/local/bin"   #如果dump.rdb存在这个目录下，redis启动时候就会自动恢复数据

rdb的优点:

1、适合大规模的数据恢复；

rdb的缺点：

1、需要一定的时间间隔，如果redis意外宕机了，最后一次修改的数据就没有了；

2、需要占用一定的空间

AOF（Append Only File）

将所有的命令都记录下来，恢复的时候将这个文件中的命令全部执行一遍。

以日志的行驶来记录每个操作，将redis执行过的所有指令全部记录下来（读操作不记录），只需追加文件但不是改写文件，redis启动之初会读取该文件重建数据，如果数据量大执行效率会比较慢。

==文件默认保存为appendonly.aof==

1 2	appendonly no #默认是不开启aof的，大多数情况下使用默认的rdb模式就可以 appendfilename "appendonly.aof" #持久化文件的名字

aof的优点

1、每一次修改都同步，数据恢复的完整性会更加好

2、每秒同步一次，可能会丢失一秒的数据

aof的缺点

1、aof效率远远低于rdb，所以redis默认使用rdb方式

扩展

1、rdb持久化方式能够在指定的时间间隔对你的数据进行快照存储；

2、aof持久化方式记录每次对服务器写的操作，当服务器重启的时候会执行这些命令来恢复原始的数据，aof命令以redis协议追加保存每次写的操作到文件末尾，redis还能对aof文件进行后台重写，使得aof文件的体积不至于过大；

3、只做缓存，如果你希望你的数据只在服务器运行的时候存在，你也可以不适用任何的持久化；

4、同时开启两种持久化方式

①在这种情况下，当redis重启的时候会优先载入aof文件来恢复原始的数据，因为在通常情况下aof文件保存的数据集要比rdb保存的数据集完整；

②rdb的数据不是实时的，同时使用两者时服务器重启也只会找aof文件，那么要不要只启动aof呢？建议不要，因为rdb更加适合用于备份数据库（aof在不断变化不是很好备份），快速重启，而且rdb不会有aof潜在的bug，应该开启rdb作为万无一失的手段。

5、性能建议：

①因为rdb只用作后备用途，建议旨在Slave上持久化rdb文件，而且只需要15分钟一次，即只用保留save 900 1;

②如果Enable AOF，好处是在最恶劣的情况下也只会丢失不超过两秒的数据，启动脚本简单只需要load自己的aof文件即可，代价一是带来了持续的IO，二是AOF rewite的最后将rewrite过程中产生的新数据写到新文件造成的阻塞几乎是不能避免的。只要硬盘允许，应该将尽量减少aof rewite的操作，即将aof默认的文件大小设置到5g以上（默认的64m太小了）。

③如果不Enable AOF，仅靠Master-Slave Repllcation实现高可用也可以，能省掉一大笔IO，也减少了rewrite时带来的系统波动。代价是如果Maste/Slave同时挂掉，会丢失十几分钟的数据，启动脚本也要比较两个Master/Slave中的rdb文件，载入较新的那个。

Redis发布订阅

Redis发布订阅（pub/sub）是一种消息通信模式：发布者(pub)发布消息，订阅者(sub)接收消息。

redis客户端可以订阅任何数量的频道。

订阅端：

127.0.0.1:6379> subscribe kuangshenshuo      #订阅一个频道
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "kuangshenshuo"

####等待接收消息
3) (integer) 1
1) "message"
2) "kuangshenshuo"
3) "hellO "
1) "message"
2) "kuangshenshuo"
3) "redis"

发送端：

127.0.0.1:6379> publish kuangshenshuo "hellO"      # 发布者发布消息到频道
(integer) 1
127.0.0.1:6379> publish kuangshenshuo "redis"
(integer) 1

应用

1、实时聊天系统

2、实时聊天室（多人订阅）

3、订阅，关注系统

稍微复杂的场景需要使用消息中间件：MQ，Kafka等

redis主从复制

通过持久化功能，Redis保证了即使在服务器重启的情况下也不会丢失（或少量丢失）数据，但是由于数据是存储在一台服务器上的，如果这台服务器出现故障，比如硬盘坏了，也会导致数据丢失。为了避免单点故障，我们需要将数据复制多份部署在多台不同的服务器上，即使有一台服务器出现故障其他服务器依然可以继续提供服务。这就要求当一台服务器上的数据更新后，自动将更新的数据同步到其他服务器上，这时候就用到了Redis的主从复制。

默认情况下，每台Redis服务器都是主节点；且一个主节点可以有多个从节点（或者没有从节点），但一个从节点有且只能有一个主节点。主从复制中数据的复制是单向的，只能由主节点到从节点。Master以写为主，Slave以读为主。

主从复制的作用：

1、数据冗余（热备份）

2、故障恢复

3、负载均衡（即读写分离，主节点负责写，从节点负责读）

4、高可用模式的基石（主从复制是哨兵和集群能够实施的基础）

主从复制，读写分离！80%的情况下都是在进行读操作，减缓服务器的压力！redis集群至少需要一主二从。

环境配置

通过配置文件配置

127.0.0.1:6379> info replication   ##查看redis的相关信息
# Replication
role:master
connected_slaves:0
master_replid:bbd62a9867a106235c3b142559bcaa41320ad11d
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:0
second_repl_offset:-1
repl_backlog_active:0
repl_backlog_size:1048576
repl_backlog_first_byte_offset:0
repl_backlog_histlen:0

复制原理：全量复制 & 增量复制

哨兵(sentinel)模式

哨兵模式是一种特殊的模式，它能够监听主机是否故障，如果故障了哨兵将选取一台从机做新的主机。

故障切换（failover）的过程。假设主服务器宕机，哨兵1先检测到这个结果，系统并不会马上进行failover过程，仅仅是哨兵1主观的认为主服务器不可用，这个现象成为主观下线。当后面的哨兵也检测到主服务器不可用，并且数量达到一定值时，那么哨兵之间就会进行一次投票，投票的结果由一个哨兵发起，进行failover操作。切换成功后，就会通过发布订阅模式，让各个哨兵把自己监控的从服务器实现切换主机，这个过程称为客观下线。这样对于客户端而言，一切都是透明的。

配置

1、在Redis安装目录下有一个sentinel.conf文件，copy一份进行修改

# 配置监听的主服务器，这里sentinel monitor代表监控，mymaster代表服务器的名称，可以自定义，192.168.11.128代表监控的主服务器，6379代表端口，2代表只有两个或两个以上的哨兵认为主服务器不可用的时候，才会进行failover操作。
sentinel monitor mymaster 192.168.11.128 6379 2
# sentinel author-pass定义服务的密码，mymaster是服务名称，123456是Redis服务器密码
# sentinel auth-pass <master-name> <password>
sentinel auth-pass mymaster 123456

2、启动哨兵

1	redis-sentinel /hthconfig/sentinel.conf

Redis缓存穿透和雪崩

缓存穿透

缓存雪崩

Redis在项目中的具体应用

数据类型	使用场景
String	比如想要封锁一个ip地址（某段时间内访问及其频繁），使用incrby记录次数
Hash	存储用户信息【id，name，age】
List	实现最新消息的排行；可以用来模拟消息队列利用push将命令储存在集合中，同时使用pop从集合中取出命令
Set	特殊之处：可以自动排除重复。可以储存好友信息之类的，这样求两个人的共同好友的操作就可以只需要求交集即可
Zset	以某一个条件为权重进行排序。电商网站商品排名，可以根据多条件的权重来实现。