[大数据]Redis-01-Nosql概述

一、Nosql的出现

单机mysql

在90年代，网络发展的还不是那么快，一个网站的访问量不大，单个数据库足够使用了。实现的业务也简单。

但随着时间推移，上网的人越来越多，网站慢慢的出现了一些问题:

1.数据量达到一定程度，单机无法存放那么多的数据

2.数据的索引(B+ Tree)，一个机器内存无法存放。

3.访问量变大后(读写混合)，一台服务器承受不住。

Memcached(缓存)+Mysql+垂直拆分(读写分离)

在大多数情况下，查询数据库都是在读，为了减轻数据库的负担，可以使用缓存来保证效率。

优化过程经历以下几个阶段:

1.优化数据库的数据结构和索引(难度大)

2.文件缓存，通过IO流获取比每次都访问数据库效率略高，但面对流量爆炸式增长时，IO流无法承受这股流量冲击。

3.MemCache，当时热门的技术，通过在数据库和数据库访问层添加一层缓存，第一次访问时查询数据库，把查询结果放入缓存，后续的查询先检查缓存，若有直接从缓冲中获取使用，效率明显提升。

分库分表+水平拆分+Mysql集群

当今年代

如今信息量海量，各种各样的数据出现（用户定位数据，图片数据等），大数据的背景下关系型数据库(RDBMS)已无法满足大量数据要求。Nosql数据库能轻松解决这些问题。

目前基本的互联网项目

为什么要用NoSQL?

用户的个人信息，社交网络，地理位置。用户自己产生的数据，用户日志等等爆发式增长。这时就需要使用NoSQL数据库，NoSQL数据库可以很好的解决以上问题。

二、什么是Nosql

NoSQL = Not Only SQL（不仅仅是SQL）

Not Only Structured Query Language

关系型数据库: 列+行，同一个表下数据的结构是一样的。

非关系型数据库: 数据存储没有固定的格式，并且可以进行横向拓展。

NoSQL泛指非关系型数据库，随着web2.0互联网的诞生，传统关系型数据库很难应付web2.0时代，暴露出来很多难以克服的问题，进而NoSQL在大数据环境下得到了发展，Redis是发展最快的。

Nosql的特点

1. 方便拓展(数据之间没有关系，很好拓展)
2. 大数据量高性能(Redis一秒可以写8万次，读11万,NoSQL缓存记录级，是一种细粒度的缓存，性能会比较高!)
3. 数据类型丰富，多样化，（不需要事先设计数据库，随取随用）
4. 传统的RDBMS和NoSQL

传统的RDBM（关系型数据库）
- 结构化组织
- SQL
- 数据和关系都存在单独的表中
- 操作，数据定义语言
- 严格的一致性
- 基础的事务等等

NoSQL（非关系型数据库）
- 不仅仅是数据
- 没有固定的查询语言
- 键值对存储，列存储，文档存储，图形数据库(社交关系)
- 最终一致性
- CAP定力和BASE
- 高性能，高拓展，高可用

了解: 3V+3高

大数据时代的3V：主要是用于描述问题的

1.海量 Velume

2.多样Variety

3.实时Velocity

大数据时代的3高: 主要是对程序的要求

1.高并发

2.高拓展

3.高性能

真正的实践是将Nosql和RDBMS联合起来一起使用。

三、阿里巴巴演进分析

推荐阅读:阿里云的这群疯子https://yq.aliyun.com/articles/653511

#商品信息

一般存放在关系型数据库，Mysql,但阿里巴巴使用的Mysql是主机内部改动的。

#商品描述，评论(文字居多)

- 文档性数据库: MongDB

# 图片

- 分布式文件系统: FastDFS
- 淘宝: TFS
- Google: GFS
- Hadoop: HDFS
- 阿里云: oos

# 商品关键字，用于搜索

- 搜索引擎: solr,elasticsearch
- 阿里: Isearch 多隆

# 商品热门的波段信息

- 内存数据库: Redis、Memcache

#商品交易,外部支付接口
- 第三方应用

四、Nosql四大分类

KV键值对

• 新浪: Redis
• 美团: Redis+Tair
• 阿里、百度: Redis+Memcache

文档型数据库(bjson数据格式)

MongDB(掌握)

• 基于分布式文件存储的数据库。C++编写，用于处理大量文档。

• MongDB是RDBMS和NoSQL的中间产品。MongDB是非关系型数据库功能最丰富的，在NOSQL中最像关系型数据库的数据库。

• ConthDB

列存储数据库

• HBase(大数据必学)
• 分布式文件系统

图关系数据库

用于广告推荐，社交网络

• Neo4j、InfoGrid

五、Redis入门

概述

Redis（Remote Dictionary Server）,即远程字典服务。

是一个开源的使用ANSI C语言编写、支持网络、可基于内存也可持久化的日志性、Key-Value数据库，并提供多种语言的API。

它通常被称为数据结构存储器，因为值（value）可以是 字符串(String), 哈希(Map), 列表(list), 集合(sets) 和 有序集合(sorted sets)等类型。

与memcache一样，为了保证效率，数据都是缓存在内存中。区别的是redis会周期性的把更新的数据写入磁盘或者修改操作写入的记录文件，并且在此基础上实现了master-slave(主从同步)。

Redis能干什么?

1. 内存存储、持久化，内存是断电即失的，所以需要持久化(RDB、AOF)
2. 高效率，用于高速缓存
3. 发布订阅系统
4. 地图信息分析
5. 计时器，计数器(eg: 浏览量)
6. 。。。

特性

1.多样的数据类型

2.持久化

3.集群

4.事务

六、Redis环境搭建

官网:https://redis.io/
中文官网:https://www.redis.net.cn/

Redis安装分为Windows安装和Linux安装
Windows版本的Redis停更很久了！

Windows安装

下载地址:https://github.com/dmajkic/redis

这里到狂神的公众号下载:链接: https://pan.baidu.com/s/1a0mjGhZVzfyUCdsK4PeVug 提取码: t63c

别忘了给狂神点赞三连!

1.解压安装包

2.开启redis-server.exe

3.启动redis-cli.exe测试

Linux安装

1.下载安装包 。 redis-5.0.8.tar.gz

2.把安装包丢入到远程服务器ftp上，并解压文件

3.基本环境安装

yum install gcc-c++
# 然后进入redis目录下执行
make
# 然后执行
make install

4.解压后将redis移动到 /usr/local/bin

5.将redis的配置文件复制到/usr/local/bin/lconfig下

6.修改redis后台运行,默认不是后台启动的

7.通过制定的配置文件启动redis服务。

8.使用redis-cli连接指定的端口号测试,Redis的默认端口号为6379
在这里插入图片描述

9.查看redis进程是否开启

10.关闭Redis服务 shutdown

11.再次查看进程是否存在

七、测试性能

redis-benchmark:官方性能测试工具

redis性能测试工具可选参数如下:

简单测试：

测试：100个并发连接 10000请求


基础知识

redis默认有16个数据库

默认使用第0个

16个数据库为: DB 0~DB 15
默认使用DB 0,可以使用select n 切换到DB n,dbsize 可以查看当前数据库大小，与key数量相关。

不同数据源之间的数据是互不通的，并且dbsize是根据库中key的个数。

命令说明:

• keys * ：查看当前数据库中所有的key。

• flushdb：清空当前数据库中的键值对。

• flushall：清空所有数据库的键值对。

Redis是单线程的，Redis是基于内存操作的

