缓存

程序员文章站 2022-06-18 23:18:07

...

为什么要使用缓存

缓存的分类
客户端缓存
页面缓存
浏览器缓存
App客户端缓存
网络缓存
代理缓存
CDN缓存

服务器缓存
数据库缓存
平台缓存级缓存

缓存最终的目的是为减轻服务端压力，减少网络传输请求
基于Map集合实现本地缓存
定义Map缓存工具类
@Component
public class MapEhcaChe<K, V> {
private Map<K, V> ehcaCheMap = new ConcurrentHashMap<K, V>();

public void put(K k, V value) {
	ehcaCheMap.put(k, value);
}

public V get(K k) {
	return ehcaCheMap.get(k);
}

public void remove(K k) {
	ehcaCheMap.remove(k);
}

}

使用案例

@RequestMapping("/ehcaChePut")
public String put(String key, String value) {
	mapEhcaChe.put(key, value);
	return "success";
}

@RequestMapping("/get")
public String get(String key) {
	String value = mapEhcaChe.get(key);
	return value;
}

本地缓存Ehcache
什么是Ehcache
Ehcache是纯java的开源缓存框架，具有快速、精干等特点，是Hibernate中默认的CacheProvider。它主要面向通用缓存、Java EE和轻量级容器，具有内存和磁盘存储、缓存加载器、缓存扩展、缓存异常处理程序。
Ehcache最初由Greg Luck于2003年开始开发。2009年,该项目被Terracotta购买。软件仍然开源,但一些新的主要功能(例如，快速可重启性之间的一致性的)只能在商业产品中使用。
Ehcache 被广泛用于在Hibernate、Spring、Cocoon等其他开源系统。
Ehcache的主要特性
1.快速；
2.简单；
3.多种缓存策略；
4.缓存数据有两级：内存和磁盘，因此无需担心容量问题；
5.缓存数据会在虚拟机重启的过程中写入磁盘；
6.可以通过 RMI、可插入 API 等方式进行分布式缓存；
7.具有缓存和缓存管理器的侦听接口；
8.支持多缓存管理器实例，以及一个实例的多个缓存区域；
9.提供 Hibernate 的缓存实现；

Ehcache使用介绍

Ehcache是用来管理缓存的一个工具，其缓存的数据可以是存放在内存里面的，也可以是存放在硬盘上的。其核心是CacheManager，一切Ehcache的应用都是从CacheManager开始的。它是用来管理Cache（缓存）的，一个应用可以有多个CacheManager，而一个CacheManager下又可以有多个Cache。Cache内部保存的是一个个的Element，而一个Element中保存的是一个key和value的配对，相当于Map里面的一个Entry。

Ehcache缓存过期策略

当缓存需要被清理时（比如空间占用已经接近临界值了），需要使用某种淘汰算法来决定清理掉哪些数据。常用的淘汰算法有下面几种：
FIFO：First In First Out，先进先出。判断被存储的时间，离目前最远的数据优先被淘汰。
LRU：Least Recently Used，最近最少使用。判断最近被使用的时间，目前最远的数据优先被淘汰。
LFU：Least Frequently Used，最不经常使用。在一段时间内，数据被使用次数最少的，优先被淘汰。

Ehcache基本用法

SpringBoot2.0整合Ehcache框架

Maven环境依赖

org.springframework.boot
spring-boot-starter-parent
2.0.0.RELEASE

org.projectlombok
lombok

org.springframework.boot
spring-boot-starter-web

org.springframework.boot
spring-boot-starter-tomcat

org.apache.tomcat.embed
tomcat-embed-jasper

org.springframework.boot
spring-boot-starter-log4j
1.3.8.RELEASE

org.springframework.boot
spring-boot-starter-aop

commons-lang
commons-lang
2.6

org.apache.httpcomponents
httpclient

com.alibaba
fastjson
1.2.47

javax.servlet
jstl

taglibs
standard
1.1.2

org.springframework.boot
spring-boot-starter-cache

net.sf.ehcache
ehcache
2.9.1

org.mybatis.spring.boot
mybatis-spring-boot-starter
1.1.1

mysql
mysql-connector-java

YML配置文件信息
###端口号配置
server:
port: 8081
###数据库配置
spring:
datasource:
url: jdbc:mysql://localhost:3306/test
username: root
password: root
driver-class-name: com.mysql.jdbc.Driver
test-while-idle: true
test-on-borrow: true
validation-query: SELECT 1 FROM DUAL
time-between-eviction-runs-millis: 300000
min-evictable-idle-time-millis: 1800000

缓存配置读取

cache:
type: ehcache
ehcache:
config: classpath:app2_ehcache.xml

App启动方式
@MapperScan(basePackages = { “com.itmayiedu.mapper” })
@EnableCaching
@SpringBootApplication
public class App {

public static void main(String[] args) {
	SpringApplication.run(App.class, args);
}

}
@EnableCaching 开启ehcache缓存模式

