Redis中的布隆过滤器（Bloom Filter）是一种空间效率很高的数据结构，用于判断一个元素是否在一个集合中。它的特点是能够快速判断元素不在集合中；如果判断元素在集合中，这种判断有一定的误判率。布隆过滤器通过使用多个哈希函数对元素进行处理，然后将结果映射到一个位数组上。如果一个元素对应的所有位都是1，那么布隆过滤器会认为这个元素可能存在；如果任何一位不是1，则元素肯定不在集合中。Redis布隆过滤器的应用场景包括但不限于：防止缓存穿透：通过布隆过滤器预判查询的键是否存在，避免对不存在的键进行数据库查询。去重：比如邮件列表、用户邀请等场景，使用布隆过滤器快速判断元素是否已经存在。在Spring Boot中使用Redis布隆过滤器，可以通过以下步骤实现：1. 添加依赖首先，在pom.xml中添加Redis和Spring Boot相关依赖。<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <dependency> <groupId>org.springframework.data</groupId> <artifactId>spring-data-redis</artifactId> <version>YOUR_VERSION_HERE</version> </dependency>确保版本兼容你的Spring Boot版本。2. 配置Redis在application.properties或application.yml中配置Redis服务器的地址、端口等信息。spring: redis: host: localhost port: 63793. 使用Redis布隆过滤器由于Spring Data Redis本身不直接支持布隆过滤器，你需要使用Redis命令通过execute方法直接调用。这里我们可以创建一个服务类，用于封装布隆过滤器的操作。import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.stereotype.Service; import java.util.Collections; @Service public class BloomFilterService { @Autowired private RedisTemplate<String, String> redisTemplate; // 添加元素到布隆过滤器 public Boolean addToBloomFilter(String filterName, String value) { String script = "return redis.call('BF.ADD', KEYS[1], ARGV[1])"; return redisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class), Collections.singletonList(filterName), value) == 1L; } // 检查元素是否可能存在于布隆过滤器中 public Boolean checkBloomFilter(String filterName, String value) { String script = "return redis.call('BF.EXISTS', KEYS[1], ARGV[1])"; return redisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class), Collections.singletonList(filterName), value) == 1L; } }注意，上述示例使用了Redis的BF.ADD和BF.EXISTS命令，这些命令需要Redis服务器安装了RedisBloom模块。4. 使用示例现在，你可以在你的业务代码中调用BloomFilterService的方法来添加元素到布隆过滤器或者检查元素是否可能存在。@Autowired private BloomFilterService bloomFilterService; public void demo() { String filterName = "myFilter"; String value = "testValue"; // 添加元素到布隆过滤器 bloomFilterService.addToBloomFilter(filterName, value); // 检查元素是否存在 boolean exists = bloomFilterService.checkBloomFilter(filterName, value); System.out.println("Element exists: " + exists); }这个简单的例子展示了如何在Spring Boot应用中通过Redis执行布隆过滤器的基本操作。在实际应用中，可能需要根据业务需求调整布隆过滤器的参数，比如预计元素数量、误判率等，以达到最优的性能和空间效率。
Redis集群本章是基于CentOS7下的Redis集群教程，包括：单机安装RedisRedis主从Redis分片集群1.单机安装Redis首先需要安装Redis所需要的依赖：yum install -y gcc tcl然后将课前资料提供的Redis安装包上传到虚拟机的任意目录：例如，我放到了/tmp目录：解压缩：tar -xvf redis-6.2.9.tar.gz解压后：进入redis目录：cd redis-6.2.9运行编译命令：make && make install如果没有出错，应该就安装成功了。然后修改redis.conf文件中的一些配置：# 绑定地址，默认是127.0.0.1，会导致只能在本地访问。修改为0.0.0.0则可以在任意IP访问 bind 0.0.0.0 # 数据库数量，设置为1 databases 1启动Redis：redis-server redis.conf停止redis服务：redis-cli shutdown2.Redis主从集群2.1.集群结构我们搭建的主从集群结构如图：共包含三个节点，一个主节点，两个从节点。这里我们会在同一台虚拟机中开启3个redis实例，模拟主从集群，信息如下：IPPORT角色172.100.xxx.xxx7001master172.100.xxx.xxx7002slave172.100.xxx.xxx7003slave2.2.准备实例和配置要在同一台虚拟机开启3个实例，必须准备三份不同的配置文件和目录，配置文件所在目录也就是工作目录。1）创建目录我们创建三个文件夹，名字分别叫7001、7002、7003：# 进入/tmp目录 cd /tmp # 创建目录 mkdir 7001 7002 7003如图：2）恢复原始配置修改redis-6.2.9/redis.conf文件，将其中的持久化模式改为默认的RDB模式，AOF保持关闭状态。# 开启RDB # save "" save 3600 1 save 300 100 save 60 10000 # 关闭AOF appendonly no3）拷贝配置文件到每个实例目录然后将redis-6.2.9/redis.conf文件拷贝到三个目录中（在/tmp目录执行下列命令）：# 方式一：逐个拷贝 cp redis-6.2.9/redis.conf 7001 cp redis-6.2.9/redis.conf 7002 cp redis-6.2.9/redis.conf 7003 # 方式二：管道组合命令，一键拷贝 echo 7001 7002 7003 | xargs -t -n 1 cp redis-6.2.9/redis.conf4）修改每个实例的端口、工作目录修改每个文件夹内的配置文件，将端口分别修改为7001、7002、7003，将rdb文件保存位置都修改为自己所在目录（在/tmp目录执行下列命令）：sed -i -e 's/6379/7001/g' -e 's/dir .\//dir \/tmp\/7001\//g' 7001/redis.conf sed -i -e 's/6379/7002/g' -e 's/dir .\//dir \/tmp\/7002\//g' 7002/redis.conf sed -i -e 's/6379/7003/g' -e 's/dir .\//dir \/tmp\/7003\//g' 7003/redis.conf5）修改每个实例的声明IP虚拟机本身有多个IP，为了避免将来混乱，我们需要在redis.conf文件中指定每一个实例的绑定ip信息，格式如下：# redis实例的声明 IP replica-announce-ip 172.100.xxx.xxx每个目录都要改，我们一键完成修改（在/tmp目录执行下列命令）：# 逐一执行 sed -i '1a replica-announce-ip 172.100.xxx.xxx' 7001/redis.conf sed -i '1a replica-announce-ip 172.100.xxx.xxx' 7002/redis.conf sed -i '1a replica-announce-ip 172.100.xxx.xxx' 7003/redis.conf # 或者一键修改 printf '%s\n' 7001 7002 7003 | xargs -I{} -t sed -i '1a replica-announce-ip 172.100.xxx.xxx' {}/redis.conf2.3.启动为了方便查看日志，我们打开3个ssh窗口，分别启动3个redis实例，启动命令：# 第1个 redis-server 7001/redis.conf # 第2个 redis-server 7002/redis.conf # 第3个 redis-server 7003/redis.conf启动后：如果要一键停止，可以运行下面命令：printf '%s\n' 7001 7002 7003 | xargs -I{} -t redis-cli -p {} shutdown2.4.开启主从关系现在三个实例还没有任何关系，要配置主从可以使用replicaof 或者slaveof（5.0以前）命令。有临时和永久两种模式：修改配置文件（永久生效）在redis.conf中添加一行配置：slaveof <masterip> <masterport>使用redis-cli客户端连接到redis服务，执行slaveof命令（重启后失效）：slaveof <masterip> <masterport>注意：在5.0以后新增命令replicaof，与salveof效果一致。这里我们为了演示方便，使用方式二。通过redis-cli命令连接7002，执行下面命令：# 连接 7002 redis-cli -p 7002 # 执行slaveof replicaof 172.100.xxx.xxx 7001通过redis-cli命令连接7003，执行下面命令：# 连接 7003 redis-cli -p 7003 # 执行slaveof replicaof 172.100.xxx.xxx 7001然后连接 7001节点，查看集群状态：# 连接 7001 redis-cli -p 7001 # 查看状态 info replication结果：2.5.测试执行下列操作以测试：利用redis-cli连接7001，执行set num 123利用redis-cli连接7002，执行get num，再执行set num 666利用redis-cli连接7003，执行get num，再执行set num 888可以发现，只有在7001这个master节点上可以执行写操作，7002和7003这两个slave节点只能执行读操作。3.搭建哨兵集群3.1.集群结构这里我们搭建一个三节点形成的Sentinel集群，来监管之前的Redis主从集群。如图：三个sentinel实例信息如下：节点IPPORTs1172.100.xxx.xxx27001s2172.100.xxx.xxx27002s3172.100.xxx.xxx270033.2.准备实例和配置要在同一台虚拟机开启3个实例，必须准备三份不同的配置文件和目录，配置文件所在目录也就是工作目录。我们创建三个文件夹，名字分别叫s1、s2、s3：# 进入/tmp目录 cd /tmp # 创建目录 mkdir s1 s2 s3如图：然后我们在s1目录创建一个sentinel.conf文件，添加下面的内容：port 27001 sentinel announce-ip 172.100.xxx.xxx sentinel monitor mymaster 172.100.xxx.xxx 7001 2 sentinel down-after-milliseconds mymaster 5000 sentinel failover-timeout mymaster 60000 dir "/tmp/s1"解读：port 27001：是当前sentinel实例的端口sentinel monitor mymaster 172.100.xxx.xxx 7001 2：指定主节点信息mymaster：主节点名称，自定义，任意写172.100.xxx.xxx 7001：主节点的ip和端口2：选举master时的quorum值然后将s1/sentinel.conf文件拷贝到s2、s3两个目录中（在/tmp目录执行下列命令）：# 方式一：逐个拷贝 cp s1/sentinel.conf s2 cp s1/sentinel.conf s3 # 方式二：管道组合命令，一键拷贝 echo s2 s3 | xargs -t -n 1 cp s1/sentinel.conf修改s2、s3两个文件夹内的配置文件，将端口分别修改为27002、27003：sed -i -e 's/27001/27002/g' -e 's/s1/s2/g' s2/sentinel.conf sed -i -e 's/27001/27003/g' -e 's/s1/s3/g' s3/sentinel.conf3.3.启动为了方便查看日志，我们打开3个ssh窗口，分别启动3个redis实例，启动命令：# 第1个 redis-sentinel s1/sentinel.conf # 第2个 redis-sentinel s2/sentinel.conf # 第3个 redis-sentinel s3/sentinel.conf启动后：3.4.测试尝试让master节点7001宕机，查看sentinel日志：查看7003的日志：查看7002的日志：4.搭建分片集群4.1.集群结构分片集群需要的节点数量较多，这里我们搭建一个最小的分片集群，包含3个master节点，每个master包含一个slave节点，结构如下：这里我们会在同一台虚拟机中开启6个redis实例，模拟分片集群，信息如下：IPPORT角色172.100.xxx.xxx7001master172.100.xxx.xxx7002master172.100.xxx.xxx7003master172.100.xxx.xxx8001slave172.100.xxx.xxx8002slave172.100.xxx.xxx8003slave4.2.准备实例和配置删除之前的7001、7002、7003这几个目录，重新创建出7001、7002、7003、8001、8002、8003目录：# 进入/tmp目录 cd /tmp # 删除旧的，避免配置干扰 rm -rf 7001 7002 7003 # 创建目录 mkdir 7001 7002 7003 8001 8002 8003在/tmp下准备一个新的redis.conf文件，内容如下：port 6379 # 开启集群功能 cluster-enabled yes # 集群的配置文件名称，不需要我们创建，由redis自己维护 cluster-config-file /tmp/6379/nodes.conf # 节点心跳失败的超时时间 cluster-node-timeout 5000 # 持久化文件存放目录 dir /tmp/6379 # 绑定地址 bind 0.0.0.0 # 让redis后台运行 daemonize yes # 注册的实例ip replica-announce-ip 172.100.xxx.xxx # 保护模式 protected-mode no # 数据库数量 databases 1 # 日志 logfile /tmp/6379/run.log将这个文件拷贝到每个目录下：# 进入/tmp目录 cd /tmp # 执行拷贝 echo 7001 7002 7003 8001 8002 8003 | xargs -t -n 1 cp redis.conf修改每个目录下的redis.conf，将其中的6379修改为与所在目录一致：# 进入/tmp目录 cd /tmp # 修改配置文件 printf '%s\n' 7001 7002 7003 8001 8002 8003 | xargs -I{} -t sed -i 's/6379/{}/g' {}/redis.conf4.3.启动因为已经配置了后台启动模式，所以可以直接启动服务：# 进入/tmp目录 cd /tmp # 一键启动所有服务 printf '%s\n' 7001 7002 7003 8001 8002 8003 | xargs -I{} -t redis-server {}/redis.conf通过ps查看状态：ps -ef | grep redis发现服务都已经正常启动：如果要关闭所有进程，可以执行命令：ps -ef | grep redis | awk '{print $2}' | xargs kill或者（推荐这种方式）：printf '%s\n' 7001 7002 7003 8001 8002 8003 | xargs -I{} -t redis-cli -p {} shutdown4.4.创建集群虽然服务启动了，但是目前每个服务之间都是独立的，没有任何关联。我们需要执行命令来创建集群，在Redis5.0之前创建集群比较麻烦，5.0之后集群管理命令都集成到了redis-cli中。1）Redis5.0之前Redis5.0之前集群命令都是用redis安装包下的src/redis-trib.rb来实现的。因为redis-trib.rb是有ruby语言编写的所以需要安装ruby环境。# 安装依赖 yum -y install zlib ruby rubygems gem install redis然后通过命令来管理集群：# 进入redis的src目录 cd /tmp/redis-6.2.9/src # 创建集群 ./redis-trib.rb create --replicas 1 172.100.xxx.xxx:7001 172.100.xxx.xxx:7002 172.100.xxx.xxx:7003 172.100.xxx.xxx:8001 172.100.xxx.xxx:8002 172.100.xxx.xxx:80032）Redis5.0以后我们使用的是Redis6.2.9版本，集群管理以及集成到了redis-cli中，格式如下：redis-cli --cluster create --cluster-replicas 1 172.100.xxx.xxx:7001 172.100.xxx.xxx:7002 172.100.xxx.xxx:7003 172.100.xxx.xxx:8001 172.100.xxx.xxx:8002 172.100.xxx.xxx:8003命令说明：redis-cli --cluster或者./redis-trib.rb：代表集群操作命令create：代表是创建集群--replicas 1或者--cluster-replicas 1 ：指定集群中每个master的副本个数为1，此时节点总数 ÷ (replicas + 1) 得到的就是master的数量。因此节点列表中的前n个就是master，其它节点都是slave节点，随机分配到不同master运行后的样子：这里输入yes，则集群开始创建：通过命令可以查看集群状态：redis-cli -p 7001 cluster nodes4.5.测试尝试连接7001节点，存储一个数据：# 连接 redis-cli -p 7001 # 存储数据 set num 123 # 读取数据 get num # 再次存储 set a 1结果悲剧了：集群操作时，需要给redis-cli加上-c参数才可以：redis-cli -c -p 7001这次可以了：5.设置密码对于哨兵模式的主从节点和哨兵节点，您可以按照以下方式配置密码：主从节点的redis.conf配置：主节点redis.conf：requirepass your_master_passwordmasterauth your_master_password从节点redis.conf：requirepass your_slave_passwordmasterauth your_master_password注意： master和slave都需要配置 requirepass 域masterauth，主要防止发生主从切换后，从节点连接不上主节点。哨兵节点的sentinel.conf配置：在sentinel monitor命令后，为每一个被监视的主节点添加认证信息。例如：sentinel monitor mymaster 127.0.0.1 6379 2 sentinel auth-pass mymaster your_master_password这里，mymaster是主节点的名字，your_master_password是之前为主节点设置的密码。确保哨兵节点可以通过密码连接到主节点和从节点。请注意，这些配置中的"your_master_password"和"your_slave_password"只是示例，您应该使用自己的强密码替换它们。配置完毕后，重启Redis和Sentinel服务，使得配置生效。对于分片集群，设置密码的方式与上述步骤类似你需要对每个分片（shard）的配置文件进行相应的密码设置。通常，这涉及到对每个shard的redis.conf文件进行编辑，添加或修改requirepass参数来设置密码。由于分片集群通常有多个节点，建议创建一个密码管理策略，确保每个节点都使用强密码，并定期更换密码。请注意，具体的操作步骤可能会根据你使用的Redis版本和分片集群的具体配置有所不同。因此，在进行任何配置更改之前，一定要查阅官方文档或咨询专业人士，确保你了解所有操作的风险和后果。
分布式缓存-- 基于Redis集群解决单机Redis存在的问题单机的Redis存在四大问题：0.学习目标1.Redis持久化Redis有两种持久化方案：RDB持久化AOF持久化1.1.RDB持久化RDB全称Redis Database Backup file（Redis数据备份文件），也被叫做Redis数据快照。简单来说就是把内存中的所有数据都记录到磁盘中。当Redis实例故障重启后，从磁盘读取快照文件，恢复数据。快照文件称为RDB文件，默认是保存在当前运行目录。1.1.1.执行时机RDB持久化在四种情况下会执行：执行save命令执行bgsave命令Redis停机时触发RDB条件时1）save命令执行下面的命令，可以立即执行一次RDB：save命令会导致主进程执行RDB，这个过程中其它所有命令都会被阻塞。只有在数据迁移时可能用到。2）bgsave命令下面的命令可以异步执行RDB：这个命令执行后会开启独立进程完成RDB，主进程可以持续处理用户请求，不受影响。3）停机时Redis停机时会执行一次save命令，实现RDB持久化。4）触发RDB条件Redis内部有触发RDB的机制，可以在redis.conf文件中找到，格式如下：# 900秒内，如果至少有1个key被修改，则执行bgsave ， 如果是save "" 则表示禁用RDB save 900 1 save 300 10 save 60 10000 RDB的其它配置也可以在redis.conf文件中设置：# 是否压缩 ,建议不开启，压缩也会消耗cpu，磁盘的话不值钱 rdbcompression yes # RDB文件名称 dbfilename dump.rdb # 文件保存的路径目录 dir ./ 1.1.2.RDB原理bgsave开始时会fork主进程得到子进程，子进程共享主进程的内存数据。完成fork后读取内存数据并写入 RDB 文件。fork采用的是copy-on-write技术：当主进程执行读操作时，访问共享内存；当主进程执行写操作时，则会拷贝一份数据，执行写操作。1.1.3.小结RDB方式bgsave的基本流程？fork主进程得到一个子进程，共享内存空间子进程读取内存数据并写入新的RDB文件用新RDB文件替换旧的RDB文件RDB会在什么时候执行？save 60 1000代表什么含义？默认是服务停止时代表60秒内至少执行1000次修改则触发RDBRDB的缺点？RDB执行间隔时间长，两次RDB之间写入数据有丢失的风险fork子进程、压缩、写出RDB文件都比较耗时1.2.AOF持久化1.2.1.AOF原理AOF全称为Append Only File（追加文件）。Redis处理的每一个写命令都会记录在AOF文件，可以看做是命令日志文件。1.2.2.AOF配置AOF默认是关闭的，需要修改redis.conf配置文件来开启AOF：# 是否开启AOF功能，默认是no appendonly yes # AOF文件的名称 appendfilename "appendonly.aof"AOF的命令记录的频率也可以通过redis.conf文件来配：# 表示每执行一次写命令，立即记录到AOF文件 appendfsync always # 写命令执行完先放入AOF缓冲区，然后表示每隔1秒将缓冲区数据写到AOF文件，是默认方案 appendfsync everysec # 写命令执行完先放入AOF缓冲区，由操作系统决定何时将缓冲区内容写回磁盘 appendfsync no三种策略对比：1.2.3.AOF文件重写因为是记录命令，AOF文件会比RDB文件大的多。而且AOF会记录对同一个key的多次写操作，但只有最后一次写操作才有意义。通过执行bgrewriteaof命令，可以让AOF文件执行重写功能，用最少的命令达到相同效果。如图，AOF原本有三个命令，但是set num 123 和 set num 666都是对num的操作，第二次会覆盖第一次的值，因此第一个命令记录下来没有意义。所以重写命令后，AOF文件内容就是：mset name jack num 666Redis也会在触发阈值时自动去重写AOF文件。阈值也可以在redis.conf中配置：# AOF文件比上次文件 增长超过多少百分比则触发重写 auto-aof-rewrite-percentage 100 # AOF文件体积最小多大以上才触发重写 auto-aof-rewrite-min-size 64mb 1.3.RDB与AOF对比RDB和AOF各有自己的优缺点，如果对数据安全性要求较高，在实际开发中往往会结合两者来使用。2.Redis主从2.1.搭建主从架构单节点Redis的并发能力是有上限的，要进一步提高Redis的并发能力，就需要搭建主从集群，实现读写分离。具体搭建流程参考[post cid="6437" cover="" size=""/]2.2.主从数据同步原理2.2.1.全量同步主从第一次建立连接时，会执行全量同步，将master节点的所有数据都拷贝给slave节点，流程：这里有一个问题，master如何得知salve是第一次来连接呢？？有几个概念，可以作为判断依据：Replication Id：简称replid，是数据集的标记，id一致则说明是同一数据集。每一个master都有唯一的replid，slave则会继承master节点的replidoffset：偏移量，随着记录在repl_baklog中的数据增多而逐渐增大。slave完成同步时也会记录当前同步的offset。如果slave的offset小于master的offset，说明slave数据落后于master，需要更新。因此slave做数据同步，必须向master声明自己的replication id 和offset，master才可以判断到底需要同步哪些数据。因为slave原本也是一个master，有自己的replid和offset，当第一次变成slave，与master建立连接时，发送的replid和offset是自己的replid和offset。master判断发现slave发送来的replid与自己的不一致，说明这是一个全新的slave，就知道要做全量同步了。master会将自己的replid和offset都发送给这个slave，slave保存这些信息。以后slave的replid就与master一致了。因此，master判断一个节点是否是第一次同步的依据，就是看replid是否一致。如图：完整流程描述：slave节点请求增量同步master节点判断replid，发现不一致，拒绝增量同步master将完整内存数据生成RDB，发送RDB到slaveslave清空本地数据，加载master的RDBmaster将RDB期间的命令记录在repl_baklog，并持续将log中的命令发送给slaveslave执行接收到的命令，保持与master之间的同步2.2.2.增量同步全量同步需要先做RDB，然后将RDB文件通过网络传输个slave，成本太高了。因此除了第一次做全量同步，其它大多数时候slave与master都是做增量同步。什么是增量同步？就是只更新slave与master存在差异的部分数据。如图：那么master怎么知道slave与自己的数据差异在哪里呢?2.2.3.repl_backlog原理master怎么知道slave与自己的数据差异在哪里呢?这就要说到全量同步时的repl_baklog文件了。这个文件是一个固定大小的数组，只不过数组是环形，也就是说角标到达数组末尾后，会再次从0开始读写，这样数组头部的数据就会被覆盖。repl_baklog中会记录Redis处理过的命令日志及offset，包括master当前的offset，和slave已经拷贝到的offset：slave与master的offset之间的差异，就是salve需要增量拷贝的数据了。随着不断有数据写入，master的offset逐渐变大，slave也不断的拷贝，追赶master的offset：直到数组被填满：此时，如果有新的数据写入，就会覆盖数组中的旧数据。不过，旧的数据只要是绿色的，说明是已经被同步到slave的数据，即便被覆盖了也没什么影响。因为未同步的仅仅是红色部分。但是，如果slave出现网络阻塞，导致master的offset远远超过了slave的offset：如果master继续写入新数据，其offset就会覆盖旧的数据，直到将slave现在的offset也覆盖：棕色框中的红色部分，就是尚未同步，但是却已经被覆盖的数据。此时如果slave恢复，需要同步，却发现自己的offset都没有了，无法完成增量同步了。只能做全量同步。2.3.主从同步优化主从同步可以保证主从数据的一致性，非常重要。可以从以下几个方面来优化Redis主从就集群：在master中配置repl-diskless-sync yes启用无磁盘复制，避免全量同步时的磁盘IO。Redis单节点上的内存占用不要太大，减少RDB导致的过多磁盘IO适当提高repl_baklog的大小，发现slave宕机时尽快实现故障恢复，尽可能避免全量同步限制一个master上的slave节点数量，如果实在是太多slave，则可以采用主-从-从链式结构，减少master压力主从从架构图：2.4.小结简述全量同步和增量同步区别？全量同步：master将完整内存数据生成RDB，发送RDB到slave。后续命令则记录在repl_baklog，逐个发送给slave。增量同步：slave提交自己的offset到master，master获取repl_baklog中从offset之后的命令给slave什么时候执行全量同步？slave节点第一次连接master节点时slave节点断开时间太久，repl_baklog中的offset已经被覆盖时什么时候执行增量同步？slave节点断开又恢复，并且在repl_baklog中能找到offset时3.Redis哨兵Redis提供了哨兵（Sentinel）机制来实现主从集群的自动故障恢复。3.1.哨兵原理3.1.1.集群结构和作用哨兵的结构如图：哨兵的作用如下：监控：Sentinel 会不断检查您的master和slave是否按预期工作自动故障恢复：如果master故障，Sentinel会将一个slave提升为master。当故障实例恢复后也以新的master为主通知：Sentinel充当Redis客户端的服务发现来源，当集群发生故障转移时，会将最新信息推送给Redis的客户端3.1.2.集群监控原理Sentinel基于心跳机制监测服务状态，每隔1秒向集群的每个实例发送ping命令：主观下线：如果某sentinel节点发现某实例未在规定时间响应，则认为该实例主观下线。客观下线：若超过指定数量（quorum）的sentinel都认为该实例主观下线，则该实例客观下线。quorum值最好超过Sentinel实例数量的一半。3.1.3.集群故障恢复原理一旦发现master故障，sentinel需要在salve中选择一个作为新的master，选择依据是这样的：首先会判断slave节点与master节点断开时间长短，如果超过指定值（down-after-milliseconds * 10）则会排除该slave节点然后判断slave节点的slave-priority值，越小优先级越高，如果是0则永不参与选举如果slave-prority一样，则判断slave节点的offset值，越大说明数据越新，优先级越高最后是判断slave节点的运行id大小，越小优先级越高。当选出一个新的master后，该如何实现切换呢？流程如下：sentinel给备选的slave1节点发送slaveof no one命令，让该节点成为mastersentinel给所有其它slave发送slaveof 192.168.150.101 7002 命令，让这些slave成为新master的从节点，开始从新的master上同步数据。最后，sentinel将故障节点标记为slave，当故障节点恢复后会自动成为新的master的slave节点3.1.4.小结Sentinel的三个作用是什么？监控故障转移通知Sentinel如何判断一个redis实例是否健康？每隔1秒发送一次ping命令，如果超过一定时间没有相向则认为是主观下线如果大多数sentinel都认为实例主观下线，则判定服务下线故障转移步骤有哪些？首先选定一个slave作为新的master，执行slaveof no one然后让所有节点都执行slaveof 新master修改故障节点配置，添加slaveof 新master3.2.搭建哨兵集群具体搭建流程参考[post cid="6437" cover="" size=""/]3.3.RedisTemplate在Sentinel集群监管下的Redis主从集群，其节点会因为自动故障转移而发生变化，Redis的客户端必须感知这种变化，及时更新连接信息。Spring的RedisTemplate底层利用lettuce实现了节点的感知和自动切换。下面，我们通过一个测试来实现RedisTemplate集成哨兵机制。3.3.1.导入Demo工程首先，我们引入课前资料提供的Demo工程：3.3.2.引入依赖在项目的pom文件中引入依赖：<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency>3.3.3.配置Redis地址然后在配置文件application.yml中指定redis的sentinel相关信息：spring: redis: sentinel: master: mymaster nodes: - 192.168.150.101:27001 - 192.168.150.101:27002 - 192.168.150.101:270033.3.4.配置读写分离在项目的启动类中，添加一个新的bean：@Bean public LettuceClientConfigurationBuilderCustomizer clientConfigurationBuilderCustomizer(){ return clientConfigurationBuilder -> clientConfigurationBuilder.readFrom(ReadFrom.REPLICA_PREFERRED); }这个bean中配置的就是读写策略，包括四种：MASTER：从主节点读取MASTER_PREFERRED：优先从master节点读取，master不可用才读取replicaREPLICA：从slave（replica）节点读取REPLICA _PREFERRED：优先从slave（replica）节点读取，所有的slave都不可用才读取master4.Redis分片集群4.1.搭建分片集群主从和哨兵可以解决高可用、高并发读的问题。但是依然有两个问题没有解决：海量数据存储问题高并发写的问题使用分片集群可以解决上述问题，如图:分片集群特征：集群中有多个master，每个master保存不同数据每个master都可以有多个slave节点master之间通过ping监测彼此健康状态客户端请求可以访问集群任意节点，最终都会被转发到正确节点具体搭建流程参考[post cid="6437" cover="" size=""/]4.2.散列插槽4.2.1.插槽原理Redis会把每一个master节点映射到0~16383共16384个插槽（hash slot）上，查看集群信息时就能看到：数据key不是与节点绑定，而是与插槽绑定。redis会根据key的有效部分计算插槽值，分两种情况：key中包含"{}"，且“{}”中至少包含1个字符，“{}”中的部分是有效部分key中不包含“{}”，整个key都是有效部分例如：key是num，那么就根据num计算，如果是{itcast}num，则根据itcast计算。计算方式是利用CRC16算法得到一个hash值，然后对16384取余，得到的结果就是slot值。如图，在7001这个节点执行set a 1时，对a做hash运算，对16384取余，得到的结果是15495，因此要存储到103节点。到了7003后，执行get num时，对num做hash运算，对16384取余，得到的结果是2765，因此需要切换到7001节点4.2.1.小结Redis如何判断某个key应该在哪个实例？将16384个插槽分配到不同的实例根据key的有效部分计算哈希值，对16384取余余数作为插槽，寻找插槽所在实例即可如何将同一类数据固定的保存在同一个Redis实例？这一类数据使用相同的有效部分，例如key都以{typeId}为前缀4.3.集群伸缩redis-cli --cluster提供了很多操作集群的命令，可以通过下面方式查看：比如，添加节点的命令：4.3.1.需求分析需求：向集群中添加一个新的master节点，并向其中存储 num = 10启动一个新的redis实例，端口为7004添加7004到之前的集群，并作为一个master节点给7004节点分配插槽，使得num这个key可以存储到7004实例这里需要两个新的功能：添加一个节点到集群中将部分插槽分配到新插槽4.3.2.创建新的redis实例创建一个文件夹：mkdir 7004拷贝配置文件：cp redis.conf /7004修改配置文件：sed /s/6379/7004/g 7004/redis.conf启动redis-server 7004/redis.conf4.3.3.添加新节点到redis添加节点的语法如下：执行命令：redis-cli --cluster add-node 192.168.150.101:7004 192.168.150.101:7001通过命令查看集群状态：redis-cli -p 7001 cluster nodes如图，7004加入了集群，并且默认是一个master节点：但是，可以看到7004节点的插槽数量为0，因此没有任何数据可以存储到7004上4.3.4.转移插槽我们要将num存储到7004节点，因此需要先看看num的插槽是多少：如上图所示，num的插槽为2765.我们可以将0~3000的插槽从7001转移到7004，命令格式如下：具体命令如下：建立连接：得到下面的反馈：询问要移动多少个插槽，我们计划是3000个：新的问题来了：那个node来接收这些插槽？？显然是7004，那么7004节点的id是多少呢？复制这个id，然后拷贝到刚才的控制台后：这里询问，你的插槽是从哪里移动过来的？all：代表全部，也就是三个节点各转移一部分具体的id：目标节点的iddone：没有了这里我们要从7001获取，因此填写7001的id：填完后，点击done，这样插槽转移就准备好了：确认要转移吗？输入yes：然后，通过命令查看结果：可以看到：目的达成。4.4.故障转移集群初识状态是这样的：其中7001、7002、7003都是master，我们计划让7002宕机。4.4.1.自动故障转移当集群中有一个master宕机会发生什么呢？直接停止一个redis实例，例如7002：redis-cli -p 7002 shutdown1）首先是该实例与其它实例失去连接2）然后是疑似宕机：3）最后是确定下线，自动提升一个slave为新的master：4）当7002再次启动，就会变为一个slave节点了：4.4.2.手动故障转移利用cluster failover命令可以手动让集群中的某个master宕机，切换到执行cluster failover命令的这个slave节点，实现无感知的数据迁移。其流程如下：这种failover命令可以指定三种模式：缺省：默认的流程，如图1~6歩force：省略了对offset的一致性校验takeover：直接执行第5歩，忽略数据一致性、忽略master状态和其它master的意见案例需求：在7002这个slave节点执行手动故障转移，重新夺回master地位步骤如下：1）利用redis-cli连接7002这个节点2）执行cluster failover命令如图：效果：4.5.RedisTemplate访问分片集群RedisTemplate底层同样基于lettuce实现了分片集群的支持，而使用的步骤与哨兵模式基本一致：1）引入redis的starter依赖2）配置分片集群地址3）配置读写分离与哨兵模式相比，其中只有分片集群的配置方式略有差异，如下：spring: redis: cluster: nodes: - 192.168.150.101:7001 - 192.168.150.101:7002 - 192.168.150.101:7003 - 192.168.150.101:8001 - 192.168.150.101:8002 - 192.168.150.101:8003
节点信息先看一下当前路径信息[root@localhost sentinelRedis]# ll 总用量 8 drwxr-xr-x 2 root root 40 11月 9 17:48 7001 drwxr-xr-x 2 root root 40 11月 9 17:48 7002 drwxr-xr-x 2 root root 24 11月 3 13:21 7003 drwxrwxr-x 7 root root 4096 11月 3 11:44 redis-6.2.9 -rwxr-xr-x 1 root root 2656 11月 9 17:53 redis_sentinel.sh drwxr-xr-x 2 root root 79 11月 3 13:21 s1 drwxr-xr-x 2 root root 53 11月 3 13:21 s2 drwxr-xr-x 2 root root 53 11月 3 13:21 s3共包含三个节点，一个主节点，两个从节点。IPPORT角色172.100.xxx.xxx7001master172.100.xxx.xxx7002slave172.100.xxx.xxx7003slave分别有设置密码都是:123456在路径/usr/local/sentinelRedis中打开3个ssh窗口，分别启动3个redis实例，启动命令：# 第1个 redis-server 7001/redis.conf # 第2个 redis-server 7002/redis.conf # 第3个 redis-server 7003/redis.conf停止printf '%s\n' 7001 7002 7003 | xargs -I{} -t redis-cli -a 123456 -p {} shutdown三个sentinel实例信息如下：节点IPPORTs1172.100.xxx.xxx27001s2172.100.xxx.xxx27002s3172.100.xxx.xxx27003在路径/usr/local/sentinelRedis中打开3个ssh窗口，分别启动3个redis实例，启动命令：# 第1个 redis-sentinel s1/sentinel.conf # 第2个 redis-sentinel s2/sentinel.conf # 第3个 redis-sentinel s3/sentinel.conf快捷启动和停止脚本脚本位置：/usr/local/sentinelRedis/redis_sentinel.sh#!/bin/bash # 环境变量用于存储密码 export REDIS_PASSWORD="123456" export HOST="172.100.xxx.xxx" # 检查 Redis 和 Sentinel 实例的状态 check_status() { # 检查 Redis 实例状态 redis-cli -p 7001 ping > /dev/null 2>&1 if [ $? -eq 0 ]; then echo "Redis 实例 7001 正常运行。" else echo "Redis 实例 7001 未运行。" fi redis-cli -p 7002 ping > /dev/null 2>&1 if [ $? -eq 0 ]; then echo "Redis 实例 7002 正常运行。" else echo "Redis 实例 7002 未运行。" fi redis-cli -p 7003 ping > /dev/null 2>&1 if [ $? -eq 0 ]; then echo "Redis 实例 7003 正常运行。" else echo "Redis 实例 7003 未运行。" fi # 检查 Sentinel 实例状态 redis-cli -p 27001 ping > /dev/null 2>&1 if [ $? -eq 0 ]; then echo "Sentinel 实例 27001 正常运行。" else echo "Sentinel 实例 27001 未运行。" fi redis-cli -p 27002 ping > /dev/null 2>&1 if [ $? -eq 0 ]; then echo "Sentinel 实例 27002 正常运行。" else echo "Sentinel 实例 27002 未运行。" fi redis-cli -p 27003 ping > /dev/null 2>&1 if [ $? -eq 0 ]; then echo "Sentinel 实例 27003 正常运行。" else echo "Sentinel 实例 27003 未运行。" fi } # 启动 Redis 和 Sentinel 实例 start() { # 启动 Redis 实例 redis-server 7001/redis.conf > /dev/null 2>&1 & redis-server 7002/redis.conf > /dev/null 2>&1 & redis-server 7003/redis.conf > /dev/null 2>&1 & # 等待一段时间确保 Redis 实例已启动 sleep 5 # 启动 Sentinel 实例 redis-sentinel s1/sentinel.conf > /dev/null 2>&1 & redis-sentinel s2/sentinel.conf > /dev/null 2>&1 & redis-sentinel s3/sentinel.conf > /dev/null 2>&1 & } # 暂停 Redis 和 Sentinel 实例 stop() { # 停止 Redis 实例 printf '%s\n' 7001 7002 7003 | xargs -I{} -t sh -c 'redis-cli -h $HOST -p {} -a $REDIS_PASSWORD shutdown' # 停止 Sentinel 实例 printf '%s\n' 27001 27002 27003 | xargs -I{} -t sh -c 'redis-cli -p {} shutdown' } # 检查参数并执行相应操作 if [ "$1" = "start" ]; then start echo "Redis 和 Sentinel 实例已启动。" elif [ "$1" = "stop" ]; then stop echo "Redis 和 Sentinel 实例已暂停。" elif [ "$1" = "status" ]; then check_status else echo "用法: $0 {start|stop|status}" fi可以使用以下命令来启动和暂停 Redis 和 Sentinel 实例：启动：./redis_sentinel.sh start暂停：./redis_sentinel.sh stop查看状态：./redis_sentinel.sh status在运行脚本之前，赋予它执行权限：chmod +x redis_sentinel.sh使用效果[root@localhost sentinelRedis]# ll 总用量 8 drwxr-xr-x 2 root root 40 11月 9 17:48 7001 drwxr-xr-x 2 root root 40 11月 9 17:48 7002 drwxr-xr-x 2 root root 24 11月 3 13:21 7003 drwxrwxr-x 7 root root 4096 11月 3 11:44 redis-6.2.9 -rwxr-xr-x 1 root root 2656 11月 9 17:53 redis_sentinel.sh drwxr-xr-x 2 root root 79 11月 3 13:21 s1 drwxr-xr-x 2 root root 53 11月 3 13:21 s2 drwxr-xr-x 2 root root 53 11月 3 13:21 s3 [root@localhost sentinelRedis]# ./redis_sentinel.sh status Redis 实例 7001 未运行。 Redis 实例 7002 未运行。 Redis 实例 7003 未运行。 Sentinel 实例 27001 未运行。 Sentinel 实例 27002 未运行。 Sentinel 实例 27003 未运行。 [root@localhost sentinelRedis]# ./redis_sentinel.sh start Redis 和 Sentinel 实例已启动。 [root@localhost sentinelRedis]# ./redis_sentinel.sh status Redis 实例 7001 正常运行。 Redis 实例 7002 正常运行。 Redis 实例 7003 正常运行。 Sentinel 实例 27001 正常运行。 Sentinel 实例 27002 正常运行。 Sentinel 实例 27003 正常运行。 [root@localhost sentinelRedis]# ./redis_sentinel.sh stop sh -c redis-cli -h $HOST -p 7001 -a $REDIS_PASSWORD shutdown > /dev/null 2>&1 & sh -c redis-cli -h $HOST -p 7002 -a $REDIS_PASSWORD shutdown > /dev/null 2>&1 & sh -c redis-cli -h $HOST -p 7003 -a $REDIS_PASSWORD shutdown > /dev/null 2>&1 & sh -c redis-cli -p 27001 shutdown sh -c redis-cli -p 27002 shutdown sh -c redis-cli -p 27003 shutdown Redis 和 Sentinel 实例已暂停。 [root@localhost sentinelRedis]# ./redis_sentinel.sh status Redis 实例 7001 未运行。 Redis 实例 7002 未运行。 Redis 实例 7003 未运行。 Sentinel 实例 27001 未运行。 Sentinel 实例 27002 未运行。 Sentinel 实例 27003 未运行。 [root@localhost sentinelRedis]#
介绍：配置文件中可根据是否配置了哨兵模式，如果配置了则按照哨兵模式运行，如果没有配置哨兵模式，则按照单机模式运行。一、引入依赖<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> <exclusions> <exclusion> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> </exclusion> <exclusion> <groupId>io.lettuce</groupId> <artifactId>lettuce-core</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>${redis.version}</version> </dependency>二、配置文件spring: redis: host: 172.100.xxx.xxx #单机模式，哨兵模式可不配置 port: 6379 #单机模式，哨兵模式可不配置 password: xxxx timeout: 10000 database: 2 lettuce: pool: max-active: 180 #连接池最大连接数（使用负值表示没有限制） max-idle: 10 #连接池中的最大空闲连接 max-wait: 1500 #连接池最大阻塞等待时间（使用负值表示没有限制） # 哨兵模式，单机模式不要配置(配置文件中可根据是否配置了哨兵模式，如果配置了则按照哨兵模式运行，如果没有配置哨兵模式，则按照单机模式运行) sentinel: master: mymaster nodes: 172.100.xxx.xxx:27001,172.100.xxx.xxx:27002,172.100.xxx.xxx:27003三、配置类import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisNode; import org.springframework.data.redis.connection.RedisSentinelConfiguration; import org.springframework.data.redis.connection.RedisStandaloneConfiguration; import org.springframework.data.redis.connection.jedis.JedisClientConfiguration; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.StringRedisSerializer; import redis.clients.jedis.JedisPoolConfig; import java.time.Duration; import java.util.Arrays; import java.util.HashSet; import java.util.List; import java.util.Set; @Configuration @EnableAutoConfiguration public class RedisConfig { @Value("${spring.redis.host:}") private String host; @Value("${spring.redis.port:6379}") private int port; @Value("${spring.redis.timeout:10000}") private int timeout; @Value("${spring.redis.lettuce.pool.max-idle:10}") private int maxIdle; @Value("${spring.redis.lettuce.pool.max-wait:1500}") private long maxWaitMillis; @Value("${spring.redis.lettuce.pool.max-active:180}") private int maxActive; @Value("${spring.redis.password:}") private String password; @Value("${spring.redis.database:}") private String database; @Value("${spring.redis.sentinel.nodes:}") private String nodes; @Value("${spring.redis.sentinel.master:}") private String master; @Bean @ConfigurationProperties(prefix="spring.redis") public JedisPoolConfig getRedisConfig(){ JedisPoolConfig config = new JedisPoolConfig(); config.setMaxIdle(maxIdle); config.setMaxWaitMillis(maxWaitMillis); config.setMaxTotal(maxActive); return config; } // @Bean // public RedisSentinelConfiguration sentinelConfiguration(){ // RedisSentinelConfiguration redisSentinelConfiguration = new RedisSentinelConfiguration(); // //配置matser的名称 // redisSentinelConfiguration.master(master); // //配置redis的哨兵sentinel // Set<RedisNode> redisNodeSet = new HashSet<>(); // if (StringUtils.isNotBlank(master) && StringUtils.isNotBlank(nodes)) { // String[] nodeSplit = nodes.split(","); // List<String> nodeList = Arrays.asList(nodeSplit); // nodeList.forEach(x->{ // redisNodeSet.add(new RedisNode(x.split(":")[0],Integer.parseInt(x.split(":")[1]))); // }); // } // redisSentinelConfiguration.setSentinels(redisNodeSet); redisSentinelConfiguration.setSentinelPassword(password); // return redisSentinelConfiguration; // } @Bean public JedisConnectionFactory jedisConnectionFactory(JedisPoolConfig jedisPoolConfig) { JedisConnectionFactory jedisConnectionFactory = null; //判断redis是单机模式还是哨兵模式 if (StringUtils.isBlank(master) || StringUtils.isBlank(nodes)) { RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration(); redisStandaloneConfiguration.setHostName(host); redisStandaloneConfiguration.setPassword(password); redisStandaloneConfiguration.setDatabase(Integer.parseInt(database)); redisStandaloneConfiguration.setPort(port); JedisClientConfiguration.JedisClientConfigurationBuilder jedisClientConfiguration = JedisClientConfiguration.builder(); jedisClientConfiguration.connectTimeout(Duration.ofMillis(timeout)); JedisClientConfiguration jedisClientConfigurationNew = jedisClientConfiguration.usePooling().poolConfig(jedisPoolConfig).build(); jedisConnectionFactory = new JedisConnectionFactory(redisStandaloneConfiguration,jedisClientConfigurationNew); } else { RedisSentinelConfiguration redisSentinelConfiguration = new RedisSentinelConfiguration(); //配置matser的名称 redisSentinelConfiguration.master(master); //配置redis的哨兵sentinel Set<RedisNode> redisNodeSet = new HashSet<>(); if (StringUtils.isNotBlank(nodes)) { String[] nodeSplit = nodes.split(","); List<String> nodeList = Arrays.asList(nodeSplit); nodeList.forEach(x->{ redisNodeSet.add(new RedisNode(x.split(":")[0],Integer.parseInt(x.split(":")[1]))); }); } redisSentinelConfiguration.setSentinels(redisNodeSet); redisSentinelConfiguration.setPassword(password); jedisConnectionFactory = new JedisConnectionFactory(redisSentinelConfiguration,jedisPoolConfig); } return jedisConnectionFactory; } @Bean public RedisTemplate<String, Object> redisTemplate(JedisConnectionFactory jedisConnectionFactory) { RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>(); redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setHashKeySerializer(new StringRedisSerializer()); redisTemplate.setHashValueSerializer(new StringRedisSerializer()); redisTemplate.setValueSerializer(new StringRedisSerializer()); redisTemplate.setConnectionFactory(jedisConnectionFactory); return redisTemplate; } }四、RedisTemplate测试import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.test.context.junit4.SpringRunner; @SpringBootTest(classes = Application.class) @RunWith(SpringRunner.class) public class MyApplicationTest { @Autowired private RedisTemplate<String, Object> redisTemplate; @Test public void contextLoads() { // 测试内容 redisTemplate.opsForValue().set("test", "test"); } @Test public void contextGets() { // 测试内容 System.out.println(redisTemplate.opsForValue().get("test")); } }五、低版本配置我在项目部署中如果使用低版本Spring boot可能会出现一些配置类不存在等情况，比如我使用spring-boot:1.5.4.RELEASE，那么就会缺少import org.springframework.data.redis.connection.RedisStandaloneConfiguration; import org.springframework.data.redis.connection.jedis.JedisClientConfiguration;这两个配置，而且还会存在依赖冲突，版本不匹配，我们修改依赖如下：<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> <exclusions> <exclusion> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> </exclusion> <exclusion> <groupId>io.lettuce</groupId> <artifactId>lettuce-core</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>2.9.3</version> </dependency> <dependency> <groupId>org.springframework.data</groupId> <artifactId>spring-data-redis</artifactId> <version>1.7.11.RELEASE</version> </dependency>yamlspring: redis: host: 172.100.xxx.xxx #单机模式，哨兵模式可不配置 port: 6379 #单机模式，哨兵模式可不配置 password: xxxx timeout: 10000 database: 2 lettuce: pool: max-active: 180 #连接池最大连接数（使用负值表示没有限制） max-idle: 10 #连接池中的最大空闲连接 max-wait: 1500 #连接池最大阻塞等待时间（使用负值表示没有限制） # 哨兵模式，单机模式不要配置(配置文件中可根据是否配置了哨兵模式，如果配置了则按照哨兵模式运行，如果没有配置哨兵模式，则按照单机模式运行) sentinel: master: mymaster nodes: 172.100.xxx.xxx:27001,172.100.xxx.xxx:27002,172.100.xxx.xxx:27003 password: Topway12#$修改RedisConfig.java中的配置，我这里哨兵也配置了密码。package com.tw.nms.fttx.config; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisNode; import org.springframework.data.redis.connection.RedisSentinelConfiguration; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.StringRedisSerializer; import redis.clients.jedis.JedisPoolConfig; import java.util.Arrays; import java.util.HashSet; import java.util.List; import java.util.Set; @Configuration @EnableAutoConfiguration public class RedisSentinelAndStandaloneConfig { @Value("${spring.redis.host:}") private String host; @Value("${spring.redis.port:6379}") private int port; @Value("${spring.redis.timeout:10000}") private int timeout; @Value("${spring.redis.lettuce.pool.max-idle:10}") private int maxIdle; @Value("${spring.redis.lettuce.pool.max-wait:1500}") private long maxWaitMillis; @Value("${spring.redis.lettuce.pool.max-active:180}") private int maxActive; @Value("${spring.redis.password:}") private String password; @Value("${spring.redis.database:}") private String database; @Value("${spring.redis.sentinel.nodes:}") private String nodes; @Value("${spring.redis.sentinel.master:}") private String master; @Value("${spring.redis.sentinel.password:}") private String sentinelPassword; @Bean("JedisPoolConfig") @ConfigurationProperties(prefix = "spring.redis") public JedisPoolConfig getRedisConfig() { JedisPoolConfig config = new JedisPoolConfig(); config.setMaxIdle(maxIdle); config.setMaxWaitMillis(maxWaitMillis); config.setMaxTotal(maxActive); return config; } @Bean public JedisConnectionFactory jedisConnectionFactory(@Qualifier("JedisPoolConfig") JedisPoolConfig jedisPoolConfig) { JedisConnectionFactory jedisConnectionFactory = null; //判断redis是单机模式还是哨兵模式 if (StringUtils.isBlank(master) || StringUtils.isBlank(nodes)) { jedisConnectionFactory = new JedisConnectionFactory(); jedisConnectionFactory.setHostName(host); jedisConnectionFactory.setPort(port); jedisConnectionFactory.setTimeout(timeout); jedisConnectionFactory.setPassword(password); jedisConnectionFactory.setDatabase(Integer.parseInt(database)); jedisConnectionFactory.setPoolConfig(jedisPoolConfig); } else { //哨兵模式，设置密码 RedisSentinelConfiguration redisSentinelConfiguration = new RedisSentinelConfiguration(); redisSentinelConfiguration.setMaster(master); Set<RedisNode> redisNodeSet = new HashSet<>(); List<String> nodeList = Arrays.asList(nodes.split(",")); for (String node : nodeList) { String[] nodeArray = node.split(":"); redisNodeSet.add(new RedisNode(nodeArray[0], Integer.parseInt(nodeArray[1]))); } redisSentinelConfiguration.setSentinels(redisNodeSet); jedisConnectionFactory = new JedisConnectionFactory(redisSentinelConfiguration, jedisPoolConfig); jedisConnectionFactory.setPassword(sentinelPassword); } return jedisConnectionFactory; } @Bean public RedisTemplate<String, Object> redisTemplate(JedisConnectionFactory jedisConnectionFactory) { RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>(); redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setHashKeySerializer(new StringRedisSerializer()); redisTemplate.setHashValueSerializer(new StringRedisSerializer()); redisTemplate.setValueSerializer(new StringRedisSerializer()); redisTemplate.setConnectionFactory(jedisConnectionFactory); return redisTemplate; } }五、改进：哨兵节点不可用则切换为单机节点如果同时配置哨兵节点和单机节点，当哨兵节点不可用，则自适应选择单机节点package com.nms.tr069.config; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisNode; import org.springframework.data.redis.connection.RedisSentinelConfiguration; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.StringRedisSerializer; import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPoolConfig; import redis.clients.jedis.exceptions.JedisException; import java.util.Arrays; import java.util.HashSet; import java.util.List; import java.util.Set; @Slf4j @Configuration @EnableAutoConfiguration public class RedisSentinelAndStandaloneConfig { @Value("${spring.redis.host:}") private String host; @Value("${spring.redis.port:6379}") private int port; @Value("${spring.redis.timeout:10000}") private int timeout; @Value("${spring.redis.lettuce.pool.max-idle:10}") private int maxIdle; @Value("${spring.redis.lettuce.pool.max-wait:1500}") private long maxWaitMillis; @Value("${spring.redis.lettuce.pool.max-active:180}") private int maxActive; @Value("${spring.redis.password:}") private String password; @Value("${spring.redis.database:}") private String database; @Value("${spring.redis.sentinel.nodes:}") private String nodes; @Value("${spring.redis.sentinel.master:}") private String master; @Value("${spring.redis.sentinel.password:}") private String sentinelPassword; @Bean("JedisPoolConfig") @ConfigurationProperties(prefix = "spring.redis") public JedisPoolConfig getRedisConfig() { JedisPoolConfig config = new JedisPoolConfig(); config.setMaxIdle(maxIdle); config.setMaxWaitMillis(maxWaitMillis); config.setMaxTotal(maxActive); return config; } @Bean public JedisConnectionFactory jedisConnectionFactory(@Qualifier("JedisPoolConfig") JedisPoolConfig jedisPoolConfig) { JedisConnectionFactory jedisConnectionFactory = null; //判断redis是否配置了哨兵节点 if (StringUtils.isBlank(master) || StringUtils.isBlank(nodes)) { jedisConnectionFactory = new JedisConnectionFactory(); jedisConnectionFactory.setHostName(host); jedisConnectionFactory.setPort(port); jedisConnectionFactory.setTimeout(timeout); jedisConnectionFactory.setPassword(password); jedisConnectionFactory.setDatabase(Integer.parseInt(database)); jedisConnectionFactory.setPoolConfig(jedisPoolConfig); } else { //哨兵模式 RedisSentinelConfiguration redisSentinelConfiguration = new RedisSentinelConfiguration(); redisSentinelConfiguration.setMaster(master); Set<RedisNode> redisNodeSet = new HashSet<>(); List<String> nodeList = Arrays.asList(nodes.split(",")); for (String node : nodeList) { String[] nodeArray = node.split(":"); redisNodeSet.add(new RedisNode(nodeArray[0], Integer.parseInt(nodeArray[1]))); } redisSentinelConfiguration.setSentinels(redisNodeSet); // 尝试连接到所有哨兵节点，如果所有哨兵节点都不可用，则选择单机节点 boolean sentinelAvailable = false; for (RedisNode redisNode : redisNodeSet) { try { Jedis jedis = new Jedis(redisNode.getHost(), redisNode.getPort()); jedis.connect(); sentinelAvailable = true; jedis.disconnect(); break; } catch (JedisException e) { log.error("redis sentinel node {}:{} is not available", redisNode.getHost(), redisNode.getPort()); log.error(e.getMessage(), e); } } if (sentinelAvailable) { jedisConnectionFactory = new JedisConnectionFactory(redisSentinelConfiguration, jedisPoolConfig); jedisConnectionFactory.setPassword(sentinelPassword); } else { // 哨兵节点不可用，选择单机节点 jedisConnectionFactory = new JedisConnectionFactory(); jedisConnectionFactory.setHostName(host); jedisConnectionFactory.setPort(port); jedisConnectionFactory.setTimeout(timeout); jedisConnectionFactory.setPassword(password); jedisConnectionFactory.setDatabase(Integer.parseInt(database)); jedisConnectionFactory.setPoolConfig(jedisPoolConfig); } } return jedisConnectionFactory; } @Bean public RedisTemplate<String, Object> redisTemplate(JedisConnectionFactory jedisConnectionFactory) { RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>(); redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setHashKeySerializer(new StringRedisSerializer()); redisTemplate.setHashValueSerializer(new StringRedisSerializer()); redisTemplate.setValueSerializer(new StringRedisSerializer()); redisTemplate.setConnectionFactory(jedisConnectionFactory); return redisTemplate; } }
1、每种数据类型的应用场景String这个其实没啥好说的，最常规的set/get操作，value可以是String也可以是数字。一般做一些复杂的计数功能的缓存。hash这里value存放的是结构化的对象，比较方便的就是操作其中的某个字段。在做单点登录的时候，就是用这种数据结构存储用户信息，以cookieId作为key，设置30分钟为缓存过期时间，能很好的模拟出类似session的效果。list使用List的数据结构，可以做简单的消息队列的功能。另外还有一个就是，可以利用lrange命令，做基于redis的分页功能，性能极佳，用户体验好。set因为set堆放的是一堆不重复值的集合。所以可以做全局去重的功能。为什么不用JVM自带的Set进行去重？因为我们的系统一般都是集群部署，使用JVM自带的Set，比较麻烦，难道为了一个做一个全局去重，再起一个公共服务，太麻烦了。另外，就是利用交集、并集、差集等操作，可以计算共同喜好，全部的喜好，自己独有的喜好等功能。sorted setsorted set多了一个权重参数score,集合中的元素能够按score进行排列。可以做排行榜应用，取TOP N操作。2、过期策略及内存淘汰机制分析:如果你的redis只能存5G数据，可是你写了10G，那会删5G的数据。怎么删的？还有，你的数据已经设置了过期时间，但是时间到了，内存占用率还是比较高，怎么办?回答:redis采用的是定期删除+惰性删除策略。为什么不用定时删除策略?定时删除,用一个定时器来负责监视key,过期则自动删除。虽然内存及时释放，但是十分消耗CPU资源。在大并发请求下，CPU要将时间应用在处理请求，而不是删除key,因此没有采用这一策略.定期删除+惰性删除是如何工作的呢?定期删除，redis默认每个100ms检查，是否有过期的key,有过期key则删除。需要说明的是，redis不是每个100ms将所有的key检查一次，而是随机抽取进行检查(如果每隔100ms,全部key进行检查，redis岂不是卡死)。因此，如果只采用定期删除策略，会导致很多key到时间没有删除。于是，惰性删除派上用场。也就是说在你获取某个key的时候，redis会检查一下，这个key如果设置了过期时间那么是否过期了？如果过期了此时就会删除。采用定期删除+惰性删除就没其他问题了么?不管是定期删除还是惰性删除，都是一种不完全精确的删除策略，始终还是会存在已经过期的key无法被删除的场景。而且这两种过期策略都是只针对设置了过期时间的key，不适用于没有设置过期时间的key的淘汰，所以，Redis还提供了内存淘汰策略，用来筛选淘汰指定的key。3、Redis 的存储原理3.1 存储结构Redis 的存储方式使用的是散列表的形式，我们都知道，数据结构的最底层物理结构只有数组和链表，而其他的比如说队列、树、图等数据结构都是基于这两个物理结构所实现的逻辑数据结构。而数组和链表的区别主要在于读写的时间复杂度和内存的拓容上，因为数组要拓容的成本很高，因此大多数情况下会选用链表来实现各种逻辑数据结构。在 Redis 的底层存储中，会维护一个全局的哈希表（本质上就是数组），它有无数个哈希桶（数组元素）组成，每一个哈希桶的存储了两个指针，一个是指向 key 的指针，另一个则是指向 value 的指针。目前 Redis 的 key 只有一种形式，就是字符串，而 value 却有多种支撑，包括 字符串类型，列表，哈希，集合，有序集合。我们知道数组取下标元素的时间复杂度为 O(1), 因为想要获得获取对应的哈希桶，实际上消耗的时间是哈希函数的计算，这个计算和哈希桶的数据量是没有关系的，一般来说哈希函数对应的时间复杂度为 O(1)。这种设计十分的巧妙，把资源放在其他地方，只保留对应资源的指针，在维护上会有很大的便捷性，这就好比说，我不需要把货物都放在房子里，我只需要一张纸，记录着我的每个货物的位置即可。这相当于MySQL 的表结构和数据分开存储有异曲同工之处。3.2 rehash学过数据结构的同学都知道，没有一个绝对无敌的哈希函数可以实现无冲突处理，而一般来说解决冲突的方式有两大类（注意这里是分的大类），一个是开放地址法，另一个是链地址法。而 Redis 采用的正是第二种方式。当存在冲突的时候，会将对应的冲突实体放在一起搞成一个链表，此时哈希桶的指针会增加一个next 指针，用来指向下一个冲突实体，当数据量足够大的时候，这个链表可能会被拉的越来越长，因为链表的查询只能从头结点开始遍历下去，也就是 O(n) 的复杂度(n为链表长度),所以当链表过长时，会影响最终的查询速度。所以这时候不能袖手旁观。因此，在这种情况下，Redis 会有另外一个线程专门去处理。我们称这种方式为 rehash ，和很多语言的数组底层一样，这种rehash 机制也是基于复制状态机的，说白了就是申请另一块空间（通常是原来的两倍），然后把数据信息复制过去，再释放原来的空间。通常来说程序不会快死的时候才想着救自己，当数值达到预警值时，就会开始自救，我们常常将其称之为装填因子，计算方式十分的简单，就是 use/total, 不同的存储对装填因子的判断规则不同，内容也比较底层，这里我们就不展开叙述了。所以在 Redis 中，会存在另一张全局哈希表，其实它就是一个备胎，每次需要拓容的时候，这个备胎往往空间要比原配更大，rehash 线程会将原配的数据复制到备胎中，然后备胎就可以转正了。但是在复制的过程中，会有两个问题，一个是内存占用率会突然飙升，另一个就是Redis 阻塞的问题，复制的操作又耗时又耗空间，因此我们还需要更加聪明一点，能不能让一次的操作分成多步呢？温水煮青蛙，如果每次来一个请求我就迁徙一点，这样的话，是不是慢慢的我就复制完了。这就是 Redis 的渐进式 rehash。3.3 渐进式 rehash两个人的感情是需要慢慢的培养的，程序的处理也是可以慢慢来的，有些事情并不非要一次性搞完的，我们首先需要知道，Redis 是怎么去渐进式 rehash的，假设我要取 key 为 csdn 的 value ，而通过hash 算法得到的 索引位置为 1，但是该索引上有一个三个 entry, 此时处理的线程正常的去遍历这个链表拿到真正正确的值，此时 rehash进程 顺便把这个索引的 entry 从 ht0 复制到 ht1 中。并且释放 ht0 该索引的空间。大致流程如下图所示：这时候我们就会纠结一个问题了，如果有两种全局哈希表，那么Redis 的 crud 是怎么去操作的呢？Redis的删除、查找、更新等操作会在两个全局哈希表上进行，比如说， 要在字典里面查找一个键的话， 程序会先在 ht0 里面进行查找， 如果没找到的话， 就会继续到 ht1 里面进行查找，并且新添加到Redis的键值对一律会被保存到 ht[1] 里面， 而 ht[0] 则不再进行任何添加操作： 这一措施保证了 ht[0] 包含的键值对数量会只减不增， 并随着 rehash 操作的执行而最终变成空表，就完成了备胎转正流程。4、缓存穿透、缓存击穿、缓存雪崩在介绍这三大问题之前，我们需要先了解Redis作为一个缓存中间件，在项目中是如何工作的。首先看一下在没有缓存中间件的时候的系统数据访问的架构图：客户端发起一个查询请求的时候，首先去缓存中查询，如果数据在缓存中存在，则直接将缓存中的数据返回给客户端；如果数据在缓存中不存在，则继续查询数据库，如果数据在数据库中存在，则将该数据放入缓存中，并返回给客户端，如果数据在数据库中也不存在，则直接返回null给客户端。4.1、缓存穿透缓存穿透是指查询一个缓存中和数据库中都不存在的数据，导致每次查询这条数据都会透过缓存，直接查库，最后返回空。当用户使用这条不存在的数据疯狂发起查询请求的时候，对数据库造成的压力就非常大，甚至可能直接挂掉。这种情况的流程就变成下图这样了：解决缓存穿透的方法一般有两种，第一种是缓存空对象，第二种是使用布隆过滤器。第一种方法比较好理解，就是当数据库中查不到数据的时候，我缓存一个空对象，然后给这个空对象的缓存设置一个过期时间，这样下次再查询该数据的时候，就可以直接从缓存中拿到，从而达到了减小数据库压力的目的。但这种解决方式有两个缺点：（1）需要缓存层提供更多的内存空间来缓存这些空对象，当这种空对象很多的时候，就会浪费更多的内存；（2）会导致缓存层和存储层的数据不一致，即使在缓存空对象时给它设置了一个很短的过期时间，那也会导致这一段时间内的数据不一致问题。第二种方案是使用布隆过滤器，这是比较推荐的方法。所谓布隆过滤器，就是一种数据结构，它是由一个长度为m bit的位数组与n个hash函数组成的数据结构，位数组中每个元素的初始值都是0。在初始化布隆过滤器时，会先将所有key进行n次hash运算，这样就可以得到n个位置，然后将这n个位置上的元素改为1。这样，就相当于把所有的key保存到了布隆过滤器中了。举个例子，比如我们一共有3个key，我们对这3个key分别进行3次hash运算，key1经过三次hash运算后的结果分别为2/6/10，那么就把布隆过滤器中下标为2/6/10的元素值更新为1，然后再分别对key2和key3做同样操作，结果如下图：这样，当客户端查询时，也对查询的key做3次hash运算得到3个位置，然后看布隆过滤器中对应位置元素的值是否为1，如果所有对应位置元素的值都为1，就证明key在库中存在，则继续向下查询；如果3个位置中有任意一个位置的值不为1，那么就证明key在库中不存在，直接返回客户端空即可。如下图：当客户端查询key4时，key4的3次hash运算中，有一个位置8的值为0，就说明key4在库中不存在，直接返回客户端空即可。所以，布隆过滤器就相当于一个位于客户端与缓存层中间的拦截器一样，负责判断key是否在集合中存在。如下图：布隆过滤器的好处就是解决了第一种缓存空值的不足，但布隆过滤器也存在缺陷，首先，它有误判的可能，比如在上面客户端查询key4的图中，假如key4经过3次hash运算得到的位置分别是2/4/6，由于这3个位置的值都是1，所以，布隆过滤器就认为key4在库中存在，进而继续向下查询了。所以，布隆过滤器判断存在的key实际上可能是不存在的，但布隆过滤器判断不存在的key是一定不存在的。它的第二个缺点就是删除元素比较难，比如现在要删除key2这个元素，那么需要将2/7/11三个位置的元素值改为0，但这样就会影响到key1和key3的判断。4.2、缓存击穿缓存击穿是指当缓存中某个热点数据过期了，在该热点数据重新载入缓存之前，有大量的查询请求穿过缓存，直接查询数据库。这种情况会导致数据库压力瞬间骤增，造成大量请求阻塞，甚至直接挂掉。缓存击穿解决方案解决缓存击穿的方法也有两种；第一种是设置key永不过期；在设置热点key的时候，不给key设置过期时间即可。第二方式也可以达到key不过期的目的，就是正常给key设置过期时间，不过在后台同时启一个定时任务去定时地更新这个缓存。第二种方式使用了加锁的方式，锁的对象就是key，这样，当大量查询同一个key的请求并发进来时，只能有一个请求获取到锁，然后获取到锁的线程查询数据库，然后将结果放入到缓存中，然后释放锁，此时，其他处于锁等待的请求即可继续执行，由于此时缓存中已经有了数据，所以直接从缓存中获取到数据返回，并不会查询数据库。4.3、缓存雪崩缓存雪崩是指当缓存中有大量的key在同一时刻过期，或者Redis直接宕机了，导致大量的查询请求全部到达数据库，造成数据库查询压力骤增，甚至直接挂掉。缓存雪崩解决方案1、将失效时间分散开通过使用自动生成随机数使得key的过期时间是随机的，防止集体过期2、使用多级架构使用nginx缓存+redis缓存+其他缓存，不同层使用不同的缓存，可靠性更强3、设置缓存标记记录缓存数据是否过期，如果过期会触发通知另外的线程在后台去更新实际的key4、使用锁或者队列的方式如果查不到就加上排它锁，其他请求只能进行等待9、springboot整合redis依赖：<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> <version>2.3.12.RELEASE</version> </dependency>public String redisTest(String username){ ValueOperations valueOperations = redisTemplate.opsForValue(); valueOperations.set("user",username); valueOperations.set("userInfo","lutiancheng",1000, TimeUnit.MICROSECONDS); return valueOperations.get("user").toString(); }
实践周老师教ssm，想要在其中整合redis做缓存，但是之前都是使用的springBoot，所以在这里记录一下整合的过程1. 首先在pom.xml中添加相关的依赖： <dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>2.10.1</version> </dependency> <dependency> <groupId>org.springframework.data</groupId> <artifactId>spring-data-redis</artifactId> <version>2.1.8.RELEASE</version> </dependency>2. 然后在resources目录下创建redis.properties文件#redis.host=192.168.2.200 redis.host=127.0.0.1 redis.port=6379 redis.password=123456 redis.maxIdle=300 redis.minIdle=10 redis.maxWaitMillis=1000 redis.maxTotal=500 redis.testOnBorrow=true redis.testOnReturn=true redis.testWhileIdle=true redis.blockWhenExhausted=false redis.numTestsPerEvictionRun=1024 redis.timeBetweenEvictionRunsMillis=30000 redis.minEvictableIdleTimeMillis=1800000上述各种配置除了前四个为必需之外，其他的参数可以根据需要设置。3. application-redis.xml<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> <context:property-placeholder location="classpath:redis.properties" ignore-unresolvable="true"/> <bean id="poolConfig" class="redis.clients.jedis.JedisPoolConfig"> <property name="maxIdle" value="${redis.maxIdle}"></property> <property name="minIdle" value="${redis.minIdle}"></property> <property name="maxTotal" value="${redis.maxTotal}"></property> <property name="maxWaitMillis" value="${redis.maxWaitMillis}"></property> <property name="testOnBorrow" value="${redis.testOnBorrow}"></property> </bean> <bean id="connectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory" p:hostName="${redis.host}" p:port="${redis.port}" p:password="${redis.password}" p:pool-config-ref="poolConfig" p:timeout="100000"/> <bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate" p:connection-factory-ref="connectionFactory" > <property name="keySerializer"> <bean class="org.springframework.data.redis.serializer.StringRedisSerializer" /> </property> <property name="valueSerializer"> <bean class="org.springframework.data.redis.serializer.StringRedisSerializer" /> </property> <property name="hashKeySerializer"> <bean class="org.springframework.data.redis.serializer.StringRedisSerializer" /> </property> </bean> </beans>在主配置文件applicationContext.xml中引入redis的配置文件，即：<import resource="classpath:application-redis.xml"/>插入到以下位置： <beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:util="http://www.springframework.org/schema/util" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.0.xsd http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.0.xsd"> <import resource="classpath:application-redis.xml"/> </beans>4. 测试@RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration("classpath:applicationContext.xml") public class redisTest { @Autowired private RedisTemplate redisTemplate; @Test public void test() { redisTemplate.opsForValue().set("name", "zhangsan"); Object name = redisTemplate.opsForValue().get("name"); System.out.println(name); } }5. 特别的如果想在配置类中加上xml配置文件我们加入这个注解:@ImportResource("classpath:applicationContext.xml")比如在我这里就是：@Configuration @ComponentScan({"com.tedustore.service"}) @PropertySource("classpath:mysql.properties") @Import({JdbcConfig.class, MyBatisConfig.class}) @ImportResource("classpath:applicationContext.xml") @EnableTransactionManagement //开启Spring事务管理 public class SpringConfig { }测试类就不是导入xml配置了，直接导入SpringConfig配置类即可。@RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(classes = SpringConfig.class) public class redisTest { @Autowired private RedisTemplate redisTemplate; @Test public void test() { redisTemplate.opsForValue().set("name3", "zhangsan"); Object name = redisTemplate.opsForValue().get("name"); System.out.println(name); } }
1、拉取redis镜像Redis的版本选择与安装 在Redis的版本计划中，版本号第二位为奇数，为非稳定版本，如2.7、2.9、3.1；版本号第二为偶数，为稳定版本如2.6、2.8、3.0；一般来说当前奇数版本是下一个稳定版本的开发版本，如2.9是3.0的开发版本。建议去https://hub.docker.com/找到当前Redis版本的最新稳定版本。docker pull redis2、在本地某个位置创建以下内容建议将以下内容放在一起，方便以后管理和查看# 以/docker/redis为例 mkdir -p /docker/redis mkdir -p /docker/redis/data touch /docker/redis/redis.conf touch /docker/redis/redis.bash3、编辑配置文件 vim /docker/redis/redis.conf# Redis配置文件 # Redis默认不是以守护进程的方式运行，可以通过该配置项修改，使用yes启用守护进程 daemonize no # 指定Redis监听端口，默认端口为6379 port 6379 # 绑定的主机地址，不要绑定容器的本地127.0.0.1地址，因为这样就无法在容器外部访问 bind 0.0.0.0 #需要密码则打开 requirepass 123123 # 持久化 appendonly yes4、编辑 vim /docker/redis/redis.bashdocker run -p 6379:6379 --name redis -v /docker/redis/redis.conf:/etc/redis/redis.conf -v /docker/redis/data:/data -d redis redis-server /etc/redis/redis.conf 解释以上命令docker run redis # 从redis镜像运行容器 -p 6379:6379 # 映射本地6379端口到容器6379端口，前为本地端口 --name redis # 设置容器名称为redis，方便以后使用docker ps进行管理 -v /docker/redis/redis.conf:/etc/redis/redis.conf # 关联本地/docker/redis/redis.conf文件到容器中/etc/redis/redis.conf，同样，前为本地 -v /docker/redis/data:/data # 关联本地/docker/redis/data到容器内/data目录，此为存放redis数据的目录，为方便以后升级redis，而数据可以留存 -d # 后台启动，使用此方式启动，则redis.conf中daemonize必须设置为no，否则会无法启动 redis # 指定要运行的镜像为redis镜像 redis-server /etc/redis/redis.conf # 传递给Redis镜像的命令，在容器内部执行该命令以启动Redis服务器并使用指定的配置文件5、给予执行权限sudo chmod 777 /docker/redis/redis.bash6、启动 /docker/redis/redis.bash # 查看是否已启动 docker ps # 如果无法启动或者docker ps中无对应内容，将bash中命令复制出来，删除-d参数启动，查看报错信息 # 使用redis-cli或者rdm访问 localhost:6379 # 如需访问容器，可使用 docker exec -it redis bash # 或直接使用redis-cli访问容器内redis docker exec -it redis redis-cli [-a 123123]注意：#出现以下错误则是没有输入密码导致 127.0.0.1:6379> set name "hello" (error) NOAUTH Authentication required. #运行以下命令即可 127.0.0.1:6379> auth "123123"7、下一次启动方法docker restart redis
课程内容Redis 入门Redis 数据类型Redis 常用命令在 Java 中操作 Redis1. 前言1.1 什么是 RedisRedis 是一个基于内存的 key-value 结构数据库。Redis 是互联网技术领域使用最为广泛的存储中间件，它是「Remote Dictionary Service」的首字母缩写，也就是「远程字典服务」。基于内存存储，读写性能高适合存储热点数据（热点商品、资讯、新闻）企业应用广泛1.2 使用 Redis 能做什么数据缓存消息队列注册中心发布订阅2. Redis 入门2.1 Redis 简介Redis is an open source (BSD licensed), in-memory data structure store, used as a database, cache, and message broker. 翻译为：Redis 是一个开源的内存中的数据结构存储系统，它可以用作：数据库、缓存和消息中间件。官网：https://redis.ioRedis 是用 C 语言开发的一个开源的高性能键值对(key-value)数据库，官方提供的数据是可以达到 100000+的 QPS（每秒内查询次数）。它存储的 value 类型比较丰富，也被称为结构化的 NoSql 数据库。NoSql（Not Only SQL），不仅仅是 SQL，泛指非关系型数据库。NoSql 数据库并不是要取代关系型数据库，而是关系型数据库的补充。关系型数据库(RDBMS)：MysqlOracleDB2SQLServer非关系型数据库(NoSql)：RedisMongo dbMemCached2.2 Redis 下载与安装2.2.1 Redis 下载Redis 安装包分为 windows 版和 Linux 版：Windows 版下载地址：https://github.com/microsoftarchive/redis/releasesLinux 版下载地址： https://download.redis.io/releases/下载后得到下面安装包：2.2.2 Redis 安装1）在 Linux 中安装 Redis在 Linux 系统安装 Redis 步骤：将 Redis 安装包上传到 Linux解压安装包，命令：tar -zxvf redis-4.0.0.tar.gz -C /usr/local安装 Redis 的依赖环境 gcc，命令：yum install gcc-c++进入/usr/local/redis-4.0.0，进行编译，命令：make进入 redis 的 src 目录进行安装，命令：make install安装后重点文件说明：/usr/local/redis-4.0.0/src/redis-server：Redis 服务启动脚本/usr/local/redis-4.0.0/src/redis-cli：Redis 客户端脚本/usr/local/redis-4.0.0/redis.conf：Redis 配置文件2）在 Windows 中安装 RedisRedis 的 Windows 版属于绿色软件，直接解压即可使用，解压后目录结构如下：2.3 Redis 服务启动与停止1）Linux 系统中启动和停止 Redis执行 Redis 服务启动脚本文件redis-server：通过启动日志可以看到，Redis 默认端口号为6379。Ctrl + C停止 Redis 服务通过redis-cli可以连接到本地的 Redis 服务，默认情况下不需要认证即可连接成功。退出客户端可以输入exit或者quit命令。2）Windows 系统中启动和停止 RedisWindows 系统中启动 Redis，直接双击 redis-server.exe 即可启动 Redis 服务，redis 服务默认端口号为 6379Ctrl + C停止 Redis 服务双击redis-cli.exe即可启动 Redis 客户端，默认连接的是本地的 Redis 服务，而且不需要认证即可连接成功。退出客户端可以输入exit或者quit命令。2.4 Redis 配置文件前面我们已经启动了 Redis 服务，默认情况下 Redis 启动后是在前台运行，而且客户端不需要密码就可以连接到 Redis 服务。如果我们希望 Redis 服务启动后是在后台运行，同时希望客户端认证通过后才能连接到 Redis 服务，应该如果做呢？此时就需要修改 Redis 的配置文件：Linux 系统中 Redis 配置文件：REDIS_HOME/redis.confWindows 系统中 Redis 配置文件：REDIS_HOME/redis.windows.conf通过修改 Redis 配置文件可以进行如下配置：1）设置 Redis 服务后台运行将配置文件中的daemonize配置项改为 yes，默认值为 no。注意：Windows 版的 Redis 不支持后台运行。2）设置 Redis 服务密码将配置文件中的 # requirepass foobared 配置项取消注释，默认为注释状态。foobared 为密码，可以根据情况自己指定。3）设置允许客户端远程连接 Redis 服务Redis 服务默认只能客户端本地连接，不允许客户端远程连接。将配置文件中的 bind 127.0.0.1 配置项注释掉。解释说明：Redis 配置文件中 # 表示注释Redis 配置文件中的配置项前面不能有空格，需要顶格写daemonize：用来指定 redis 是否要用守护线程的方式启动，设置成 yes 时，代表开启守护进程模式。在该模式下，redis 会在后台运行requirepass：设置 Redis 的连接密码bind：如果指定了 bind，则说明只允许来自指定网卡的 Redis 请求。如果没有指定，就说明可以接受来自任意一个网卡的 Redis 请求。注意：修改配置文件后需要重启 Redis 服务配置才能生效，并且启动 Redis 服务时需要显示的指定配置文件：1）Linux 中启动 Redis 服务# 进入Redis安装目录 cd /usr/local/redis-4.0.0 # 启动Redis服务，指定使用的配置文件 ./src/redis-server ./redis.conf2）Windows 中启动 Redis 服务由于 Redis 配置文件中开启了认证校验，即客户端连接时需要提供密码，此时客户端连接方式变为：解释说明：-h：指定连接的 Redis 服务的 ip 地址-p：指定连接的 Redis 服务的端口号-a：指定连接的 Redis 服务的密码2.5 Redis外网访问如果是redis 3.2以上的版本默认是不允许外网访问的，需要修改redis.conf配置文件修改peotected-mode修改peotected-mode yes 改为：protected-mode no. #protected-mode参数是为了禁止外网访问redis,如果启用了，则只能够通过localhost ip (127.0.0.1)访问Redis关闭 bind 127.0.0.1注释掉bind 127.0.0.1, 或者修改bind 0.0.0.0,表示允许所有ip地址访问修改完配置文件记得重启redis./redis-server /usr/local/redis/redis.conf3. Redis 数据类型3.1 介绍Redis 存储的是 key-value 结构的数据，其中 key 是字符串类型，value 有 5 种常用的数据类型：字符串 string哈希 hash列表 list集合 set有序集合 sorted set / zset3.2 Redis 5 种常用数据类型解释说明：字符串(string)：普通字符串，常用哈希(hash)：适合存储对象列表(list)：按照插入顺序排序，可以有重复元素集合(set)：无序集合，没有重复元素有序集合(sorted set / zset)：集合中每个元素关联一个分数（score），根据分数升序排序，没有重复元素4. Redis 常用命令4.1 字符串 string 操作命令Redis 中字符串类型常用命令：SET key value 设置指定 key 的值GET key 获取指定 key 的值SETEX key seconds value 设置指定 key 的值，并将 key 的过期时间设为 seconds 秒SETNX key value 只有在 key 不存在时设置 key 的值更多命令可以参考 Redis 中文网：https://www.redis.net.cn4.2 哈希 hash 操作命令Redis hash 是一个 string 类型的 field 和 value 的映射表，hash 特别适合用于存储对象，常用命令：HSET key field value 将哈希表 key 中的字段 field 的值设为 valueHGET key field 获取存储在哈希表中指定字段的值HDEL key field 删除存储在哈希表中的指定字段HKEYS key 获取哈希表中所有字段HVALS key 获取哈希表中所有值HGETALL key 获取在哈希表中指定 key 的所有字段和值4.3 列表 list 操作命令Redis 列表是简单的字符串列表，按照插入顺序排序，常用命令：LPUSH key value1 [value2] 将一个或多个值插入到列表头部LRANGE key start stop 获取列表指定范围内的元素RPOP key 移除并获取列表最后一个元素LLEN key 获取列表长度BRPOP key1 [key2 ] timeout 移出并获取列表的最后一个元素， 如果列表没有元素会阻塞列表直到等待超 时或发现可弹出元素为止4.4 集合 set 操作命令Redis set 是 string 类型的无序集合。集合成员是唯一的，这就意味着集合中不能出现重复的数据，常用命令：SADD key member1 [member2] 向集合添加一个或多个成员SMEMBERS key 返回集合中的所有成员SCARD key 获取集合的成员数SINTER key1 [key2] 返回给定所有集合的交集SUNION key1 [key2] 返回所有给定集合的并集SDIFF key1 [key2] 返回给定所有集合的差集SREM key member1 [member2] 移除集合中一个或多个成员4.5 有序集合 sorted set 操作命令Redis sorted set 有序集合是 string 类型元素的集合，且不允许重复的成员。每个元素都会关联一个 double 类型的分数(score) 。redis 正是通过分数来为集合中的成员进行从小到大排序。有序集合的成员是唯一的，但分数却可以重复。常用命令：ZADD key score1 member1 [score2 member2] 向有序集合添加一个或多个成员，或者更新已存在成员的 分数ZRANGE key start stop [WITHSCORES] 通过索引区间返回有序集合中指定区间内的成员ZINCRBY key increment member 有序集合中对指定成员的分数加上增量 incrementZREM key member [member ...] 移除有序集合中的一个或多个成员4.6 通用命令Redis 中的通用命令，主要是针对 key 进行操作的相关命令：KEYS pattern 查找所有符合给定模式( pattern)的 keyEXISTS key 检查给定 key 是否存在TYPE key 返回 key 所储存的值的类型TTL key 返回给定 key 的剩余生存时间(TTL, time to live)，以秒为单位DEL key 该命令用于在 key 存在是删除 key5. 在 Java 中操作 Redis5.1 介绍前面我们讲解了 Redis 的常用命令，这些命令是我们操作 Redis 的基础，那么我们在 java 程序中应该如何操作 Redis 呢？这就需要使用 Redis 的 Java 客户端，就如同我们使用 JDBC 操作 MySQL 数据库一样。Redis 的 Java 客户端很多，官方推荐的有三种：JedisLettuceRedissonSpring 对 Redis 客户端进行了整合，提供了 Spring Data Redis，在 Spring Boot 项目中还提供了对应的 Starter，即 spring-boot-starter-data-redis。5.2 JedisJedis 是 Redis 的 Java 版本的客户端实现。maven 坐标：<dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>2.8.0</version> </dependency>使用 Jedis 操作 Redis 的步骤：获取连接执行操作关闭连接示例代码：package com.itheima.test; import org.junit.Test; import redis.clients.jedis.Jedis; import java.util.Set; /** * 使用Jedis操作Redis */ public class JedisTest { @Test public void testRedis(){ //1 获取连接 Jedis jedis = new Jedis("localhost",6379); //2 执行具体的操作 jedis.set("username","xiaoming"); String value = jedis.get("username"); System.out.println(value); //jedis.del("username"); jedis.hset("myhash","addr","bj"); String hValue = jedis.hget("myhash", "addr"); System.out.println(hValue); Set<String> keys = jedis.keys("*"); for (String key : keys) { System.out.println(key); } //3 关闭连接 jedis.close(); } }5.3 Spring Data Redis5.3.1 介绍Spring Data Redis 是 Spring 的一部分，提供了在 Spring 应用中通过简单的配置就可以访问 Redis 服务，对 Redis 底层开发包进行了高度封装。在 Spring 项目中，可以使用 Spring Data Redis 来简化 Redis 操作。网址：https://spring.io/projects/spring-data-redismaven 坐标：<dependency> <groupId>org.springframework.data</groupId> <artifactId>spring-data-redis</artifactId> <version>2.4.8</version> </dependency>Spring Boot 提供了对应的 Starter，maven 坐标：<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency>Spring Data Redis 中提供了一个高度封装的类：RedisTemplate，针对 Jedis 客户端中大量 api 进行了归类封装,将同一类型操作封装为 operation 接口，具体分类如下：ValueOperations：简单 K-V 操作SetOperations：set 类型数据操作ZSetOperations：zset 类型数据操作HashOperations：针对 hash 类型的数据操作ListOperations：针对 list 类型的数据操作5.3.2 使用方式5.3.2.1 环境搭建第一步：创建 maven 项目 springdataredis_demo，配置 pom.xml 文件<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.4.5</version> <relativePath/> </parent> <groupId>com.itheima</groupId> <artifactId>springdataredis_demo</artifactId> <version>1.0-SNAPSHOT</version> <properties> <java.version>1.8</java.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> <version>2.4.5</version> </plugin> </plugins> </build> </project>第二步：编写启动类package com.itheima; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class App { public static void main(String[] args) { SpringApplication.run(App.class,args); } }第三步：配置 application.ymlspring: application: name: springdataredis_demo #Redis相关配置 redis: host: localhost port: 6379 #password: 123456 database: 0 #操作的是0号数据库 jedis: #Redis连接池配置 pool: max-active: 8 #最大连接数 max-wait: 1ms #连接池最大阻塞等待时间 max-idle: 4 #连接池中的最大空闲连接 min-idle: 0 #连接池中的最小空闲连接解释说明：spring.redis.database：指定使用 Redis 的哪个数据库，Redis 服务启动后默认有 16 个数据库，编号分别是从 0 到 15。可以通过修改 Redis 配置文件来指定数据库的数量。第四步：提供配置类package com.itheima.config; import org.springframework.cache.annotation.CachingConfigurerSupport; 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.StringRedisSerializer; /** * Redis配置类 */ @Configuration public class RedisConfig extends CachingConfigurerSupport { @Bean public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory connectionFactory) { RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>(); //默认的Key序列化器为：JdkSerializationRedisSerializer redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setHashKeySerializer(new StringRedisSerializer()); redisTemplate.setConnectionFactory(connectionFactory); return redisTemplate; } }解释说明：当前配置类不是必须的，因为 Spring Boot 框架会自动装配 RedisTemplate 对象，但是默认的 key 序列化器为 JdkSerializationRedisSerializer，导致我们存到 Redis 中后的数据和原始数据有差别第五步：提供测试类package com.itheima.test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; @SpringBootTest @RunWith(SpringRunner.class) public class SpringDataRedisTest { @Autowired private RedisTemplate redisTemplate; }5.3.2.2 操作字符串类型数据/** * 操作String类型数据 */ @Test public void testString(){ //存值 redisTemplate.opsForValue().set("city123","beijing"); //取值 String value = (String) redisTemplate.opsForValue().get("city123"); System.out.println(value); //存值，同时设置过期时间 redisTemplate.opsForValue().set("key1","value1",10l, TimeUnit.SECONDS); //存值，如果存在则不执行任何操作 Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent("city1234", "nanjing"); System.out.println(aBoolean); }5.3.2.3 操作哈希类型数据/** * 操作Hash类型数据 */ @Test public void testHash(){ HashOperations hashOperations = redisTemplate.opsForHash(); //存值 hashOperations.put("002","name","xiaoming"); hashOperations.put("002","age","20"); hashOperations.put("002","address","bj"); //取值 String age = (String) hashOperations.get("002", "age"); System.out.println(age); //获得hash结构中的所有字段 Set keys = hashOperations.keys("002"); for (Object key : keys) { System.out.println(key); } //获得hash结构中的所有值 List values = hashOperations.values("002"); for (Object value : values) { System.out.println(value); } }5.3.2.4 操作列表类型数据/** * 操作List类型的数据 */ @Test public void testList(){ ListOperations listOperations = redisTemplate.opsForList(); //存值 listOperations.leftPush("mylist","a"); listOperations.leftPushAll("mylist","b","c","d"); //取值 List<String> mylist = listOperations.range("mylist", 0, -1); for (String value : mylist) { System.out.println(value); } //获得列表长度 llen Long size = listOperations.size("mylist"); int lSize = size.intValue(); for (int i = 0; i < lSize; i++) { //出队列 String element = (String) listOperations.rightPop("mylist"); System.out.println(element); } }5.3.2.5 操作集合类型数据/** * 操作Set类型的数据 */ @Test public void testSet(){ SetOperations setOperations = redisTemplate.opsForSet(); //存值 setOperations.add("myset","a","b","c","a"); //取值 Set<String> myset = setOperations.members("myset"); for (String o : myset) { System.out.println(o); } //删除成员 setOperations.remove("myset","a","b"); //取值 myset = setOperations.members("myset"); for (String o : myset) { System.out.println(o); } }5.3.2.6 操作有序集合类型数据/** * 操作ZSet类型的数据 */ @Test public void testZset(){ ZSetOperations zSetOperations = redisTemplate.opsForZSet(); //存值 zSetOperations.add("myZset","a",10.0); zSetOperations.add("myZset","b",11.0); zSetOperations.add("myZset","c",12.0); zSetOperations.add("myZset","a",13.0); //取值 Set<String> myZset = zSetOperations.range("myZset", 0, -1); for (String s : myZset) { System.out.println(s); } //修改分数 zSetOperations.incrementScore("myZset","b",20.0); //取值 myZset = zSetOperations.range("myZset", 0, -1); for (String s : myZset) { System.out.println(s); } //删除成员 zSetOperations.remove("myZset","a","b"); //取值 myZset = zSetOperations.range("myZset", 0, -1); for (String s : myZset) { System.out.println(s); } }5.3.2.7 通用操作/** * 通用操作，针对不同的数据类型都可以操作 */ @Test public void testCommon(){ //获取Redis中所有的key Set<String> keys = redisTemplate.keys("*"); for (String key : keys) { System.out.println(key); } //判断某个key是否存在 Boolean itcast = redisTemplate.hasKey("itcast"); System.out.println(itcast); //删除指定key redisTemplate.delete("myZset"); //获取指定key对应的value的数据类型 DataType dataType = redisTemplate.type("myset"); System.out.println(dataType.name()); }