所以Redis的性能瓶颈不是CPU,而是机器内存和网络带宽。
那么为什么Redis的速度如此快呢，性能这么高?QPS达到10W+

Redis为什么单线程还这么快?

• 误区1: 高性能的服务器一定是多线程的?
• 误区2: 多线程(CPU上下文会切换!)一定是比单线效率高!

核心: Redis是将所有数据放在内存中的，所以说使用单线程去操作效率就是最高的，多次读写都是在一个CPU上的，在内存存储数据情况下，单线程就是最佳的方案!

八、五大数据类型

Redis是一个开源(BSD许可)，内存存储的数据结存储，可用作数据库，高速缓存以及消息中间件。它支持字符串，哈希表，列表，集合，有序集合，位图，hyperloglogs等数据类型。

Redis内置复制、Lua脚本、LRU收回、事务以及不同级别磁盘持久化功能，同时通过Redis Sentinel（哨兵模式）提高高可用，通过Redis Cluster(集群)提供自动分区。

Redis-key

在redis无论是什么数据类型,都可以使用key-value形式保存，通过进行Redis-key的操作，来完成对数据库中数据的操作。

下面学习的命令有：

• exists key: 判断键是否存在
• del key:删除键值对
• move key db: 将键值对移动到指定数据库
• expire key second: 设置键值对过期时间
• type key: 查看value的数据类型
• ttl key :查看你当前键值对的存活剩余时间

关于TTL的命令

Redis的key,通过TTL命令返回key的过期时间，一般来说有三种。

1. 当前key没有设置过期时间，所以返回为key
2. 当前key有设置过期时间，且key已经过期，会返回-2
3. 当前key有设置过期时间，且key没有过期，会返回key的剩余存活时间。

关于重命名RENAME和RENAMENX

1. RENAME key newkey 修改key的名称
2. RENAMENX key neewkey仅当newkey不存在key改名为呢我开始用。

更多命令学习:查看以下官方命令文章:

https://www.redis.net.cn/order/

1.String(字符串)

演示

127.0.0.1:6379> set k1 hello
OK
127.0.0.1:6379> APPEND k1 liang
(integer) 10
127.0.0.1:6379> INCR k1
(error) ERR value is not an integer or out of range
127.0.0.1:6379> set k2 1
OK
127.0.0.1:6379> INCR k2
(integer) 2
127.0.0.1:6379> get k2
"2"
127.0.0.1:6379> DECR k2
(integer) 1
127.0.0.1:6379> get k2
"1"
127.0.0.1:6379> INCRBY k2 2
(integer) 3
127.0.0.1:6379> get k2
"3"
127.0.0.1:6379> STRLEN k1
(integer) 10
127.0.0.1:6379> GETRANGE k1 0 -1
"helloliang"
127.0.0.1:6379> GETRANGE k1 0 2
"hel"
127.0.0.1:6379> SETRANGE k1 2 xx
(integer) 10
127.0.0.1:6379> get k1
"hexxoliang"
127.0.0.1:6379> GETSET k1 hello
"hexxoliang"
127.0.0.1:6379> get k1
"hello"
127.0.0.1:6379> SETNX k4 4
(integer) 1
127.0.0.1:6379> SETEX k4 30 5
OK
127.0.0.1:6379> ttl k4
(integer) 22
127.0.0.1:6379> MSET name1 liang name2 hello
OK
127.0.0.1:6379> MGET name1 name2
1) "liang"
2) "hello"
127.0.0.1:6379> PSETEX name2 6000 hello
OK
127.0.0.1:6379> ttl name2
(integer) 2

String 类似的使用场景：value除了是字符串还可以是数字，用途举例:

• 计算器
• 统计多单位的数量
• 粉丝数
• 对象存储缓存

2.List(列表)

Redis列表是个简单的字符串列表，按照插入顺序排序。你可以添加一个元素到列表的头部(左边)或者尾部(右边)。

一个Redis列表最多可以包含2^32-1个元素(4294967295, 每个列表超过40亿个元素)。

通过插入的顺序规则，我们可以将redis列表定义为队列，栈，双向队列。

通过上图可知Redis列表类型是通过双端操作的，所以命令也分为了LXXX和RXXX两类，可以将L理解为Left,R理解为Right，通过左右来控制列表的插入顺序。

List常见Redis命令如下

演示

127.0.0.1:6379> clear
127.0.0.1:6379> LPUSH list1 1
(integer) 1
127.0.0.1:6379> LPUSH list1 2
(integer) 2
127.0.0.1:6379> RPUSH list1 3
(integer) 3
127.0.0.1:6379> LRANGE list1 0 -1
1) "2"
2) "1"
3) "3"
127.0.0.1:6379> LRANGE list1 0 2
1) "2"
2) "1"
3) "3"
127.0.0.1:6379> LPUSHX list 1
(integer) 0
127.0.0.1:6379> LPUSHX list1 4 5
(integer) 5
127.0.0.1:6379> LRANGE list1 0 -1
1) "5"
2) "4"
3) "2"
4) "1"
5) "3"
127.0.0.1:6379> LINSERT list1 before 4 6
(integer) 6
127.0.0.1:6379> LRANGE list1 0 -1
1) "5"
2) "6"
3) "4"
4) "2"
5) "1"
6) "3"
127.0.0.1:6379> LLEN list1
(integer) 6
127.0.0.1:6379> LINDEX list1 2
"4"
127.0.0.1:6379> LSET list1 2 7
OK
127.0.0.1:6379> LRANGE list 0 -1
(empty list or set)
127.0.0.1:6379> LRANGE list1 0  -1
1) "5"
2) "6"
3) "7"
4) "2"
5) "1"
6) "3"
127.0.0.1:6379> LPOP list1
"5"
127.0.0.1:6379> RPOP list1
"3"
127.0.0.1:6379> RPOPLPUSH list1 newlist
"1"
127.0.0.1:6379> LRANGE list1 0 -1
1) "6"
2) "7"
3) "2"
127.0.0.1:6379> LRANGE newlist 0 -1
1) "1"
127.0.0.1:6379> LTRIM list1 0 1
OK
127.0.0.1:6379> LRANGE list1 0 -1
1) "6"
2) "7"
127.0.0.1:6379> LREM list1 1 6
(integer) 1
127.0.0.1:6379> LRANGE list1 0 -1
1) "7"
127.0.0.1:6379> BLPOP newlist list1 30
1) "newlist"
2) "1"
127.0.0.1:6379> LRANGE newlist 0 -1
(empty list or set)
127.0.0.1:6379> BLPOP list 30
(nil)
(30.09s)

小结

• list 本质上是一个链表，拥有Before Node After, left , right可以插入值
• 如果key不存在，则创建新链表
• 如果key存在，新增内容
• 如果移除了所有值，空链表，也代表不存在。
• 和链表一样，两边插入或改动元素的值，效率高，中间插入或改动元素，效率比较低。

应用场景:
消息队列(Lpush Rpop),栈（Lpush Lpop）

3.Set(集合)

Redis的Set是string类型的无序集合。集合成员是唯一的，这意味着集合中不能出现重复的数据。

Redis中集合是通过哈希表实现的，所以添加，删除，查找的复杂度都是O(1)。集合中最大的成员数为2^32-1，和List的数目一样。

演示