项目使用
@CacheConfig(cacheNames = “userCache”)
public interface UserMapper {
@Select(“SELECT ID ,NAME,AGE FROM users where id=#{id}”)
@Cacheable
List getUser(@Param(“id”) Long id);
}

@Cacheable 加了该注解的方法表示可以缓存
@CacheConfig 表示创建缓存配置，Key为userCache
EhCache配置
app1_ehcache

<?xml version="1.0" encoding="UTF-8"?>

<diskStore path="java.io.tmpdir/ehcache-rmi-4000" />

<!-- 多台机器配置 rmiUrls=//192.168.8.32:400002/demoCache|//192.168.5.231:400003/demoCache -->
<cacheManagerPeerProviderFactory
	class="net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory"
	properties="peerDiscovery=manual,rmiUrls=//127.0.0.1:5000/userCache">
</cacheManagerPeerProviderFactory>
<!-- 配置 rmi 集群模式 -->
<cacheManagerPeerListenerFactory
	class="net.sf.ehcache.distribution.RMICacheManagerPeerListenerFactory"
	properties="hostName=127.0.0.1,port=4000,socketTimeoutMillis=120000" />

<!-- 多播方式配置 搜索某个网段上的缓存 timeToLive 0是限制在同一个服务器 1是限制在同一个子网 32是限制在同一个网站 64是限制在同一个region 
	128是限制在同一个大洲 255是不限制 <cacheManagerPeerProviderFactory class="net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory" 
	properties="peerDiscovery=automatic, multicastGroupAddress=224.1.1.1, multicastGroupPort=40000, 
	timeToLive=32" /> -->

<!-- 默认缓存 -->
<defaultCache maxElementsInMemory="1000" eternal="true"
	timeToIdleSeconds="120" timeToLiveSeconds="120" overflowToDisk="true"
	diskSpoolBufferSizeMB="30" maxElementsOnDisk="10000000"
	diskPersistent="true" diskExpiryThreadIntervalSeconds="120"
	memoryStoreEvictionPolicy="LRU">
</defaultCache>

<!-- demo缓存 -->
<cache name="userCache" maxElementsInMemory="1000" eternal="false"
	timeToIdleSeconds="120" timeToLiveSeconds="120" overflowToDisk="true"
	diskSpoolBufferSizeMB="30" maxElementsOnDisk="10000000"
	diskPersistent="false" diskExpiryThreadIntervalSeconds="120"
	memoryStoreEvictionPolicy="LRU">
	<cacheEventListenerFactory
		class="net.sf.ehcache.distribution.RMICacheReplicatorFactory" />
	<!-- 用于在初始化缓存，以及自动设置 -->
	<bootstrapCacheLoaderFactory
		class="net.sf.ehcache.distribution.RMIBootstrapCacheLoaderFactory" />
</cache>

app2_ehcache

<?xml version="1.0" encoding="UTF-8"?>

<diskStore path="java.io.tmpdir/ehcache-rmi-5000" />

<!-- 多台机器配置 -->
<cacheManagerPeerProviderFactory
	class="net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory"
	properties="peerDiscovery=manual,rmiUrls=//127.0.0.1:4000/userCache">
</cacheManagerPeerProviderFactory>

<cacheManagerPeerListenerFactory
	class="net.sf.ehcache.distribution.RMICacheManagerPeerListenerFactory"
	properties="hostName=127.0.0.1,port=5000,socketTimeoutMillis=120000" />

<!-- 多播方式配置 搜索某个网段上的缓存 timeToLive 0是限制在同一个服务器 1是限制在同一个子网 32是限制在同一个网站 64是限制在同一个region 
	128是限制在同一个大洲 255是不限制 <cacheManagerPeerProviderFactory class="net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory" 
	properties="peerDiscovery=automatic, multicastGroupAddress=224.1.1.1, multicastGroupPort=40000, 
	timeToLive=32" /> -->

<!-- 默认缓存 -->
<defaultCache maxElementsInMemory="1000" eternal="true"
	timeToIdleSeconds="120" timeToLiveSeconds="120" overflowToDisk="true"
	diskSpoolBufferSizeMB="30" maxElementsOnDisk="10000000"
	diskPersistent="true" diskExpiryThreadIntervalSeconds="120"
	memoryStoreEvictionPolicy="LRU">
</defaultCache>

<!-- demo缓存 -->
<cache name="userCache" maxElementsInMemory="1000" eternal="false"
	timeToIdleSeconds="120" timeToLiveSeconds="120" overflowToDisk="true"
	diskSpoolBufferSizeMB="30" maxElementsOnDisk="10000000"
	diskPersistent="false" diskExpiryThreadIntervalSeconds="120"
	memoryStoreEvictionPolicy="LRU">
	<cacheEventListenerFactory
		class="net.sf.ehcache.distribution.RMICacheReplicatorFactory" />
	<!-- 用于在初始化缓存，以及自动设置 -->
	<bootstrapCacheLoaderFactory
		class="net.sf.ehcache.distribution.RMIBootstrapCacheLoaderFactory" />
</cache>

清除缓存
@Autowired
private CacheManager cacheManager;

@RequestMapping("/remoKey")
public void remoKey() {
	cacheManager.getCache("userCache").clear();
}

参数相关配置

1、diskStore ：指定数据(.data and .index)存储位置，可指定磁盘中的文件夹位置期 The diskStore element is optional. It must be configured if you have overflowToDisk or diskPersistent enabled for any cache. If it is not configured, a warning will be issues and java.io.tmpdir will be used.

２、defaultCache ：默认的管理策略

Ehcache 使用Map集合实现的 element 其实就是 key 和value
一、以下属性是必须的：
　　1、name： Cache的名称，必须是唯一的(ehcache会把这个cache放到HashMap里)。
　　2、maxElementsInMemory：在内存中缓存的element的最大数目。
　　3、maxElementsOnDisk：在磁盘上缓存的element的最大数目，默认值为0，表示不限制。
　　４、eternal：设定缓存的elements是否永远不过期。如果为true，则缓存的数据始终有效，如果为false那么还要根据timeToIdleSeconds，timeToLiveSeconds判断。
　　５、overflowToDisk：如果内存中数据超过内存限制，是否要缓存到磁盘上。
二、以下属性是可选的：
　　１、timeToIdleSeconds：对象空闲时间，指对象在多长时间没有被访问就会失效。只对eternal为false的有效。默认值0，表示一直可以访问。
　　２、timeToLiveSeconds：对象存活时间，指对象从创建到失效所需要的时间。只对eternal为false的有效。默认值0，表示一直可以访问。
　　３、diskPersistent：是否在磁盘上持久化。指重启jvm后，数据是否有效。默认为false。
　　４、diskExpiryThreadIntervalSeconds：对象检测线程运行时间间隔。标识对象状态的线程多长时间运行一次。
　　５、diskSpoolBufferSizeMB： DiskStore使用的磁盘大小，默认值30MB。每个cache使用各自的DiskStore。
　　６、memoryStoreEvictionPolicy：如果内存中数据超过内存限制，向磁盘缓存时的策略。默认值LRU，可选FIFO、LFU。

Ehcache集群模式

由于 EhCache 是进程中的缓存系统，一旦将应用部署在集群环境中，每一个节点维护各自的缓存数据，当某个节点对缓存数据进行更新，这些更新的数据无法在其它节点*享，这不仅会降低节点运行的效率，而且会导致数据不同步的情况发生。例如某个网站采用 A、B 两个节点作为集群部署，当 A 节点的缓存更新后，而 B 节点缓存尚未更新就可能出现用户在浏览页面的时候，一会是更新后的数据，一会是尚未更新的数据，尽管我们也可以通过 Session Sticky 技术来将用户锁定在某个节点上，但对于一些交互性比较强或者是非 Web 方式的系统来说，Session Sticky 显然不太适合。

常用集群模式
EhCache从1.7版本开始，支持五种集群方案，分别是：
Terracotta、RMI、JMS、JGroups、EhCache Server

RMi集群模式
你如何知道集群环境中的其他缓存？
• 分布式传送的消息是什么形式？
• 什么情况需要进行复制？增加（Puts），更新（Updates）或是失效（Expiries）？
• 采用什么方式进行复制？同步还是异步方式？

1、正确的元素类型：只有可序列化的元素可以进行复制。一些操作，比如移除，只需要元素的键值而不用整个元素；在这样的操作中即使元素不是可序列化的但键值是可序列化的也可以被复制。
2、成员发现（Peer Discovery）：Ehcache进行集群的时候有一个cache组的概念。每个cache都是其他cache的一个peer，没有主cache的存在。成员发现（Peer Discovery）正是用来解决 “你如何知道集群环境中的其他缓存?” 这个问题的。Ehcache提供了两种机制用来进行成员发现，即：自动成员发现和手动成员发现。要使用一个内置的成员发现机制要在ehcache的配置文件中指定cacheManagerPeerProviderFactory元素的class属性为
net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory。

Ehcache的使用场景
使用纯java的ehcache作为本地缓存
Reids 作为远程分布式缓存
解决redis缓存压力过大，提高缓存速度，以及缓存性能。

Redis和Ehcache缓存的区别
如果是单个应用或者对缓存访问要求很高的应用，用ehcache。
如果是大型系统，存在缓存共享、分布式部署、缓存内容很大的，建议用redis。

实际工作中使用Ehcache

我们在项目中使用集中式缓存(Redis或者式Memcached等)，通常都是检查缓存中是否存在
期望值的数据，如果存在直接返回，如果不存在就查询数据库让后在将数据库缓存，

这个时候如果缓存系统因为某写原因宕机，造成服务无法访问，那么大的量请求直接穿透到数据库，最数据库压力非常大。

这时候我们让ehcache作为二级缓存，当redis服务器宕机后，可以查询ehcache缓存。
这样能够有效的扛住服务器请求压力。

分布式Redis缓存

什么是Redis

Redis是由意大利人Salvatore Sanfilippo（网名：antirez）开发的一款内存高速缓存数据库。Redis全称为：Remote Dictionary Server（远程数据服务），该软件使用C语言编写，Redis是一个key-value存储系统，它支持丰富的数据类型，如：string、list、set、zset(sorted set)、hash。
Redis采用的是基于内存的采用的是单进程单线程模型的KV数据库，由C语言编写。官方提供的数据是可以达到100000+的qps

Redis单线程好处

代码更清晰，处理逻辑更简单
不用去考虑各种锁的问题，不存在加锁释放锁操作，没有因为可能出现死锁而导致的性能消耗
不存在多进程或者多线程导致的切换而消耗CPU
所以redis线程是安全的

单进程单线程弊端
无法发挥多核CPU性能，不过可以通过在单机开多个Redis实例来完善；

Redis应用场景

令牌(Token)生成
短信验证码
发布订阅
相当于消息系统，ActiveMQ，RocketMQ等工具类似，但是个人觉得简单用一下还行，如果对于数据一致性要求高的话还是用RocketMQ等专业系统。
由于redis把数据添加到队列是返回添加元素在队列的第几位，所以可以做判断用户是第几个访问这种业务
队列不仅可以把并发请求变成串行，并且还可以做队列或者栈使用
分布式锁

验证前端的重复请求（可以*扩展类似情况），可以通过redis进行过滤：每次请求将request Ip、参数、接口等hash作为key存储redis（幂等性请求），设置多长时间有效期，然后下次请求过来的时候先在redis中检索有没有这个key，进而验证是不是一定时间内过来的重复提交
秒杀系统，基于redis是单线程特征，防止出现数据库“**”
全局增量ID生成，类似“秒杀”
5. 计数器
诸如统计点击数等应用。由于单线程，可以避免并发问题，保证不会出错，而且100%毫秒级性能！
6. 缓存(热点数据)
热点数据（经常会被查询，但是不经常被修改或者删除的数据），首选是使用redis缓存，毕竟强大到冒泡的QPS和极强的稳定性不是所有类似工具都有的，而且相比于memcached还提供了丰富的数据类型可以使用，另外，内存中的数据也提供了AOF和RDB等持久化机制可以选择，要冷、热的还是忽冷忽热的都可选。
结合具体应用需要注意一下：很多人用spring的AOP来构建redis缓存的自动生产和清除，过程可能如下：
Select 数据库前查询redis，有的话使用redis数据，放弃select 数据库，没有的话，select 数据库，然后将数据插入redis
update或者delete数据库钱，查询redis是否存在该数据，存在的话先删除redis中数据，然后再update或者delete数据库中的数据
上面这种操作，如果并发量很小的情况下基本没问题，但是高并发的情况请注意下面场景：
为了update先删掉了redis中的该数据，这时候另一个线程执行查询，发现redis中没有，瞬间执行了查询SQL，并且插入到redis中一条数据，回到刚才那个update语句，这个悲催的线程压根不知道刚才那个该死的select线程犯了一个弥天大错！于是这个redis中的错误数据就永远的存在了下去，直到下一个update或者delete。

Linux环境下安装Redis
下载Redis安装包
wget http://download.redis.io/releases/redis-3.2.9.tar.gz
解压Redis安装包
tar -zxvf redis-3.2.9.tar.gz
安装
cd redis-3.2.9
Make
Cd src
make install PREFIX=/usr/local/redis
移动配置文件到安装目录下
cd …/
mkdir /usr/local/redis/etc
mv redis.conf /usr/local/redis/etc

配置redis为后台启动
vi /usr/local/redis/etc/redis.conf //将daemonize no 改成daemonize yes
vi /usr/local/redis/etc/redis.conf // requirepass 123
开启redis
/usr/local/redis/bin/redis-server /usr/local/redis/etc/redis.conf

连接Redis客户端

./redis-cli -h 127.0.0.1 -p 6379 -a “123456”
PING 结果表示成功

关闭防火墙

//临时关闭
systemctl stop firewalld
//禁止开机启动
systemctl disable firewalld
Removed symlink /etc/systemd/system/multi-user.target.wants/firewalld.service.
Removed symlink /etc/systemd/system/dbus-org.fedoraproject.FirewallD1.service.

停止Redis服务
./redis-cli -h 127.0.0.1 -p 6379 -a “123456” shutdown
修改redis.conf
注释掉
#bind 127.0.0.1 开启外网访问

Redis基本数据类型
Redis目前支持5种数据类型，分别是：
String（字符串）
List（列表）
Hash（字典）
Set（集合）
Sorted Set（有序集合）
下面就分别介绍这五种数据类型及其相应的操作命令。

String（字符串）
String是简单的 key-value 键值对，value 不仅可以

是 String，也可以是数字。String在redis内部存储默认就是一个字符串，被redisObject所引用，当遇到incr,decr等操作时会转成数值型进行计算，此时redisObject的encoding字段为int。
String在redis内部存储默认就是一个字符串，被redisObject所引用，当遇到incr,decr等操作时会转成数值型进行计算，此时redisObject的encoding字段为int。
应用场景:
String是最常用的一种数据类型，普通的key/value存储都可以归为此类，这里就不所做解释了。

redis 127.0.0.1:6379> SET mykey “redis”
OK
redis 127.0.0.1:6379> GET mykey
“redis”

List(列表)

Redis列表是简单的字符串列表，可以类比到C++中的std::list，简单的说就是一个链表或者说是一个队列。可以从头部或尾部向Redis列表添加元素。列表的最大长度为2^32 - 1，也即每个列表支持超过40亿个元素。

Redis list的实现为一个双向链表，即可以支持反向查找和遍历，更方便操作，不过带来了部分额外的内存开销，Redis内部的很多实现，包括发送缓冲队列等也都是用的这个数据结构。

应用场景
Redis list的应用场景非常多，也是Redis最重要的数据结构之一，比如twitter的关注列表、粉丝列表等都可以用Redis的list结构来实现，再比如有的应用使用Redis的list类型实现一个简单的轻量级消息队列，生产者push，消费者pop/bpop。

redis 127.0.0.1:6379> LPUSH runoobkey redis
(integer) 1
redis 127.0.0.1:6379> LPUSH runoobkey mongodb
(integer) 2
redis 127.0.0.1:6379> LPUSH runoobkey mysql
(integer) 3
redis 127.0.0.1:6379> LRANGE runoobkey 0 10

“mysql”
“mongodb”
“redis”

Hash（字典）
Redis Hash对应Value内部实际就是一个HashMap，实际这里会有2种不同实现，这个Hash的成员比较少时Redis为了节省内存会采用类似一维数组的方式来紧凑存储，而不会采用真正的HashMap结构，对应的value redisObject的encoding为zipmap,当成员数量增大时会自动转成真正的HashMap,此时encoding为ht。

127.0.0.1:6379> HMSET runoobkey name “redis tutorial”
127.0.0.1:6379> HGETALL runoobkey

“name”
“redis tutorial”
“description”
“redis basic commands for caching”
“likes”
“20”
“visitors”
“23000”

Set（集合）

Redis的Set是string类型的无序集合。集合成员是唯一的，这就意味着集合中不能出现重复的数据。
Redis 中集合是通过哈希表实现的，所以添加，删除，查找的复杂度都是O(1)。
集合中最大的成员数为 232 - 1 (4294967295, 每个集合可存储40多亿个成员)。

redis 127.0.0.1:6379> SADD runoobkey redis
(integer) 1
redis 127.0.0.1:6379> SADD runoobkey mongodb
(integer) 1
redis 127.0.0.1:6379> SADD runoobkey mysql
(integer) 1
redis 127.0.0.1:6379> SADD runoobkey mysql
(integer) 0
redis 127.0.0.1:6379> SMEMBERS runoobkey

“mysql”
“mongodb”
“redis”

Sorted Set（有序集合）

Redis 有序集合和集合一样也是string类型元素的集合,且不允许重复的成员。
不同的是每个元素都会关联一个double类型的分数。redis正是通过分数来为集合中的成员进行从小到大的排序。
有序集合的成员是唯一的,但分数(score)却可以重复。
集合是通过哈希表实现的，所以添加，删除，查找的复杂度都是O(1)。集合中最大的成员数为 232 - 1 (4294967295, 每个集合可存储40多亿个成员)。

redis 127.0.0.1:6379> ZADD runoobkey 1 redis
(integer) 1
redis 127.0.0.1:6379> ZADD runoobkey 2 mongodb
(integer) 1
redis 127.0.0.1:6379> ZADD runoobkey 3 mysql
(integer) 1
redis 127.0.0.1:6379> ZADD runoobkey 3 mysql
(integer) 0
redis 127.0.0.1:6379> ZADD runoobkey 4 mysql
(integer) 0
redis 127.0.0.1:6379> ZRANGE runoobkey 0 10 WITHSCORES

“redis”
“1”
“mongodb”
“2”
“mysql”
“4”

Redis客户端连接工具
redisclient-win32.x86.1.5

SrpingBoot整合Redis
Maven依赖

org.springframework.boot
spring-boot-starter-parent
2.0.0.RELEASE

org.springframework.boot
spring-boot-starter-web

org.springframework.boot
spring-boot-starter-data-redis

<build>
	<plugins>
		<plugin>
			<groupId>org.apache.maven.plugins</groupId>
			<artifactId>maven-compiler-plugin</artifactId>
			<configuration>
				<source>1.8</source>
				<target>1.8</target>
			</configuration>
		</plugin>
		<plugin>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-maven-plugin</artifactId>
			<configuration>
				<maimClass>
					com.itmayiedu.controller.IndexController</maimClass>
			</configuration>
			<executions>
				<execution>
					<goals>
						<goal>repackage</goal>
					</goals>
				</execution>
			</executions>