127.0.0.1:6379> SADD myset m1 m2 m3 m4
(integer) 4
127.0.0.1:6379> SCARD myset
(integer) 4
127.0.0.1:6379> smembers myset
1) "m4"
2) "m3"
3) "m2"
4) "m1"
127.0.0.1:6379> SISMEMBER smember m1
(integer) 0
127.0.0.1:6379> SISMEMBER myset m1
(integer) 1
127.0.0.1:6379> SRANDMEMBER myset 3
1) "m3"
2) "m1"
3) "m2"
127.0.0.1:6379> SRANDMEMBER myset
"m3"
127.0.0.1:6379> SPOP myset 1
1) "m4"
127.0.0.1:6379> SMOVE myset newset m3
(integer) 1
127.0.0.1:6379> SMEMBERS myset
1) "m2"
2) "m1"
127.0.0.1:6379> SMEMBERS newset
1) "m3"
127.0.0.1:6379> SADD setx m1 m2 m4 m6
(integer) 4
127.0.0.1:6379> SADD sety m2 m5 m6
(integer) 3
127.0.0.1:6379> SADD setz m1 m3 m6
(integer) 3
127.0.0.1:6379> SDIFF setx sety setz
1) "m4"
127.0.0.1:6379> SINTER setx sety setz
1) "m6"
127.0.0.1:6379> SUNION setx sety setz
1) "m6"
2) "m2"
3) "m1"
4) "m5"
5) "m3"
6) "m4"

4.Hash(哈希)

Redis hash 是一个string类型的field和value的映射表，hash特别适合用于存储对象。

Set就是一种简化的Hash，只变动key,而value使用默认值填充。可以将一个Hash表作为一个对象进行存储，表中存放对象的信息。

演练

127.0.0.1:6379> HSET student name liang
(integer) 1
127.0.0.1:6379> HSET student age 18
(integer) 1
127.0.0.1:6379> HGET student name
"liang"
127.0.0.1:6379> HMGET student name age
1) "liang"
2) "18"
127.0.0.1:6379> HSET student age 20
(integer) 0
127.0.0.1:6379> HSETNX student email 1457902738@qq.com
(integer) 1
127.0.0.1:6379> HEXISTS student age
(integer) 1
127.0.0.1:6379> HEXISTS student tel
(integer) 0
127.0.0.1:6379> HEXISTS student tel
(integer) 0
127.0.0.1:6379> HEXISTS student name
(integer) 1
127.0.0.1:6379> clear
127.0.0.1:6379> HGETALL student
1) "name"
2) "liang"
3) "age"
4) "20"
5) "email"
6) "1457902738@qq.com"
127.0.0.1:6379> HKEYS student
1) "name"
2) "age"
3) "email"
127.0.0.1:6379> HVALS student
1) "liang"
2) "20"
3) "1457902738@qq.com"
127.0.0.1:6379> HGETALL student
1) "name"
2) "liang"
3) "age"
4) "20"
5) "email"
6) "1457902738@qq.com"
127.0.0.1:6379> HKEYS student
1) "name"
2) "age"
3) "email"
127.0.0.1:6379> HVALS studnet
(empty list or set)
127.0.0.1:6379> HVALS student
1) "liang"
2) "20"
3) "1457902738@qq.com"
127.0.0.1:6379> clear
127.0.0.1:6379> HLEN student
(integer) 3
127.0.0.1:6379> HLEN studen
(integer) 0
127.0.0.1:6379> HLEN student
(integer) 3
127.0.0.1:6379> clear
127.0.0.1:6379> HKEYS student
1) "name"
2) "age"
3) "email"
127.0.0.1:6379> HDEL student email
(integer) 1
127.0.0.1:6379> HKEYS student
1) "name"
2) "age"
127.0.0.1:6379> HINCRBY student age 2
(integer) 22
127.0.0.1:6379> HINCRBY key name 2
(integer) 2
127.0.0.1:6379> HINCRBY student name 2
(error) ERR hash value is not an integer
127.0.0.1:6379> HINCRBY student weight 120
(integer) 120
127.0.0.1:6379> HVALS student
1) "liang"
2) "22"
3) "120"
127.0.0.1:6379> 

小结

Hash变更的数据 user name age, 尤其是用户信息之类的，经常变动的信息！Hash更适合于对象的存储，String 更适合字符串的存储!

5.Zset(有序集合)

不同的是每个元素都有关联一个double类型的分数，redis正是通过分数来为集合中的成员进行从小到大的排序。

score相同: 按字典顺序排序
有序集合的成员是唯一的，但分数(score)却可以重复。

演示:

127.0.0.1:6379> ZADD zset 1 m1 2 m2 3 m3 4 m4
(integer) 4
127.0.0.1:6379> ZCARD zset
(integer) 4
127.0.0.1:6379> ZCOUNT zset 0 1
(integer) 1
127.0.0.1:6379> ZINCRBY zset 2 m2
"4"
127.0.0.1:6379> ZSCORE zset m1
"1"
127.0.0.1:6379> ZSCORE zset m2
"4"
127.0.0.1:6379> ZRANK zset m1
(integer) 0
127.0.0.1:6379> ZRANK zset m2
(integer) 2
127.0.0.1:6379> ZRANGE zset 0 1
1) "m1"
2) "m3"
127.0.0.1:6379> zRANGE zset 0 -1
1) "m1"
2) "m3"
3) "m2"
4) "m4"
127.0.0.1:6379> ZRANGEBYLEX zset - +
1) "m1"
2) "m3"
3) "m2"
4) "m4"
127.0.0.1:6379> ZRANGEBYLEX zset - + LIMIT 0 3 
1) "m1"
2) "m3"
3) "m2"
127.0.0.1:6379> ZRANGEBYLEX zset - + LIMIT 3 3
1) "m4"
127.0.0.1:6379> ZRANGE zset (- [m2
(error) ERR value is not an integer or out of range
127.0.0.1:6379> ZRANGEBYLEX zset (- [m2
1) "m1"
127.0.0.1:6379> ZRANGEBYLEX zset [m1 [m4
1) "m1"
2) "m3"
3) "m2"
4) "m4"
127.0.0.1:6379> ZRANGEBYSCORE zset 1 10
1) "m1"
2) "m3"
3) "m2"
4) "m4"
127.0.0.1:6379> ZLEXCOUNT zset - +
(integer) 4
127.0.0.1:6379> ZLEXCOUNT zset [m1 [m3
(integer) 3
127.0.0.1:6379> ZREM zset m1 
(integer) 1
127.0.0.1:6379> ZREMRANGEBYLEX key [m2 [m3
(integer) 0
127.0.0.1:6379> ZREMRANGEBYLEX key [m3 [m2
(integer) 0
127.0.0.1:6379> ZREMRANGEBYLEX zset [m2 [m4
(integer) 3
127.0.0.1:6379> ZCARD zset
(integer) 0
127.0.0.1:6379> ZADD zset 1 m1 2 m2 3 m3 4 m4
(integer) 4
127.0.0.1:6379> ZREMRANGEBYRANK zset 0 1
(integer) 2
127.0.0.1:6379> ZREMRANGEBYSCORE zset 0 3
(integer) 1
127.0.0.1:6379> ZRANGE zset 0 -1
1) "m4"
127.0.0.1:6379> ZADD zset 1 m1 2 m2 3 m3
(integer) 3
127.0.0.1:6379> ZREVRANGE zset 0 3
1) "m4"
2) "m3"
3) "m2"
4) "m1"
127.0.0.1:6379> ZREVRANGE zset 2 4
1) "m2"
2) "m1"
127.0.0.1:6379> ZREVRANGEBYSCORE zset  3 1
1) "m3"
2) "m2"
3) "m1"
127.0.0.1:6379> ZREVRANGEBYLEX zset [m4 [m1
1) "m4"
2) "m3"
3) "m2"
4) "m1"
127.0.0.1:6379> ZREVRANK myset m3
(nil)
127.0.0.1:6379> ZREVRANK zset m3
(integer) 1
127.0.0.1:6379> ZADD mathscore 90 xm  95 xh 87 xg
(integer) 3
127.0.0.1:6379> ZADD enscore 70 xm 93 xh 90 xg
(integer) 3
127.0.0.1:6379> ZINTERSTORE sumscore 2 mathscore enscore
(integer) 3
127.0.0.1:6379> ZRANGE sumscore 0 -1 withscores
1) "xm"
2) "160"
3) "xg"
4) "177"
5) "xh"
6) "188"
127.0.0.1:6379> ZUNIONSTORE lowestscore 2 mathscore enscore AGGREGATE MIN 
(integer) 3
127.0.0.1:6379> ZRANGE lowestscore  0 -1 withscores
1) "xm"
2) "70"
3) "xg"
4) "87"
5) "xh"
6) "93"

应用案例:

• set排序 存储班级成绩表 工资表排序!
• 普通消息 1.重要消息 2.带权重判断
• 排行榜应用实现,取TOP N测试!

九、三种特殊数据类型

1. Geospatial(地理位置)

Geospatial是使用经纬度定位地理坐标并用一个有序集合zset保存，所以zset也可以使用

有效经纬度

• 有效的经度从-180度到180度。
• 有效的纬度从-85.05112878度到85.05112878度。

有效参数单位

• m 表示单位为米。

• km 表示单位为千米。

• mi 表示单位为英里。

• ft 表示单位为英尺。

关于GEORADIUS的参数

通过georadius就可以完成 附近的人功能

withcoord:带上坐标

withdist:带上距离，单位与半径单位相同

COUNT n : 只显示前n个(按距离递增排序)

演示

127.0.0.1:6379> geoadd china:city 116.40 39.90 beijing
(integer) 1
127.0.0.1:6379> geoadd china:city 121.47 31.23 shanghai
(integer) 1
127.0.0.1:6379> geoadd china:city 112.98 28.19 changsha
(integer) 1
127.0.0.1:6379> geoadd china:city 113.28 23.12 guangzhou
(integer) 1
127.0.0.1:6379> geoadd china:city 120.15 30.29 hangzhou
(integer) 1
127.0.0.1:6379> geoadd china:city 114.08 22.55 shenzhen
(integer) 1
127.0.0.1:6379> geopos china:city shenzhen
1) 1) "114.08000081777572632"
   2) "22.5500010475923105"
127.0.0.1:6379> geopos china:city beijing
1) 1) "116.39999896287918091"
   2) "39.90000009167092543"
127.0.0.1:6379> geodist china:city shenzhen beijing
"1942148.5325"
127.0.0.1:6379> geodist china:city shenzhen beijing km
"1942.1485"

127.0.0.1:6379> GEORADIUS china:city 110 30 1000 km 
1) "shenzhen"
2) "guangzhou"
3) "changsha"
4) "hangzhou"
127.0.0.1:6379> GEORADIUS china:city 110 30 500 km 
1) "changsha"
127.0.0.1:6379> GEORADIUS china:city 110 30 500 km withdist
1) 1) "changsha"
   2) "352.6997"
127.0.0.1:6379> GEORADIUS china:city 110 30 500 km  WITHCOORD
1) 1) "changsha"
   2) 1) "112.9800000786781311"
      2) "28.1899989603527743"
127.0.0.1:6379> geohash china:city beijing hanghzhou
1) "wx4fbxxfke0"
2) (nil)
127.0.0.1:6379> geohash china:city beijing hangzhou
1) "wx4fbxxfke0"
2) "wtmkq44s1c0"

2.Hyperloglog(基数统计)

Redis HyperLogLog 是用来做基数统计的算法，HyperLogLog 的优点是，在输入元素的数量或者体积非常非常大时，计算基数所需的空间总是固定的、并且是很小的。

花费 12 KB 内存，就可以计算接近 2^64 个不同元素的基数。

因为 HyperLogLog 只会根据输入元素来计算基数，而不会储存输入元素本身，所以 HyperLogLog 不能像集合那样，返回输入的各个元素。其底层使用string数据类型。

什么是基数?

数据集中不重复的元素的个数。

传统实现，存储用户的id,然后每次进行比较。当用户变多之后这种方式及其浪费空间，而我们的目的只是计数，Hyperloglog就能帮助我们利用最小的空间完成。

演示

127.0.0.1:6379> PFCOUNT hyperlog
(integer) 11
127.0.0.1:6379> PFCOUNT hyperlog
(integer) 11
127.0.0.1:6379> PFCOUNT hyperlog
(integer) 11
127.0.0.1:6379> PFMERGE hyperlog hyperlog1 hyperlog2
OK

如果允许容错，那么一定可以使用Hyperloglog !

如果不允许容错，就使用set或者自己的数据类型即可 ！

应用场景：

网页的访问量（UV）：一个用户多次访问，也只能算作一个人。

3.BitMaps（位图）

使用位存储，信息状态只有 0 和 1

Bitmap是一串连续的二进制数据（0或1），每一位所在的位置为偏移(offset)，在bitmap上可执行AND,OR,XOR,NOT以及其它位操作。

应用场景:
签到统计、状态统计

演示

127.0.0.1:6379> clear
127.0.0.1:6379> setbit sign 0 1  #设置sign的第0位为1
(integer) 0
127.0.0.1:6379> setbit sign 2 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> type sign
string
127.0.0.1:6379> getbit sign 1# 获取第1位的数值
(integer) 0
127.0.0.1:6379> getbit sign 2
(integer) 1
127.0.0.1:6379> BITCOUNT sign # 统计sign中为1的位数
(integer) 4

bitmaps的底层:bitmaps是一串从左到右的二进制串

十、Redis事务

Redis的单条命令是保证原子性的，但是redis事务不能保证原子性。

Redis事务本质

是一组命令的集合。事务中每条命令都会被序列化，执行过程中按顺序执行，不允许其他命令进行干扰。

因此有以下特性:

• 一次性
• 排他性
• 顺序性

Redis事务没有隔离级别而言。
Redis单条命令可以保证原子性，但事务不可保证原子性。

Redis事务操作过程:

• 开启事务(multi)
• 命令入队
• 执行事务(exec)

所有的事务命令刚开始加入到事务中并不会被执行，只有提交才会开始执行(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> get k1
QUEUED
127.0.0.1:6379> set k3 v3
QUEUED
127.0.0.1:6379> keys *
QUEUED
127.0.0.1:6379> exec #事务执行
1) OK
2) OK
3) "v1"
4) OK
5) 1) "k2"
   2) "k3"
   3) "k1"

--------------------------------------

#取消事务(discard)
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> DISCARD # 放弃事务
OK
127.0.0.1:6379> EXEC 
(error) ERR EXEC without MULTI #当前未开启事务
127.0.0.1:6379> get k1 #
(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> error k1 #这是一条语法错误命令
(error) ERR unknown command `error`, with args beginning with: `k1`, 
127.0.0.1:6379> get k2
QUEUED
127.0.0.1:6379> EXEC
(error) EXECABORT Transaction discarded because of previous errors. # 执行报错
127.0.0.1:6379> get k1
(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> INCR k1  # 这条命令逻辑错误（对字符串进行增量）
QUEUED
127.0.0.1:6379> get k2
QUEUED
127.0.0.1:6379> EXEC 
1) OK
2) OK
3) (error) ERR value is not an integer or out of range
4) "v2"