		</plugin>
	</plugins>
</build>

配置文件
spring:
redis:
database: 0
host: 132.232.44.194
port: 6379
password: 123456
jedis:
pool:
max-active: 8
max-wait: -1
max-idle: 8
min-idle: 0
timeout: 10000

@Component
public class RedisService {

@Autowired
private StringRedisTemplate stringRedisTemplate;

public void set(String key, Object object, Long time) {
	// 存放String 类型
	if (object instanceof String) {
		setString(key, object);
	}
	// 存放 set类型
	if (object instanceof Set) {
		setSet(key, object);
	}
	// 设置有效期 以秒为单位
	stringRedisTemplate.expire(key, time, TimeUnit.SECONDS);
}

public void setString(String key, Object object) {
	// 如果是String 类型
	String value = (String) object;
	stringRedisTemplate.opsForValue().set(key, value);
}

public void setSet(String key, Object object) {
	Set<String> value = (Set<String>) object;
	for (String oj : value) {
		stringRedisTemplate.opsForSet().add(key, oj);
	}
}

public String getString(String key) {
	return stringRedisTemplate.opsForValue().get(key);
}

}
代码整合Redis
@RestController
public class IndexControler {

@Autowired
private RedisService redisService;

@RequestMapping("/setString")
public String setString(String key, String value) {
	redisService.set(key, value, 60l);
	return "success";
}

@RequestMapping("/getString")
public String getString(String key) {
	return redisService.getString(key);
}

@RequestMapping("/setSet")
public String setSet() {
	Set<String> set = new HashSet<String>();
	set.add("yushengjun");
	set.add("lisi");
	redisService.setSet("setTest", set);
	return "success";
}

}
Redis发布订阅

Redis 发布订阅(pub/sub)是一种消息通信模式：发送者(pub)发送消息，订阅者(sub)接收消息。
Redis 客户端可以订阅任意数量的频道。
下图展示了频道 channel1 ，以及订阅这个频道的三个客户端 —— client2 、 client5 和 client1 之间的关系：

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

在我们实例中我们创建了订阅频道名为 redisChat:

redis 127.0.0.1:6379> SUBSCRIBE redisChat
Reading messages… (press Ctrl-C to quit)

“subscribe”
“redisChat”
(integer) 1

现在，我们先重新开启个 redis 客户端，然后在同一个频道 redisChat 发布两次消息，订阅者就能接收到消息。

redis 127.0.0.1:6379> PUBLISH redisChat “Redis is a great caching technique”

(integer) 1

redis 127.0.0.1:6379> PUBLISH redisChat “Learn redis by runoob.com”

(integer) 1

订阅者的客户端会显示如下消息

“message”
“redisChat”
“Redis is a great caching technique”
“message”
“redisChat”
“Learn redis by runoob.com”

Redis事务

Redis主从复制
概述
1、redis的复制功能是支持多个数据库之间的数据同步。一类是主数据库（master）一类是从数据库（slave），主数据库可以进行读写操作，当发生写操作的时候自动将数据同步到从数据库，而从数据库一般是只读的，并接收主数据库同步过来的数据，一个主数据库可以有多个从数据库，而一个从数据库只能有一个主数据库。
2、通过redis的复制功能可以很好的实现数据库的读写分离，提高服务器的负载能力。主数据库主要进行写操作，而从数据库负责读操作。
主从复制过程：见下图

过程：
1：当一个从数据库启动时，会向主数据库发送sync命令，
2：主数据库接收到sync命令后会开始在后台保存快照（执行rdb操作），并将保存期间接收到的命令缓存起来
3：当快照完成后，redis会将快照文件和所有缓存的命令发送给从数据库。
4：从数据库收到后，会载入快照文件并执行收到的缓存的命令。
修改从Redis从配置文件
修改slave从redis中的 redis.conf文件
slaveof 192.168.33.130 6379
masterauth 123456— 主redis服务器配置了密码,则需要配置
一主多从如何实现

Redis哨兵机制
什么是哨兵机制
Redis的哨兵(sentinel) 系统用于管理多个 Redis 服务器,该系统执行以下三个任务:
· 监控(Monitoring): 哨兵(sentinel) 会不断地检查你的Master和Slave是否运作正常。
· 提醒(Notification):当被监控的某个 Redis出现问题时, 哨兵(sentinel) 可以通过 API 向管理员或者其他应用程序发送通知。
· 自动故障迁移(Automatic failover):当一个Master不能正常工作时，哨兵(sentinel) 会开始一次自动故障迁移操作,它会将失效Master的其中一个Slave升级为新的Master, 并让失效Master的其他Slave改为复制新的Master; 当客户端试图连接失效的Master时,集群也会向客户端返回新Master的地址,使得集群可以使用Master代替失效Master。
哨兵(sentinel) 是一个分布式系统,你可以在一个架构中运行多个哨兵(sentinel) 进程,这些进程使用流言协议(gossipprotocols)来接收关于Master是否下线的信息,并使用投票协议(agreement protocols)来决定是否执行自动故障迁移,以及选择哪个Slave作为新的Master.
每个哨兵(sentinel) 会向其它哨兵(sentinel)、master、slave定时发送消息,以确认对方是否”活”着,如果发现对方在指定时间(可配置)内未回应,则暂时认为对方已挂(所谓的”主观认为宕机” Subjective Down,简称sdown).
若“哨兵群”中的多数sentinel,都报告某一master没响应,系统才认为该master"彻底死亡"(即:客观上的真正down机,Objective Down,简称odown),通过一定的vote算法,从剩下的slave节点中,选一台提升为master,然后自动修改相关配置.
虽然哨兵(sentinel) 释出为一个单独的可执行文件 redis-sentinel ,但实际上它只是一个运行在特殊模式下的 Redis 服务器，你可以在启动一个普通 Redis 服务器时通过给定 --sentinel 选项来启动哨兵(sentinel).
哨兵(sentinel) 的一些设计思路和zookeeper非常类似
单个哨兵(sentinel)

哨兵模式修改配置
实现步骤:
1.拷贝到etc目录
cp sentinel.conf /usr/local/redis/etc
2.修改sentinel.conf配置文件
sentinel monitor mymast 192.168.110.133 6379 1 #主节点名称 IP 端口号选举次数
sentinel auth-pass mymaster 123456
3. 修改心跳检测 30毫秒
sentinel down-after-milliseconds mymaster 30
4.sentinel parallel-syncs mymaster 2 — 做多多少合格节点
5. 启动哨兵模式
./redis-server /usr/local/redis/etc/sentinel.conf --sentinel &
6. 停止哨兵模式

Redis持久化
什么是Redis持久化
什么是Redis持久化,就是将内存数据保存到硬盘。
Redis 持久化存储 (AOF 与 RDB 两种模式)
RDB持久化
RDB 是以二进制文件，是在某个时间点将数据写入一个临时文件，持久化结束后，用这个临时文件替换上次持久化的文件，达到数据恢复。
优点：使用单独子进程来进行持久化，主进程不会进行任何 IO 操作，保证了 redis 的高性能
缺点：RDB 是间隔一段时间进行持久化，如果持久化之间 redis 发生故障，会发生数据丢失。所以这种方式更适合数据要求不严谨的时候
这里说的这个执行数据写入到临时文件的时间点是可以通过配置来自己确定的，通过配置redis 在 n 秒内如果超过 m 个 key 被修改这执行一次 RDB 操作。这个操作就类似于在这个时间点来保存一次 Redis 的所有数据，一次快照数据。所有这个持久化方法也通常叫做 snapshots。
RDB 默认开启，redis.conf 中的具体配置参数如下；
#dbfilename：持久化数据存储在本地的文件
dbfilename dump.rdb
#dir：持久化数据存储在本地的路径，如果是在/redis/redis-3.0.6/src下启动的redis-cli，则数据会存储在当前src目录下
dir ./
##snapshot触发的时机，save
##如下为900秒后，至少有一个变更操作，才会snapshot
##对于此值的设置，需要谨慎，评估系统的变更操作密集程度
##可以通过“save “””来关闭snapshot功能
#save时间，以下分别表示更改了1个key时间隔900s进行持久化存储；更改了10个key300s进行存储；更改10000个key60s进行存储。
save 900 1
save 300 10
save 60 10000
##当snapshot时出现错误无法继续时，是否阻塞客户端“变更操作”，“错误”可能因为磁盘已满/磁盘故障/OS级别异常等
stop-writes-on-bgsave-error yes
##是否启用rdb文件压缩，默认为“yes”，压缩往往意味着“额外的cpu消耗”，同时也意味这较小的文件尺寸以及较短的网络传输时间
rdbcompression yes

AOF持久化
Append-only file，将“操作 + 数据”以格式化指令的方式追加到操作日志文件的尾部，在 append 操作返回后(已经写入到文件或者即将写入)，才进行实际的数据变更，“日志文件”保存了历史所有的操作过程；当 server 需要数据恢复时，可以直接 replay 此日志文件，即可还原所有的操作过程。AOF 相对可靠，它和 mysql 中 bin.log、apache.log、zookeeper 中 txn-log 简直异曲同工。AOF 文件内容是字符串，非常容易阅读和解析。
优点：可以保持更高的数据完整性，如果设置追加 file 的时间是 1s，如果 redis 发生故障，最多会丢失 1s 的数据；且如果日志写入不完整支持 redis-check-aof 来进行日志修复；AOF 文件没被 rewrite 之前（文件过大时会对命令进行合并重写），可以删除其中的某些命令（比如误操作的 flushall）。
缺点：AOF 文件比 RDB 文件大，且恢复速度慢。
我们可以简单的认为 AOF 就是日志文件，此文件只会记录“变更操作”(例如：set/del 等)，如果 server 中持续的大量变更操作，将会导致 AOF 文件非常的庞大，意味着 server 失效后，数据恢复的过程将会很长；事实上，一条数据经过多次变更，将会产生多条 AOF 记录，其实只要保存当前的状态，历史的操作记录是可以抛弃的；因为 AOF 持久化模式还伴生了“AOF rewrite”。
AOF 的特性决定了它相对比较安全，如果你期望数据更少的丢失，那么可以采用 AOF 模式。如果 AOF 文件正在被写入时突然 server 失效，有可能导致文件的最后一次记录是不完整，你可以通过手工或者程序的方式去检测并修正不完整的记录，以便通过 aof 文件恢复能够正常；同时需要提醒，如果你的 redis 持久化手段中有 aof，那么在 server 故障失效后再次启动前，需要检测 aof 文件的完整性。
AOF 默认关闭，开启方法，修改配置文件 reds.conf：appendonly yes
##此选项为aof功能的开关，默认为“no”，可以通过“yes”来开启aof功能
##只有在“yes”下，aof重写/文件同步等特性才会生效
appendonly yes

##指定aof文件名称
appendfilename appendonly.aof

##指定aof操作中文件同步策略，有三个合法值：always everysec no,默认为everysec
appendfsync everysec
##在aof-rewrite期间，appendfsync是否暂缓文件同步，"no"表示“不暂缓”，“yes”表示“暂缓”，默认为“no”
no-appendfsync-on-rewrite no

##aof文件rewrite触发的最小文件尺寸(mb,gb),只有大于此aof文件大于此尺寸是才会触发rewrite，默认“64mb”，建议“512mb”
auto-aof-rewrite-min-size 64mb

##相对于“上一次”rewrite，本次rewrite触发时aof文件应该增长的百分比。
##每一次rewrite之后，redis都会记录下此时“新aof”文件的大小(例如A)，那么当aof文件增长到A*(1 + p)之后
##触发下一次rewrite，每一次aof记录的添加，都会检测当前aof文件的尺寸。
auto-aof-rewrite-percentage 100

AOF 是文件操作，对于变更操作比较密集的 server，那么必将造成磁盘 IO 的负荷加重；此外 linux 对文件操作采取了“延迟写入”手段，即并非每次 write 操作都会触发实际磁盘操作，而是进入了 buffer 中，当 buffer 数据达到阀值时触发实际写入(也有其他时机)，这是 linux 对文件系统的优化，但是这却有可能带来隐患，如果 buffer 没有刷新到磁盘，此时物理机器失效(比如断电)，那么有可能导致最后一条或者多条 aof 记录的丢失。通过上述配置文件，可以得知 redis 提供了 3 中 aof 记录同步选项：
always：每一条 aof 记录都立即同步到文件，这是最安全的方式，也以为更多的磁盘操作和阻塞延迟，是 IO 开支较大。
everysec：每秒同步一次，性能和安全都比较中庸的方式，也是 redis 推荐的方式。如果遇到物理服务器故障，有可能导致最近一秒内 aof 记录丢失(可能为部分丢失)。
no：redis 并不直接调用文件同步，而是交给操作系统来处理，操作系统可以根据 buffer 填充情况 / 通道空闲时间等择机触发同步；这是一种普通的文件操作方式。性能较好，在物理服务器故障时，数据丢失量会因 OS 配置有关。
其实，我们可以选择的太少，everysec 是最佳的选择。如果你非常在意每个数据都极其可靠，建议你选择一款“关系性数据库”吧。
AOF 文件会不断增大，它的大小直接影响“故障恢复”的时间, 而且 AOF 文件中历史操作是可以丢弃的。AOF rewrite 操作就是“压缩”AOF 文件的过程，当然 redis 并没有采用“基于原 aof 文件”来重写的方式，而是采取了类似 snapshot 的方式：基于 copy-on-write，全量遍历内存中数据，然后逐个序列到 aof 文件中。因此 AOF rewrite 能够正确反应当前内存数据的状态，这正是我们所需要的；rewrite 过程中，对于新的变更操作将仍然被写入到原 AOF 文件中，同时这些新的变更操作也会被 redis 收集起来(buffer，copy-on-write 方式下，最极端的可能是所有的 key 都在此期间被修改，将会耗费 2 倍内存)，当内存数据被全部写入到新的 aof 文件之后，收集的新的变更操作也将会一并追加到新的 aof 文件中，此后将会重命名新的 aof 文件为 appendonly.aof, 此后所有的操作都将被写入新的 aof 文件。如果在 rewrite 过程中，出现故障，将不会影响原 AOF 文件的正常工作，只有当 rewrite 完成之后才会切换文件，因为 rewrite 过程是比较可靠的。
触发 rewrite 的时机可以通过配置文件来声明，同时 redis 中可以通过 bgrewriteaof 指令人工干预。
redis-cli -h ip -p port bgrewriteaof
因为 rewrite 操作 /aof 记录同步 /snapshot 都消耗磁盘 IO，redis 采取了“schedule”策略：无论是“人工干预”还是系统触发，snapshot 和 rewrite 需要逐个被执行。
AOF rewrite 过程并不阻塞客户端请求。系统会开启一个子进程来完成。

AOF与RDB区别
AOF 和 RDB 各有优缺点，这是有它们各自的特点所决定：
RDB
RDB是在某个时间点将数据写入一个临时文件，持久化结束后，用这个临时文件替换上次持久化的文件，达到数据恢复。
优点：使用单独子进程来进行持久化，主进程不会进行任何IO操作，保证了redis的高性能
缺点：RDB是间隔一段时间进行持久化，如果持久化之间redis发生故障，会发生数据丢失。所以这种方式更适合数据要求不严谨的时候
AOF
Append-only file，将“操作 + 数据”以格式化指令的方式追加到操作日志文件的尾部，在append操作返回后(已经写入到文件或者即将写入)，才进行实际的数据变更，“日志文件”保存了历史所有的操作过程；当server需要数据恢复时，可以直接replay此日志文件，即可还原所有的操作过程。AOF相对可靠，它和mysql中bin.log、apache.log、zookeeper中txn-log简直异曲同工。AOF文件内容是字符串，非常容易阅读和解析。
优点：可以保持更高的数据完整性，如果设置追加file的时间是1s，如果redis发生故障，最多会丢失1s的数据；且如果日志写入不完整支持redis-check-aof来进行日志修复；AOF文件没被rewrite之前（文件过大时会对命令进行合并重写），可以删除其中的某些命令（比如误操作的flushall）。
缺点：AOF文件比RDB文件大，且恢复速度慢。

Redis事物
Redis事物
Redis 事务可以一次执行多个命令，并且带有以下两个重要的保证：
事务是一个单独的隔离操作：事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。
事务是一个原子操作：事务中的命令要么全部被执行，要么全部都不执行。
一个事务从开始到执行会经历以下三个阶段：
开始事务。
命令入队。
执行事务。

以下是一个事务的例子，它先以 MULTI 开始一个事务，然后将多个命令入队到事务中，最后由 EXEC 命令触发事务，一并执行事务中的所有命令：
redis 127.0.0.1:6379> MULTI
OK

redis 127.0.0.1:6379> SET book-name “Mastering C++ in 21 days”
QUEUED

redis 127.0.0.1:6379> GET book-name
QUEUED

redis 127.0.0.1:6379> SADD tag “C++” “Programming” “Mastering Series”
QUEUED

redis 127.0.0.1:6379> SMEMBERS tag
QUEUED

redis 127.0.0.1:6379> EXEC

OK
“Mastering C++ in 21 days”
(integer) 3
1. “Mastering Series”
2. “C++”
3. “Programming”(⊙o⊙)…

Redis集群事物
redis集群对象JedisCluster不支持事务，但是，集群里边的每个节点支持事务。
SpringBoot操作Redis事务
public void setString(String key, Object object) {
stringRedisTemplate.setEnableTransactionSupport(true);
// 开启事务
stringRedisTemplate.multi();
try {
// 如果是String 类型
String value = (String) object;
stringRedisTemplate.opsForValue().set(key, value);
} catch (Exception e) {
// 回滚
stringRedisTemplate.discard();
} finally {
// 提交
stringRedisTemplate.exec();
}

Redis+ehCache实现两级级缓存
spring boot中集成了spring cache，并有多种缓存方式的实现，如：Redis、Caffeine、JCache、EhCache等等。但如果只用一种缓存，要么会有较大的网络消耗（如Redis），要么就是内存占用太大（如Caffeine这种应用内存缓存）。在很多场景下，可以结合起来实现一、二级缓存的方式，能够很大程度提高应用的处理效率。
内容说明：
缓存、两级缓存
spring cache：主要包含spring cache定义的接口方法说明和注解中的属性说明
spring boot + spring cache：RedisCache实现中的缺陷
caffeine简介
spring boot + spring cache 实现两级缓存（redis + caffeine）

缓存、两级缓存
简单的理解，缓存就是将数据从读取较慢的介质上读取出来放到读取较快的介质上，如磁盘–>内存。平时我们会将数据存储到磁盘上，如：数据库。如果每次都从数据库里去读取，会因为磁盘本身的IO影响读取速度，所以就有了像redis这种的内存缓存。可以将数据读取出来放到内存里，这样当需要获取数据时，就能够直接从内存中拿到数据返回，能够很大程度的提高速度。但是一般redis是单独部署成集群，所以会有网络IO上的消耗，虽然与redis集群的链接已经有连接池这种工具，但是数据传输上也还是会有一定消耗。所以就有了应用内缓存，如：caffeine。当应用内缓存有符合条件的数据时，就可以直接使用，而不用通过网络到redis中去获取，这样就形成了两级缓存。应用内缓存叫做一级缓存，远程缓存（如redis）叫做二级缓存

相关代码

Maven环境准备

org.springframework.boot spring-boot-starter-parent 2.0.0.RELEASE com.alibaba fastjson 1.2.47 org.apache.commons commons-lang3 org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-data-redis

	<!--开启 cache 缓存 -->
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-cache</artifactId>
	</dependency>
	<!-- ehcache缓存 -->
	<dependency>
		<groupId>net.sf.ehcache</groupId>
		<artifactId>ehcache</artifactId>
		<version>2.9.1</version><!--$NO-MVN-MAN-VER$ -->
	</dependency>

	<dependency>
		<groupId>org.mybatis.spring.boot</groupId>
		<artifactId>mybatis-spring-boot-starter</artifactId>
		<version>1.1.1</version>
	</dependency>
	<!-- mysql 依赖 -->
	<dependency>
		<groupId>mysql</groupId>
		<artifactId>mysql-connector-java</artifactId>
	</dependency>
</dependencies>


<build>
	<plugins>
		<plugin>
			<groupId>org.apache.maven.plugins</groupId>
			<artifactId>maven-compiler-plugin</artifactId>
			<configuration>
				<source>1.8</source>
				<target>1.8</target>
			</configuration>
		</plugin>
		<plugin>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-maven-plugin</artifactId>
			<configuration>
				<maimClass>
					com.itmayiedu.controller.IndexController</maimClass>
			</configuration>
			<executions>
				<execution>
					<goals>
						<goal>repackage</goal>
					</goals>
				</execution>
			</executions>

		</plugin>
	</plugins>
</build>

Ehcache工具类
@Component
public class EhCacheUtils {

// @Autowired
// private CacheManager cacheManager;
@Autowired
private EhCacheCacheManager ehCacheCacheManager;

// 添加本地缓存 (相同的key 会直接覆盖)
public void put(String cacheName, String key, Object value) {
	Cache cache = ehCacheCacheManager.getCacheManager().getCache(cacheName);
	Element element = new Element(key, value);
	cache.put(element);
}

// 获取本地缓存
public Object get(String cacheName, String key) {
	Cache cache = ehCacheCacheManager.getCacheManager().getCache(cacheName);
	Element element = cache.get(key);
	return element == null ? null : element.getObjectValue();
}

public void remove(String cacheName, String key) {
	Cache cache = ehCacheCacheManager.getCacheManager().getCache(cacheName);
	cache.remove(key);
}

}

项目整合Ehcache环境
@Service
public class UserService {
@Autowired
private EhCacheUtils ehCacheUtils;
private static final String CACHENAME_USERCACHE = “userCache”;
@Autowired
private RedisService redisService;
@Autowired
private UserMapper userMapper;

public Users getUser(Long id) {
	String key = this.getClass().getName() + "-" + Thread.currentThread().getStackTrace()[1].getMethodName()
			+ "-id:" + id;
	// 1.先查找一级缓存(本地缓存),如果本地缓存有数据直接返回
	Users ehUser = (Users) ehCacheUtils.get(CACHENAME_USERCACHE, key);
	if (ehUser != null) {
		System.out.println("使用key:" + key + ",查询一级缓存 ehCache 获取到ehUser:" + JSONObject.toJSONString(ehUser));
		return ehUser;
	}
	// 2. 如果本地缓存没有该数据，直接查询二级缓存(redis)
	String redisUserJson = redisService.getString(key);
	if (!StringUtils.isEmpty(redisUserJson)) {
		// 将json 转换为对象(如果二级缓存redis中有数据直接返回二级缓存)
		JSONObject jsonObject = new JSONObject();
		Users user = jsonObject.parseObject(redisUserJson, Users.class);
		// 更新一级缓存
		ehCacheUtils.put(CACHENAME_USERCACHE, key, user);
		System.out.println("使用key:" + key + ",查询二级缓存 redis 获取到ehUser:" + JSONObject.toJSONString(user));
		return user;
	}
	// 3. 如果二级缓存redis中也没有数据,查询数据库
	Users user = userMapper.getUser(id);
	if (user == null) {
		return null;
	}
	// 更新一级缓存和二级缓存
	String userJson = JSONObject.toJSONString(user);
	redisService.setString(key, userJson);
	ehCacheUtils.put(CACHENAME_USERCACHE, key, user);
	System.out.println("使用key:" + key + "，一级缓存和二级都没有数据,直接查询db" + userJson);
	return user;
}

}

Redis集群方式
3.0版本之前

3.0版本之前的redis是不支持集群的，我们的redis如果想要集群的话，就需要一个中间件，然后这个中间件负责将我们需要存入redis中的数据的key通过一套算法计算得出一个值。然后根据这个值找到对应的redis节点，将这些数据存在这个redis的节点中。

   在取值的时候，同样先将key进行计算，得到对应的值，然后就去找对应的redis节点，从对应的节点中取出对应的值。

这样做有很多不好的地方，比如说我们的这些计算都需要在系统中去进行，所以会增加系统的负担。还有就是这种集群模式下，某个节点挂掉，其他的节点无法知道。而且也不容易对每个节点进行负载均衡。
常见集群方案
1.官方方案redis-cluster搭建实战
2.客户端分片技术（不推荐），扩容/缩容时，必须手动调整分片程序，出现故障不能自动转移
3.可以使用主从复制方式（不推荐）
4.使用一些代理工具

Redis-cluster原理

Redis 是一个开源的 key-value 存储系统，由于出众的性能，大部分互联网企业都用来做服务器端缓存。Redis 在3.0版本前只支持单实例模式，虽然支持主从模式、哨兵模式部署来解决单点故障，但是现在互联网企业动辄大几百G的数据，可完全是没法满足业务的需求，所以，Redis 在 3.0 版本以后就推出了集群模式。
Redis 集群采用了P2P的模式，完全去中心化。Redis 把所有的 Key 分成了 16384 个 slot，每个 Redis 实例负责其中一部分 slot 。集群中的
所有信息（节点、端口、slot等），都通过节点之间定期的数据交换而更新。
Redis 客户端可以在任意一个 Redis 实例发出请求，如果所需数据不在该实例中，通过重定向命令引导客户端访问所需的实例。

在这个图中，每一个蓝色的圈都代表着一个redis的服务器节点。它们任何两个节点之间都是相互连通的。客户端可以与任何一个节点相连接，然后就可以访问集群中的任何一个节点。对其进行存取和其他操作。
　　　　那么redis是怎么做到的呢？首先，在redis的每一个节点上，都有这么两个东西，一个是插槽（slot）可以理解为是一个可以存储两个数值的一个变量这个变量的取值范围是：0-16383。还有一个就是cluster我个人把这个cluster理解为是一个集群管理的插件。当我们的存取的key到达的时候，redis会根据crc16的算法得出一个结果，然后把结果对 16384 求余数，这样每个 key 都会对应一个编号在 0-16383 之间的哈希槽，通过这个值，去找到对应的插槽所对应的节点，然后直接自动跳转到这个对应的节点上进行存取操作。

还有就是因为如果集群的话，是有好多个redis一起工作的，那么，就需要这个集群不是那么容易挂掉，所以呢，理论上就应该给集群中的每个节点至少一个备用的redis服务。这个备用的redis称为从节点（slave）。那么这个集群是如何判断是否有某个节点挂掉了呢？

首先要说的是，每一个节点都存有这个集群所有主节点以及从节点的信息。

它们之间通过互相的ping-pong判断是否节点可以连接上。如果有一半以上的节点去ping一个节点的时候没有回应，集群就认为这个节点宕机了，然后去连接它的备用节点。如果某个节点和所有从节点全部挂掉，我们集群就进入faill状态。还有就是如果有一半以上的主节点宕机，那么我们集群同样进入发力了状态。这就是我们的redis的投票机制，具体原理如下图所示：

这套架构的特点：

分片算法：基于 slot hash桶；
分片实例之间相互独立，每组一个master 实例和多个slave；
路由信息存放到第三方存储组件，如 zookeeper 或etcd
旁路组件探活
搭建集群方案

安装部署任何一个应用其实都很简单，只要安装步骤一步一步来就行了。下面说一下 Redis 集群搭建规划，由于集群至少需要6个节点（3主3从模式），所以，没有这么多机器给我玩，我本地也起不了那么多虚拟机（电脑太烂），现在计划是在一台机器上模拟一个集群，当然，这和生产环境的集群搭建没本质区别。
创建文件夹
我们计划集群中 Redis 节点的端口号为 9001-9006 ，端口号即集群下各实例文件夹。数据存放在端口号/data 文件夹中。

mkdir /usr/local/redis-cluster
cd redis-cluster/
mkdir -p 9001/data 9002/data 9003/data 9004/data 9005/data 9006/data

复制脚本
在 /usr/local/redis-cluster 下创建 bin 文件夹，用来存放集群运行脚本，并把安装好的 Redis 的 src 路径下的运行脚本拷贝过来。看命令：

Cd /usr/local/redis-cluster
mkdir bin

Cd /usr/local/redis-3.2.9/src

cp mkreleasehdr.sh redis-benchmark redis-check-aof redis-cli redis-server redis-trib.rb /usr/local/redis-cluster/bin

复制一个新 Redis 实例

我们现在从已安装好的 Redis 中复制一个新的实例到 9001 文件夹，并修改 redis.conf 配置。
cp -r /usr/local/redis /usr/local/redis-cluster/9001

注意，修改 redis.conf 配置和单点唯一区别是下图部分，其余还是常规的这几项：

port 9001（每个节点的端口号）
daemonize yes
bind 192.168.119.131（绑定当前机器 IP）
dir /usr/local/redis-cluster/9001/data/（数据文件存放位置）
pidfile /var/run/redis_9001.pid（pid 9001和port要对应）
cluster-enabled yes（启动集群模式）
cluster-config-file nodes9001.conf（9001和port要对应）
cluster-node-timeout 15000
appendonly yes

再复制出五个新 Redis 实例
我们已经完成了一个节点了，其实接下来就是机械化的再完成另外五个节点，其实可以这么做：把 9001 实例复制到另外五个文件夹中，唯一要修改的就是 redis.conf 中的所有和端口的相关的信息即可，其实就那么四个位置。开始操作，看图：

\cp -rf /usr/local/redis-cluster/9001/* /usr/local/redis-cluster/9002
\cp -rf /usr/local/redis-cluster/9001/* /usr/local/redis-cluster/9003
\cp -rf /usr/local/redis-cluster/9001/* /usr/local/redis-cluster/9004
\cp -rf /usr/local/redis-cluster/9001/* /usr/local/redis-cluster/9005
\cp -rf /usr/local/redis-cluster/9001/* /usr/local/redis-cluster/9006

\cp -rf 命令是不使用别名来复制，因为 cp 其实是别名 cp -i，操作时会有交互式确认，比较烦人。

修改 9002-9006 的 redis.conf 文件

其实非常简单了，你通过搜索会发现其实只有四个点需要修改，我们全局替换下吧，进入相应的节点文件夹，做替换就好了。命令非常简单，看图：

vi /usr/local/redis-cluster/9002/redis/etc/redis.conf

vi /usr/local/redis-cluster/9003/redis/etc/redis.conf
vi /usr/local/redis-cluster/9004/redis/etc/redis.conf
vi /usr/local/redis-cluster/9005/redis/etc/redis.conf
vi /usr/local/redis-cluster/9006/redis/etc/redis.conf

%s/9001/9002
%s/9001/9003
%s/9001/9004
%s/9001/9005
%s/9001/9006

其实我们也就是替换了下面这四行：

port 9002
dir /usr/local/redis-cluster/9002/data/
cluster-config-file nodes-9002.conf
pidfile /var/run/redis_9002.pid

启动9001-9006六个节点

/usr/local/redis/bin/redis-server /usr/local/redis-cluster/9001/redis/etc/redis.conf

/usr/local/redis/bin/redis-server /usr/local/redis-cluster/9002/redis/etc/redis.conf
/usr/local/redis/bin/redis-server /usr/local/redis-cluster/9003/redis/etc/redis.conf
/usr/local/redis/bin/redis-server /usr/local/redis-cluster/9004/redis/etc/redis.conf
/usr/local/redis/bin/redis-server /usr/local/redis-cluster/9005/redis/etc/redis.conf
/usr/local/redis/bin/redis-server /usr/local/redis-cluster/9006/redis/etc/redis.conf

随便找一个节点测试试
/usr/local/redis-cluster/9001/redis/bin/redis-cli -c -h 192.168.212.150 -p 9001

(error) CLUSTERDOWN Hash slot not served

这是因为虽然我们配置并启动了 Redis 集群服务，但是他们暂时还并不在一个集群中，互相直接发现不了，而且还没有可存储的位置，就是所谓的slot（槽）。

安装集群所需软件

由于 Redis 集群需要使用 ruby 命令，所以我们需要安装 ruby 和相关接口。

yum install ruby
yum install rubygems
gem install redis 使用本地上传方式

gem install -l redis-3.2.1.gem

集群环境测试
/usr/local/redis-cluster/bin/redis-trib.rb create --replicas 1 192.168.212.150:9001 192.168.212.150:9002 192.168.212.150:9003 192.168.212.150:9004 192.168.212.150:9005 192.168.212.150:9006

简单解释一下这个命令：调用 ruby 命令来进行创建集群，–replicas 1 表示主从复制比例为 1:1，即一个主节点对应一个从节点；然后，默认给我们分配好了每个主节点和对应从节点服务，以及 solt 的大小，因为在 Redis 集群中有且仅有 16383 个 solt ，默认情况会给我们平均分配，当然你可以指定，后续的增减节点也可以重新分配。

M: 10222dee93f6a1700ede9f5424fccd6be0b2fb73 为主节点Id

S: 9ce697e49f47fec47b3dc290042f3cc141ce5aeb 192.168.119.131:9004 replicates 10222dee93f6a1700ede9f5424fccd6be0b2fb73 从节点下对应主节点Id

目前来看，9001-9003 为主节点，9004-9006 为从节点，并向你确认是否同意这么配置。输入 yes 后，会开始集群创建。

记住输入yes

验证集群环境

依然是通过客户端命令连接上，通过集群命令看一下状态和节点信息等。

总结一下
这一篇 Redis 集群部署搭建的文章真的是一步一步的走下来的，只要你安装我的步骤来，就保证你能成功搭建一个 Redis 集群玩玩，也可以这么说，除了步骤繁琐外，几乎不存在技术含量，估计能看完的人都感觉累（说真的，写这种文章真的很累人）。
接下来可能就是动态扩容、增加节点和减少节点，重新分配槽大小等，当然，还有最重要的就是怎么和我们程序结合起来，以及如何更好的把 Redis 缓存集群发挥出应有的效果，这些才是最重要的。

SpringBoot整合Redis集群

spring:
redis:
database: 0

host: 132.232.44.194

port: 6379

password: 123456

jedis:
  pool:
    max-active: 8
    max-wait: -1
    max-idle: 8
    min-idle: 0
timeout: 10000
cluster:
  nodes:
    - 192.168.212.149:9001
    - 192.168.212.149:9002
    - 192.168.212.149:9003
    - 192.168.212.149:9004
    - 192.168.212.149:9005
    - 192.168.212.149:9006

Redis集群的几个注意事项

Redis集群使用CRC16对key进行hash，集群固定使用16384对hash出来的值取模。因为取模结果一定在16384之内，所以集群中的sharding（分片）实际就是如何将16384个值在n个主节点间分配（从节点是主节点的近似副本，原因见3），如何分配取决于你的配置。
Redis生产级集群需要容灾，为此，一般部署为n个主+n*m个从。n大小主要取决于单机性能，m大小主要取决于机器稳定性。
Redis集群是弱一致性的，此处的一致，主要指主从之间的数据一致性。主要是因为redis在做数据更新时，不要求主从数据同步复制一定要成功。
集群最小的主数量为3，主数量应为奇数，以便做选举判决。

缓存穿透与缓存雪崩

缓存雪崩
缓存雪崩产生的原因

缓存雪崩通俗简单的理解就是：由于原有缓存失效（或者数据未加载到缓存中），新缓存未到期间（缓存正常从Redis中获取，如下图）所有原本应该访问缓存的请求都去查询数据库了，而对数据库CPU和内存造成巨大压力，严重的会造成数据库宕机，造成系统的崩溃。

缓存失效的时候如下图：

缓存失效时的雪崩效应对底层系统的冲击非常可怕！那有什么办法来解决这个问题呢？基本解决思路如下：

第一，大多数系统设计者考虑用加锁或者队列的方式保证来保证不会有大量的线程对数据库一次性进行读写，避免缓存失效时对数据库造成太大的压力，虽然能够在一定的程度上缓解了数据库的压力但是与此同时又降低了系统的吞吐量。

第二，分析用户的行为，尽量让缓存失效的时间均匀分布。

第三，如果是因为某台缓存服务器宕机，可以考虑做主备，比如：redis主备，但是双缓存涉及到更新事务的问题，update可能读到脏数据，需要好好解决。

解决方案

1：在缓存失效后，通过加锁或者队列来控制读数据库写缓存的线程数量。比如对某个key只允许一个线程查询数据和写缓存，其他线程等待。
@RequestMapping("/getUsers")
public Users getByUsers(Long id) {
// 1.先查询redis
String key = this.getClass().getName() + “-” + Thread.currentThread().getStackTrace()[1].getMethodName()
+ “-id:” + id;
String userJson = redisService.getString(key);
if (!StringUtils.isEmpty(userJson)) {
Users users = JSONObject.parseObject(userJson, Users.class);
return users;
}
Users user = null;
try {
lock.lock();
// 查询db
user = userMapper.getUser(id);
redisService.setSet(key, JSONObject.toJSONString(user));
} catch (Exception e) {

	} finally {
		lock.unlock(); // 释放锁
	}
	return user;
}

注意:加锁排队只是为了减轻数据库的压力，并没有提高系统吞吐量。假设在高并发下，缓存重建期间key是锁着的，这是过来1000个请求999个都在阻塞的。同样会导致用户等待超时，这是个治标不治本的方法。

2：不同的key，设置不同的过期时间，让缓存失效的时间点尽量均匀。
3：做二级缓存，A1为原始缓存，A2为拷贝缓存，A1失效时，可以访问A2，A1缓存失效时间设置为短期，A2设置为长期（此点为补充）

@Service
public class UserAvalanService {
@Autowired
private UserMapper userMapper;
@Autowired
private RedisService redisService;
private Lock lock = new ReentrantLock();

private String SIGN_KEY = "${NULL}";

public Users getByUsers(Long id) {
	// 1.先查询redis
	String key = this.getClass().getName() + "-" + Thread.currentThread().getStackTrace()[1].getMethodName()
			+ "-id:" + id;
	String userJson = redisService.getString(key);
	if (!StringUtils.isEmpty(userJson)) {
		Users users = JSONObject.parseObject(userJson, Users.class);
		return users;
	}
	Users user = null;
	try {
		lock.lock();
		// 查询db
		user = userMapper.getUser(id);
		redisService.setSet(key, JSONObject.toJSONString(user));
		lock.unlock();
	} catch (Exception e) {

	} finally {
		lock.unlock(); // 释放锁
	}
	return user;
}

public String getByUsers2(Long id) {
	// 1.先查询redis
	String key = this.getClass().getName() + "-" + Thread.currentThread().getStackTrace()[1].getMethodName()
			+ "-id:" + id;
	String userName = redisService.getString(key);
	if (!StringUtils.isEmpty(userName)) {
		return userName;
	}
	System.out.println("######开始发送数据库DB请求########");
	Users user = userMapper.getUser(id);
	String value = null;
	if (user == null) {
		// 标识为null
		value = SIGN_KEY;
	} else {
		value = user.getName();
	}
	redisService.setString(key, value);
	return value;
}

}

缓存穿透

缓存穿透是指用户查询数据，在数据库没有，自然在缓存中也不会有。这样就导致用户查询的时候，在缓存中找不到，每次都要去数据库再查询一遍，然后返回空。这样请求就绕过缓存直接查数据库，这也是经常提的缓存命中率问题。
解决的办法就是：如果查询数据库也为空，直接设置一个默认值存放到缓存，这样第二次到缓冲中获取就有值了，而不会继续访问数据库，这种办法最简单粗暴。

把空结果，也给缓存起来，这样下次同样的请求就可以直接返回空了，即可以避免当查询的值为空时引起的缓存穿透。同时也可以单独设置个缓存区域存储空值，对要查询的key进行预先校验，然后再放行给后面的正常缓存处理逻辑。

public String getByUsers2(Long id) {
// 1.先查询redis
String key = this.getClass().getName() + “-” + Thread.currentThread().getStackTrace()[1].getMethodName()
+ “-id:” + id;
String userName = redisService.getString(key);
if (!StringUtils.isEmpty(userName)) {
return userName;
}
System.out.println("######开始发送数据库DB请求########");
Users user = userMapper.getUser(id);
String value = null;
if (user == null) {
// 标识为null
value = SIGN_KEY;
} else {
value = user.getName();
}
redisService.setString(key, value);
return value;
}

注意：再给对应的ip存放真值的时候，需要先清除对应的之前的空缓存。

热点key

热点key:某个key访问非常频繁，当key失效的时候有打量线程来构建缓存，导致负载增加，系统崩溃。

解决办法：
①使用锁，单机用synchronized,lock等，分布式用分布式锁。
②缓存过期时间不设置，而是设置在key对应的value里。如果检测到存的时间超过过期时间则异步更新缓存。
③在value设置一个比过期时间t0小的过期时间值t1，当t1过期的时候，延长t1并做更新缓存操作。

上一篇： windows server 安装Mysql 5.7(解压缩版安装配置方法)

下一篇： Storm Trident示例Aggregator

缓存

缓存配置读取

订阅者的客户端会显示如下消息

host: 132.232.44.194

port: 6379

password: 123456

JavaWeb禁止浏览器缓存当前Web页面的方法

酷狗音乐缓存文件在哪里？PC版酷狗音乐怎么清除缓存内容？

优化SQL Server的内存占用之执行缓存

解读Django框架中的低层次缓存API

简介Django框架中可使用的各类缓存

详解Django缓存处理中Vary头部的使用

浅谈Python的Django框架中的缓存控制

Django框架下在URLconf中指定视图缓存的方法

Python的Django框架中模板碎片缓存简介

pptv本地视频怎么删除以及pptv的缓存文件在哪里？