# 虽然中间有一条命令报错了，但是后面的指令依旧正常执行成功了。
# 所以说Redis单条指令保证原子性，但是Redis事务不能保证原子性。

十一、Redis监控(乐观锁)

Java领域锁分为两种，乐观锁和悲观锁。

• 悲观锁: 认为什么时候都需要加锁，通过sychoronize或lock锁对象进行加锁。每个线程只有争取到锁才能有权利执行代码。

• 乐观锁: 认为什么时候都不会出现问题，乐观锁不会上锁，而是判断是否在此时间段有无人修改，若修改了则放弃此次操作，接着继续执行，一直等到数据未修改。

• 更新version

• 更新的时候比较version

在Redis中使用watch key监控指定数据，相当于乐观锁加锁。

未被多线程修改情况

127.0.0.1:6379> set money 100
OK
127.0.0.1:6379> set use 0
OK
127.0.0.1:6379> watch money
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> DECRBY money 20
QUEUED
127.0.0.1:6379> INCRBY use 20
QUEUED
127.0.0.1:6379> exec
1) (integer) 80
2) (integer) 20

多线程修改情况

--- 线程1先执行 ---
127.0.0.1:6379> set money 100
OK
127.0.0.1:6379> set use 0
OK
127.0.0.1:6379> watch money
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> DECRBY money 20
QUEUED
127.0.0.1:6379> INCRBY use 20
QUEUED

--- 线程2执行 ---
127.0.0.1:6379> INCRBY money 500
(integer) 600


--- 线程1后执行 ---
127.0.0.1:6379> EXEC
(nil)
127.0.0.1:6379> get money
"600"

解锁获取最新值，然后再进行事务，unwatch可以取消监视。

十二、Jedis

在Java中操作Redis,Jedis是Redis官方推荐的用于Java连接redis客户端。

1.创建一个空项目，创建Maven子模块，添加如下依赖:

<!--导入jredis的包-->
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>3.2.0</version>
</dependency>
<!--fastjson-->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.70</version>
</dependency>

2.编码测试

• 连接数据库，修改redis配置文件:
  1.将保护模式设置为no
  2.允许后台运行
  3.绑定本地注释

3. windows配置redis:打开windows下的redis-server.exe文件即可

4.linux配置redis端口(这个要收费)

若使用公网地址,则做以下配置
得到自己的IP

修改配置文件为自己的IP地址，这样就可以通过IP访问到redis

启动redis-cli时

./bin/redis-cli -h IP地址-p 端口号 -a 密码 

此时使用jedis时，IP配置公网地址即可，不是本机ip地址

连接成功

firewall-cmd --zone = public --add-port = 6379/tcp --permanet

重启防火墙服务

systemctl restart firewalld.service

5.阿里云服务器安全组

6.进入bin目录重启redis-server

redis-server lconfig/redis.conf 

操作

测试是否java能够正常连接redis

public class TestPing {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("192.168.xx.xxx", 6379);
        String response = jedis.ping();
        System.out.println(response); // PONG
    }
}

3.事务

public class JedisTX {

    public static void main(String[] args) {
        Jedis jedis = new Jedis("127.0.0.1",6379);

        //先清空数据
        jedis.flushDB();
        //序列化存入值
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("name", "liang");
        jsonObject.put("age", 18);
        String result = jsonObject.toString();
        //开启事务
        Transaction multi = jedis.multi();
        try{
            multi.set("user", result);
             // int i = 1/0;
            multi.exec();  //执行事务
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            //关闭连接
            System.out.println(jedis.get("user"));
            jedis.close();
        }

    }
}

十三、SpringBoot整合Redis

1.导入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

springboot 2.x后，原本的Jedis被lectuce替换。

jedis: 采用的直连，多个线程操作的话，是不安全的。如果要避免不安全的话，使用jedis pool！更像BIO模式。
lettuce:采用netty,实例可以在多个线程中共享，不存在线程不安全的情况，可以
减少线程池的数据，更像DIO模式。

SpringBoot轻量级框架可以与其他框架无缝集成，只要导入相应的jar包即可。使用redis，我们能够在spring.factories处找到Redis相关的AutoConfiguration。

通过点击RedisAutoConfiguration，我们能够找到和配置文件相关的RedisProperties类

之前SpringBoot2.x默认使用Lettuce来替换Jedis。
Jedis：

@ConditionalOnClass注解中有两个类是默认不存在的，所以Jedis是无法生效的

Lettuce：

Lettuce生效了。

回到RedisAutoConfiguration

只有两个简单的Bean

• RedisTemplate: 操作Redis
• StringRedis Template: 操作Redis的String数据类型。

在RedisTemplate上也有一个条件注解，我们可以自定义自己的RedisTemplate。

如何通过编写配置文件来连接Redis，通过RedisProperties来得知。

RedisProperties的基本属性:

还有连接池相关的配置。注意使用时一定使用Lettuce的连接池。

2.编写配置文件

# 配置redis
spring:
 redis:
 	host: 39.99.xxx.xx
spring:
 redis:
 	port: 6379

3.使用RedisTemplate

@SpringBootTest
class RedisSpringbootApplicationTests {

    //整合SpringBoot
    @Autowired
    private RedisTemplate redisTemplate;

    @Test
    void contextLoads() {

        //redisTemplate里面封装了redis的操作
        // 1.操作数据类型交给对应opsForXXX方法来处理

        redisTemplate.opsForValue(); //处理字符串
        redisTemplate.opsForList();  //操作列表

        // 2.获取连接对象,操作数据库
        /**
         RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
         connection.flushAll();
         connection.flushDb();
         **/
        // 3.一些常规的操作，例如设置过期时间，获取所有键值等

        redisTemplate.opsForValue().set("mykey","liang");
        System.out.println(redisTemplate.opsForValue().get("mykey"));
    }

}

3.测试结果

发现是一些乱码，设计到对象序列化问题，网上传输时需要的对象需要用到序列化，否则出现乱码。

查看默认的RedisTemplate内部是什么样子：

在最开始就能看到几个关于序列化的参数。

默认的序列化器是采用JDK序列化器

默认的RedisTemplate中的所有序列化器都是使用这个序列化器：

RedisSerializer提供了多种序列化方案：

1、直接调用RedisSerializer的静态方法来返回序列化器，然后set。
2、自己new 相应的实现类，然后set

5.定制RedisTemplate的模板：

实现

@Configuration
public class RedisConfig {

   @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException {
        // 将template 泛型设置为 <String, Object>
        RedisTemplate<String, Object> template = new RedisTemplate();
        // 连接工厂，不必修改
        template.setConnectionFactory(redisConnectionFactory);
        /*
         * 序列化设置
         */
        // key、hash的key 采用 String序列化方式
        template.setKeySerializer(RedisSerializer.string());
        template.setHashKeySerializer(RedisSerializer.string());
        // value、hash的value 采用 Jackson 序列化方式
        template.setValueSerializer(RedisSerializer.json());
        template.setHashValueSerializer(RedisSerializer.json());
        template.afterPropertiesSet();
        
        return template;
    }
}

这样就可以解决序列化问题，只要传输对象实现了序列化，就可以存任何值啦。

十四、自定义Redis工具类

针对于RedisTemplate频繁使用.opForxxx才能进行对应操作，实际开发为了提高效率，将这些常用的公共API抽取出来封装成为一个工具类，直接使用工具类来间接操作Redis。

工具类参考:
https://www.cnblogs.com/zeng1994/p/03303c805731afc9aa9c60dbbd32a323.html

https://www.cnblogs.com/zhzhlong/p/11434284.html

工具类如下:

package com.liang.config;

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

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

@Component
public final class RedisUtils {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    // =============================common============================
    /**
     * 指定缓存失效时间
     * @param key 键
     * @param time 时间(秒)
     * @return
     */
    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 可以传一个值 或多个
     */
    @SuppressWarnings("unchecked")
    public void del(String... key) {
        if (key != null && key.length > 0) {
            if (key.length == 1) {
                redisTemplate.delete(key[0]);
            } else {
                redisTemplate.delete(CollectionUtils.arrayToList(key));
            }
        }
    }

    // ============================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)
     * @return
     */
    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)
     * @return
     */
    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
     * @return 值
     */
    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 对应多个键值
     * @return true 成功 false 失败
     */
    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)
     * @return
     */
    public double hincr(String key, String item, double by) {
        return redisTemplate.opsForHash().increment(key, item, by);
    }

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

    // ============================set=============================
    /**
     * 根据key获取Set中的所有值
     * @param key 键
     * @return
     */
    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 键
     * @return
     */
    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代表所有值
     * @return
     */
    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 键
     * @return
     */
    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倒数第二个元素，依次类推
     * @return
     */
    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 值
     * time  时间(秒)
     * @return
     */
    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 时间(秒)
     * @return
     */
    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 值
     * time 时间(秒)
     * @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 键
    a * @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.conf文件

vim lconfig/redis.conf

首先查看到的是内存大小，容量单位不区分大小写

INCLUDES,用于组合多个配置问题。

NETWORK网络部分，配置IP地址和端口号以及是否设置保护模式。

bind 127.0.0.1 #ip绑定
protected-mode yes #保护模式 默认开启
port 6379  #端口

常规配置

daemonize yes #开启redis后台运行
supervised no #标识redis正在运行

日志输出级别

默认输出级别为notice

日志输出文件

设置日志输出方便以后恢复为某个状态

默认数据库个数

另外还可以显示logo

持久化方式(SNAPSHOTTING:快照)

由于Redis是基于内存的数据库，需要将数据由内存持久化文件中，否则断电即失，造成数据的丢失。

持久化方式分为两种:

• RDB
• AOF

持久化规则:
持久化到文件，使用rdb或aof两种方式
save的用法是在多少秒内，至少有多少个key进行了修改，就进行持久化。

RDB文件相关

stop-writes-on-bgsave-error yes #持久化失败后继续工作
rdbcompression yes  #压缩.rdb文件
rdbchecksum yes  #检验rdb文件
dbfilename dump.rdb  #持久化后产生的文件名
dir ./ rdb

REPLICATION主从复制

Security:可进行密码设置

requirepass " "
可以直接输入
config set requirepass "密码" 

客户端连接相关

maxclients 10000 最大客户端数量
maxmemory <byte> 最大内存限制
maxmemory-policy noeviction # 内存达到限制值的处理策略

redis 默认的过期策略是volatile-lru。

config set maxmemory-policy volatile-lru

maxmemory-policy 的六种方式

• volatile-lru：只对设置了过期时间的key进行LRU（默认值）
• allkeys-lru ： 删除lru算法的key
• volatile-random：随机删除即将过期key
• allkeys-random：随机删除
• volatile-ttl ： 删除即将过期的
• noeviction ： 永不过期，返回错误

AOF相关部分

appendonly no  #默认不开启aof，使用rdb方式持久化
appendfilename "appendonly.aof" # 默认文件名

# appendfsync always  #每次修改都进行同步
appendfsync everysec #每秒执行一次同步
# appendfsync no #不进行同步，由操作系统进行同步，速度最快。

十六、持久化讲解

Redis是一个内存数据库，所有的数据都直接保存在内存中，一旦Redis进程异常退出或者服务器本身异常宕机，我们存储Redis的数据就会消失，找不到了~，因此需要持久化数据。

Redis持久化的策略分为: RDB和AOF,用于存储在内存的数据备份在磁盘上，并且服务器重启时将备份文件进行重载。

RDB是基于内存快照，AOF是基于操作日志。

RDB

什么是RDB？

RDB的英文是Redis DataBase，中文为Redis数据集，是一个快照持久化策略。RDB是redis默认的持久化策略。

在指定时间间隔内，将内存的数据集快照写入数据库中；在启动服务器时直接读取快照文件，进行数据的恢复。

默认情况下，Redis会将数据集快照保存在一个名为dump.rbd的二进制文件中，文件名可以在配置文件中修改。

工作原理

RDB分为两种:一种是同步的，另一种是异步的。

同步:

调用save命令即可触发redis进行RDB文件生成备份数据，这是一个同步命令，在备份完成之前，Redis服务器不会去响应来自客户端的任何请求。

异步
在进行RDB的时候，redis的主线程是不会做io操作的，主线程会fork一个子线程来完成该操作。

1. Redis调用forks,同时拥有父进程和子进程。
2. 子进程将数据集写入到一个临时的RDB文件中。
3. 当子进程完成对新的RDB写入时，Redis就会用新的RDB文件去替代原来的RDB文件。

这种工作方式使得Redis可以从写时复制(copy-on-write)机制中获益（因为使用子进程操作写操作时，父进程仍然可以接收来自客户端的请求。）

异步的RDB生成策略是主流，在以上命令中，bgsave和save，这两个命令需要我们手动的在客户端发送请求才会触发，称为主动触发。

而在配置文件中我们制定了被动触发的规则。

触发机制

1. save的规则满足时，会触发rdb原则。
2. 执行flushall命令时，会触发rdb原则。
3. 退出redis,也会自动产生rdb文件。

save

使用save命令，能够立即将当前内存数据进行持久化，但是会堵塞，这种方式是不可取的。

因为save命令是同步命令，会占用redis的主进程。若Redis数据非常多时，save命令执行速度会非常慢，阻塞所有客户端请求。

flushall命令

flushall命令会触发持久化。

触发持久化规则

• 满足配置文件中的触发条件:

通过在配置文件中可以对Redis进行设置，让其满足以下这条规则，自动进行数据集保存

save N M # N秒内至少有M个改动

bgsavebgsave是异步执行的，进行数据持久化的时候，redis还可以继续响应客户端请求。

bgsave和save请求

RDB优缺点

优点:

1. 适合大规模的数据恢复
2. 对数据的完整性要求不高

缺点：
1.需要一定的时间间隔进行持久化操作，如果redis意外宕机了，那么最后一次修改的数据就没了
2.fork进程的时候，需要一定的内存消耗。

AOF

什么是AOF

AOF的英文为Append Only File

将我们所有的命令记录下来，history,恢复的时候就把这个文件全部执行一遍。

以日志的形式来记录每个写的操作，将Redis执行过的所有写的指令记录下来（读操作不记录），只许追加文件而不可修改文件，redis启动之初将读取该文件重新构建数据，总之，通过AOF的持久化策略方式，redis重启后会根据日志的内容将写指令从头到尾执行一次来实现数据的恢复。

RDB(快照功能)并不是非常耐久（durable）: 如果Redis因为某些原因而造成服务器宕机故障，会丢失最近写入未保存到快照的某些数据。从1.1版本开始，Redis增加了一种耐久的持久化方式:AOF持久化。

使用AOF,需要从配置文件中开启

默认是不开启AOF的，开启可以实现

appendonly no #默认不开启aof 使用rdb方式持久化

appendfilename "appendonly.aof"  #默认文件名

若aof文件出错，可以通过redis-check-aof --fix来修复aof文件

redis-check-aof --fix appendnoly.aof

配置文件说明

appendonly yes  # 默认是不开启aof模式的，默认是使用rdb方式持久化的，在大部分的情况下，rdb完全够用
appendfilename "appendonly.aof"

#appendfsync always # 每次修改都会sync 消耗性能
appendfsync everysec # 每秒执行一次 sync 可能会丢失这一秒的数据
#appendfsync no # 不执行 sync ,这时候操作系统自己同步数据，速度最快

优点和缺点

优点:

1. 每一次修改都会同步，文件的完整性会更好
2. 每秒同步一次，可能会丢失一秒的数据
3. 从不同步，效率最高

缺点:

1.相比于数据文件来说，aof远远大于rdb,修复速度比rdb慢！
2.AOF的运行效率比AOF慢，所以redis默认配置为rdb持久化

RDB和AOF选择

如何选择使用哪种持久化方式?

一般来说，如果想达到媲美 PostgreSQL 的数据安全性，应该选择使用两种持久化功能。

如果你关心你的数据但仍可以承受数分钟以内的数据丢失，那么你可以只使用RDB持久化。

AOF持久化不推荐单独使用，因为定时生成 RDB 快照（snapshot）非常便于进行数据库备份， 并且 RDB 恢复数据集的速度也要比 AOF 恢复的速度要快。

十七、Reids发布与订阅

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

下面展示了频道channel1,以及订阅这个频道的三个客户端 ----client2,client5和client1之间的关系。

当有新消息通过PUBLISH命令发送给频道channel1时，这个消息就会被发送给订阅它的三个客户端:

命令

演示

------------订阅端----------------------
127.0.0.1:6379> SUBSCRIBE liang
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "liang"
3) (integer) 1

--------------消息发布端-------------------
127.0.0.1:6379> PUBLISH liang "hello"
(integer) 1

-----------------查看活跃的频道------------
127.0.0.1:6379> PUBSUB channels
1) "liang"

原理:

每个Redis服务器进程都维持着一个表示服务器状态的redis.h/redisServer结构，结构pubsub_channels属性是一个字典。

这个字典就用于保存订阅频道的信息，其中，字典的键为正在被订阅的频道， 而字典的值则是一个链表， 链表中保存了所有订阅这个频道的客户端。

客户端订阅，就被链接到对应频道的链表的尾部，退订则就是将客户端节点从链表中移除。

缺点

1. 如果一个客户端订阅了频道，但自身读取速度比不上频道发布的速度，导致redis缓存区不断扩大，可能会引起redis本身速度变慢，甚至崩溃。
2. 这与数据传输可靠性有关，若订阅方断线，可能会丢失发布的信息。

应用

1. 消息订阅: 公众号订阅，微博关注
2. 多人在线聊天室

稍微复杂的场景，我们就会使用消息中间件MQ处理。

十八、Redis主从复制

主从复制,指的是将一台Redis服务器上的数据，复制到其他的Redis服务器。前者为主节点(Master/Leader),后者为从节点(Slave/Follower),数据的复制是单向的！
只能是主节点复制到从节点。(其中主节点主要是写，而从节点主要是读)

默认情况下，每台Redis服务器都是主节点，一个主节点可以有0个或者多个从节点，但每个从节点只有一个主节点。

作用

• 数据冗余:主从复制实现了数据的热备份，是持久化之外的一种数据冗余的方式。
• 故障恢复: 当主节点出现故障时，从节点会暂时替代主节点提供服务，是一种服务冗余的方式。
• 负载均衡: 在主从复制的基础上，配合读写分离，由主节点进行写操作，从节点进行读操作，分担服务器的负载，尤其是适用在多读少写的场景下，通过多个从节点分担负载，提高并发量。
• 高可用基石: 主从复制还是哨兵和集群能够实施的基础。

环境配置

replication模块: 用于查看当前库的信息

info replication

默认为主机，

既然需要启动多个服务，就需要多个配置文件。每个配置文件对应修改以下信息：

端口号

pid文件名

日志文件名

rdb文件名

查看单机多服务集群

ps -ef | grep redis

一主二从配置

默认情况下，每台Redis服务器都是主节点；==我们一般情况下只用配置从机就好了！

主机端口: 6380

从机端口: 6381,6382

在每个redis服务器上使用SLAVEOF host port设置为从机配置主机了。

主机上也能看到从机的状态：

使用命令搭建，只是暂时的，真实开发在从机的配置文件中进行配置，这样的话是永久的。

使用规则

1. 从机只能读，不能写，主机可读可写但是多用于写
   
2. 主机断电宕机后，默认情况下从机的角色不会发生什么变化，集群中只是失去了写操作，当主机恢复以后，又会连接上从机恢复原状。
3. 当从机断电宕机后，若不是使用配置文件配置的从机，再次启动后作为主机是无法获取之前主机的数据的，若此时重新配置称为从机，又可以获取到主机的所有数据。这里就要提到一个同步原理。
4. 默认情况下，主机故障后，不会出现新的主机，有两种方式可以产生新的主机：

• 从机手动执行命令slaveof no one,这样执 行以后从机会独立出来成为一个主机
• 使用哨兵模式（自动选举）

若没有老大，则需手动选择一个老大！
如果主机断开了连接，我们可以使用SLAVEOF no one让自己变成主机！其他的节点就可以手动连接到最新的主节点（手动）！如果这个时候老大修复了，那么重新连接！

十九、哨兵模式

主从切换技术

当主服务宕机后，需要手动的将一台从服务器变为主服务器，需要人工干预，还会造成短期服务不可用。更多时候考虑哨兵模式

单机哨兵模式

哨兵模式是一种特殊的模式，首先redis提供了哨兵的命令，哨兵是一个独立的进程，作为进程，它会独立运行，其原理哨兵通过发送命令，等待Redis服务器响应，从而监控运行的多个Redis实例。

• 通过发送命令，让Redis服务器返回监控其运行状态，包括主服务器和从服务器。

• 当哨兵监测到master宕机，会自动将slave切换成master,然后通过发布订阅模式通知其他的从服务器，修改配置文件，让它们切换主机。

多哨兵模式

一个哨兵进程出现故障，可能会引起问题，因此使用多个哨兵来进行监控，形成哨兵集群，哨兵之间进行监控，称为多哨兵模式。

哨兵的核心配置：

sentinel monitor mymaster 127.0.0.1 6379 1

最后的数字1表示：当一个哨兵认为主机断开，会客观的认为发生了主机故障，然后开始新的选举主机

测试:

断开主机6379,查看哨兵模式状态:

可以发现哨兵模式下重新选择了主机,即使原先的主机恢复状态，也会加入到新的主机上，称为新主机的从机。

哨兵模式的优缺点

优点

1. 哨兵集群，基于主从复制模式，所有主从复制的优点，它都有。
2. 主从可以切换，故障可以转移，系统的可用性好。
3. 哨兵模式是主从模式的升级，手动到自动，更加健壮。

缺点

1. Redis不好在线扩容，集群容量一旦达到上限，在线扩容就很麻烦。
2. 实现哨兵模式的配置很麻烦，里面有很多配置项

哨兵模式的配置文件

sentinel.conf

# Example sentinel.conf
 
# 哨兵sentinel实例运行的端口 默认26379
port 26379
 
# 哨兵sentinel的工作目录
dir /tmp
 
# 哨兵sentinel监控的redis主节点的 ip port 
# master-name  可以自己命名的主节点名字 只能由字母A-z、数字0-9 、这三个字符".-_"组成。
# quorum 当这些quorum个数sentinel哨兵认为master主节点失联 那么这时 客观上认为主节点失联了
# sentinel monitor <master-name> <ip> <redis-port> <quorum>
sentinel monitor mymaster 127.0.0.1 6379 1
 
# 当在Redis实例中开启了requirepass foobared 授权密码 这样所有连接Redis实例的客户端都要提供密码
# 设置哨兵sentinel 连接主从的密码 注意必须为主从设置一样的验证密码
# sentinel auth-pass <master-name> <password>
sentinel auth-pass mymaster MySUPER--secret-0123passw0rd
 
 
# 指定多少毫秒之后 主节点没有应答哨兵sentinel 此时 哨兵主观上认为主节点下线 默认30秒
# sentinel down-after-milliseconds <master-name> <milliseconds>
sentinel down-after-milliseconds mymaster 30000
 
# 这个配置项指定了在发生failover主备切换时最多可以有多少个slave同时对新的master进行 同步，
这个数字越小，完成failover所需的时间就越长，
但是如果这个数字越大，就意味着越 多的slave因为replication而不可用。
可以通过将这个值设为 1 来保证每次只有一个slave 处于不能处理命令请求的状态。
# sentinel parallel-syncs <master-name> <numslaves>
sentinel parallel-syncs mymaster 1
 
 
 
# 故障转移的超时时间 failover-timeout 可以用在以下这些方面： 
#1. 同一个sentinel对同一个master两次failover之间的间隔时间。
#2. 当一个slave从一个错误的master那里同步数据开始计算时间。直到slave被纠正为向正确的master那里同步数据时。
#3.当想要取消一个正在进行的failover所需要的时间。  
#4.当进行failover时，配置所有slaves指向新的master所需的最大时间。不过，即使过了这个超时，slaves依然会被正确配置为指向master，但是就不按parallel-syncs所配置的规则来了
# 默认三分钟
# sentinel failover-timeout <master-name> <milliseconds>
sentinel failover-timeout mymaster 180000
 
# SCRIPTS EXECUTION
 
#配置当某一事件发生时所需要执行的脚本，可以通过脚本来通知管理员，例如当系统运行不正常时发邮件通知相关人员。
#对于脚本的运行结果有以下规则：
#若脚本执行后返回1，那么该脚本稍后将会被再次执行，重复次数目前默认为10
#若脚本执行后返回2，或者比2更高的一个返回值，脚本将不会重复执行。
#如果脚本在执行过程中由于收到系统中断信号被终止了，则同返回值为1时的行为相同。
#一个脚本的最大执行时间为60s，如果超过这个时间，脚本将会被一个SIGKILL信号终止，之后重新执行。
 
#通知型脚本:当sentinel有任何警告级别的事件发生时（比如说redis实例的主观失效和客观失效等等），将会去调用这个脚本，
#这时这个脚本应该通过邮件，SMS等方式去通知系统管理员关于系统不正常运行的信息。调用该脚本时，将传给脚本两个参数，
#一个是事件的类型，
#一个是事件的描述。
#如果sentinel.conf配置文件中配置了这个脚本路径，那么必须保证这个脚本存在于这个路径，并且是可执行的，否则sentinel无法正常启动成功。
#通知脚本
# sentinel notification-script <master-name> <script-path>
  sentinel notification-script mymaster /var/redis/notify.sh
 
# 客户端重新配置主节点参数脚本
# 当一个master由于failover而发生改变时，这个脚本将会被调用，通知相关的客户端关于master地址已经发生改变的信息。
# 以下参数将会在调用脚本时传给脚本:
# <master-name> <role> <state> <from-ip> <from-port> <to-ip> <to-port>
# 目前<state>总是“failover”,
# <role>是“leader”或者“observer”中的一个。 
# 参数 from-ip, from-port, to-ip, to-port是用来和旧的master和新的master(即旧的slave)通信的
# 这个脚本应该是通用的，能被多次调用，不是针对性的。
# sentinel client-reconfig-script <master-name> <script-path>
sentinel client-reconfig-script mymaster /var/redis/reconfig.sh

参考博客 https://www.jianshu.com/p/06ab9daf921d

二十、缓存穿透与雪崩

缓存穿透

概念

在默认情况下，用户请求数据时，会先在缓存(Redis)中查找，若未找到即缓存未命中，则会去数据库查找，数据量少问题可能不大，但对于秒杀级别的请求数据，缓存未命中的话，就会全部转移到数据库中，造成数据库极大的压力，可能会导致系统的崩溃，而此时缓存相当于空气一样未起任务有效作用，因此被称为缓存穿透。

洪水攻击: 。网络安全中也有人恶意使用这种手段进行攻击。

解决方案

布隆过滤器

对所有可能查询的参数以Hash的形式存储，以便快速确定是否存在这个值，布隆过滤器在控制层先进行拦截校验，校验不通过直接原路返回，以此减轻存储系统的压力。

缓存空对象

倘若一次请求若在缓存和数据库中都没有找到，就会在缓存中找一个空对象用于处理后续这个请求。

这会出现一个小缺陷,存储空对象也需要空间，大量的空对象会耗费一定的空间，存储效率并不高。解决这个缺陷的方式就是给空值设置较短过期时间

即使对空值设置了过期时间，还是会存在缓存层和存储层的数据会有一段时间窗口的不一致，这对于需要保持一致性的业务会有影响。

缓存击穿

概念

相较于缓存穿透来说，缓存击穿的目的性更强，一个存在的key,在缓存过期的那一刻，同时拥有大量的请求，这些请求都会击穿到数据库(DB),造成瞬间DB请求量大，压力骤增。 这就是缓存击穿**。只是针对某中某个key的缓存不可用而导致击穿，但其他的key可以正常使用缓存响应。**

例如热搜排行榜时，可能某条新闻被同时大量请求访问可能导致缓存击穿。

解决方案

1. 设置热点数据永不过期
   这样就不会出现热点数据过期的情况，但是当Redis内存空间满的时候也会清理部分数据，而且此种方案会占用空间，一旦热点数据多了起来，就会占用部分空间。
2. 加互斥锁(分布式锁)
   在访问key之前，采用SETNX(set if not exists) 来设置一个短期key来锁住当前key的访问。访问结束后删除该短期key，但要保证只有一个线程能够访问，对锁的要求比较高。

缓存雪崩

概念
大量的key都设置相同的过期时间，导致在缓存在同一时刻全部失效，造成瞬间DB请求量大，压力骤增，引起雪崩。

解决方案

1. redis高可用:采用搭建集群的方式，来解决一台redis服务器挂了的情形
2. 限流降级: 在缓存失效后，通过加锁或者队列来控制读数据库写缓存的线程数量。比如对某个key只允许一个线程查询数据和写缓存，其他线程进入等待。
3. 数据预热: 在正式部署之前，要先把数据预访问一遍，这样部分大量访问访问的数据会加载到缓存中，在即将发生大并发访问时手动触发加载缓存不同的key，设置不同的过期时间，让缓存的失效的时间点尽量均匀。