RedisTemplate封装操作Redis缓存的基本API,大部分Redis操作都是通过RedisTemplate完成的。
POM文件依赖
<dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <!--通用spring boot mapper--> <dependency> <groupId>tk.mybatis</groupId> <artifactId>mapper-spring-boot-starter</artifactId> <version>2.0.3</version> </dependency> <!--mysql驱动--> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> </dependency> <!--swagger--> <dependency> <groupId>io.springfox</groupId> <artifactId>springfox-swagger2</artifactId> <version>2.9.2</version> </dependency> <!--swagger-ui--> <dependency> <groupId>io.springfox</groupId> <artifactId>springfox-swagger-ui</artifactId> <version>2.9.2</version> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-redis</artifactId> <version>1.4.7.RELEASE</version> </dependency> </dependencies>
配置文件
mybatis.mapper-locations=classpath*:com/agan/redis/mapper/xml/*.xml spring.datasource.driverClassName=com.mysql.jdbc.Driver spring.datasource.url=jdbc:mysql://127.0.0.1:3306/boot_user?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull spring.datasource.username=root spring.datasource.password=agan logging.level.com.agan=debug spring.swagger2.enabled=true spring.redis.database=0 spring.redis.host=127.0.0.1 spring.redis.port=6379 spring.redis.password=
Controller层:
@Api(description = "用户接口") @RestController @RequestMapping("/user") public class UserController { @Autowired private UserService userService; @ApiOperation("数据库初始化100条数据") @RequestMapping(value = "/init", method = RequestMethod.GET) public void init() { for (int i = 0; i < 100; i++) { Random rand = new Random(); User user = new User(); String temp = "un" + i; user.setUsername(temp); user.setPassword(temp); int n = rand.nextInt(2); user.setSex((byte) n); userService.createUser(user); } } @ApiOperation("单个用户查询,按userid查用户信息") @RequestMapping(value = "/findById/{id}", method = RequestMethod.GET) public UserVO findById(@PathVariable int id) { User user = this.userService.findUserById(id); UserVO userVO = new UserVO(); BeanUtils.copyProperties(user, userVO); return userVO; } @ApiOperation("修改某条数据") @PostMapping(value = "/updateUser") public void updateUser(@RequestBody UserVO obj) { User user = new User(); BeanUtils.copyProperties(obj, user); userService.updateUser(user); } }
Service层
@Service public class UserService { private static final Logger LOGGER = LoggerFactory.getLogger(UserService.class); public static final String CACHE_KEY_USER = "user:"; @Autowired private UserMapper userMapper; @Autowired private RedisTemplate redisTemplate; public void createUser(User obj){ this.userMapper.insertSelective(obj); //缓存key String key=CACHE_KEY_USER+obj.getId(); //到数据库里面,重新捞出新数据出来,做缓存 obj=this.userMapper.selectByPrimaryKey(obj.getId()); //opsForValue代表了Redis的String数据结构 //set代表了redis的SET命令 redisTemplate.opsForValue().set(key,obj); } public void updateUser(User obj){ //1.先直接修改数据库 this.userMapper.updateByPrimaryKeySelective(obj); //2.再修改缓存 //缓存key String key=CACHE_KEY_USER+obj.getId(); obj=this.userMapper.selectByPrimaryKey(obj.getId()); //修改也是用SET命令,重新设置,Redis 没有update操作,都是重新设置新值 redisTemplate.opsForValue().set(key,obj); } public User findUserById(Integer userid){ ValueOperations<String, User> operations = redisTemplate.opsForValue(); //缓存key String key=CACHE_KEY_USER+userid; //1.先去redis查 ,如果查到直接返回,没有的话直接去数据库捞 //Redis 用了GET命令 User user=operations.get(key); //2.redis没有的话,直接去数据库捞 if(user==null){ user=this.userMapper.selectByPrimaryKey(userid); //由于redis没有才到数据库捞,所以必须把捞到的数据写入redis,方便下次查询能redis命中。 operations.set(key,user); } return user; } }
步骤体验效果:
用http://127.0.0.1:9090/swagger-ui.html# 体验
问题1:进redis的数据必须序列化Serializable
问题2:如果连接不了redis
vi redis.conf bind 0.0.0.0
默认情况下,Redis序列化使用的JDK序列化方式JdkSerializationRedisSerializer,这就会导致产生两个问题:
@Table(name = "users") public class User implements Serializable {...}
127.0.0.1:6379> keys * 1) "\xac\xed\x00\x05t\x00\auser:62" 2) "\xac\xed\x00\x05t\x00\auser:65" 3) "\xac\xed\x00\x05t\x00\auser:50" 4) "\xac\xed\x00\x05t\x00\auser:36" 5) "\xac\xed\x00\x05t\x00\x06user:6" 6) "\xac\xed\x00\x05t\x00\auser:17" 7) "\xac\xed\x00\x05t\x00\auser:28" 127.0.0.1:6379> get "\xac\xed\x00\x05t\x00\auser:62" "\xac\xed\x00\x05sr\x00\x1acom.agan.redis.entity.User?\xebU\xa1\xe2\xa6\xfe\xe3\x02\x00\aL\x00\ncreateTimet \x00\x10Ljava/util/Date;L\x00\adeletedt\x00\x10Ljava/lang/Byte;L\x00\x02idt\x00\x13Ljava/lang/Integer;L\x00 \bpasswordt\x00\x12Ljava/lang/String;L\x00\x03sexq\x00~\x00\x02L\x00\nupdateTimeq\x00~\x00\x01L\x00\buser nameq\x00~\x00\x04xpsr\x00\x0ejava.util.Datehj\x81\x01KYt\x19\x03\x00\x00xpw\b\x00\x00\x01o+5\x1d\xf8xsr \x00\x0ejava.lang.Byte\x9cN`\x84\xeeP\xf5\x1c\x02\x00\x01B\x00\x05valuexr\x00\x10java.lang.Number\x86\xac \x95\x1d\x0b\x94\xe0\x8b\x02\x00\x00xp\x00sr\x00\x11java.lang.Integer\x12\xe2\xa0\xa4\xf7\x81\x878\x02\x00 \x01I\x00\x05valuexq\x00~\x00\t\x00\x00\x00>t\x00\x04un59q\x00~\x00\nsq\x00~\x00\x06w\b\x00\x00\x01o+5\x1d \xf8xt\x00\x04un59"
获取的值都是乱码。
@Configuration public class RedisConfiguration { /** * 重写Redis序列化方式,使用Json方式: * 当我们的数据存储到Redis的时候,我们的键(key)和值(value)都是通过Spring提供的Serializer序列化到Redis的。 * RedisTemplate默认使用的是JdkSerializationRedisSerializer, * StringRedisTemplate默认使用的是StringRedisSerializer。 * * Spring Data JPA为我们提供了下面的Serializer: * GenericToStringSerializer、Jackson2JsonRedisSerializer、 * JacksonJsonRedisSerializer、JdkSerializationRedisSerializer、 * OxmSerializer、StringRedisSerializer。 * 在此我们将自己配置RedisTemplate并定义Serializer。 */ @Bean public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) { RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>(); redisTemplate.setConnectionFactory(redisConnectionFactory); //创建一个json的序列化对象 GenericJackson2JsonRedisSerializer jackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer(); //设置value的序列化方式json redisTemplate.setValueSerializer(jackson2JsonRedisSerializer); //设置key序列化方式string redisTemplate.setKeySerializer(new StringRedisSerializer()); //设置hash key序列化方式string redisTemplate.setHashKeySerializer(new StringRedisSerializer()); //设置hash value的序列化方式json redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer); redisTemplate.afterPropertiesSet(); return redisTemplate; } }
测试:
1. 先把user的序列化删除 2. 创建类RedisConfiguration 3. flushdb 清空redis的旧数据,因为改了序列化,老数据以及不能兼容了,必须清空旧数据 4. 往redis 初始化100条数据 5. 用 keys * 命令查看所有key
127.0.0.1:6379> keys *
127.0.0.1:6379> get user:187
“{”@class":“com.agan.redis.entity.User”,“id”:187,“username”:“un84”,“password”:“un84”,
“sex”:0,“deleted”:0,“updateTime”:[“java.util.Date”,1576983528000],
“createTime”:[“java.util.Date”,1576983528000]}"
### 总结 1. 对于Redis的存储对象信息,其实就是 redisTemplate.opsForValue().set(key,value)就可以解决 2. 对于Redis,DB操作顺序问题,一般都是先操作DB,再操作Redis,尽可能避免产生脏数据。 3. 如果先更新Redis,再更新DB,如果更新DB失败,那么Redis数据就是脏数据。 4. 由于Redis使用了JDK序列化方式,对象需要实现序列化接口,Redis存储的值有乱码问题,可读性差,所以需要设置Redis key,value的序列化方式。 # SpringCache - SpringCache 他是对使用缓存进行封装和抽象,通过在方法上使用annotation注解就能拿到缓存结果; - 用了Annotation解决了业务代码和缓存代码的耦合度问题,即在不侵入业务代码的基础上让现有代码支持缓存; - 开发人员无感知使用了缓存 - 特别注意:(注意:对于redis的缓存,springcache只支持String,其他的Hash 、List、set、ZSet都不支持,要特别注意) ## 代码 POM依赖 ```java <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <!--spring cache--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-cache</artifactId> </dependency> <!--spring cache连接池依赖包--> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-pool2</artifactId> <version>2.6.2</version> </dependency>
配置文件
## Redis 配置 # Redis数据库索引(默认为0) spring.redis.database=0 # Redis服务器地址 spring.redis.host=127.0.0.1 # Redis服务器连接端口 spring.redis.port=6379 # Redis服务器连接密码(默认为空) spring.redis.password= # 连接池最大连接数(使用负值表示没有限制) spring.redis.lettuce.pool.max-active=8 # 连接池最大阻塞等待时间 spring.redis.lettuce.pool.max-wait=-1ms # 连接池中的最大空闲连接 spring.redis.lettuce.pool.max-idle=8 # 连接池中的最小空闲连接 spring.redis.lettuce.pool.min-idle=0 # 连接超时时间(毫秒) spring.redis.timeout=5000ms
开启缓存配置,设置序列化
@Configuration @EnableCaching public class RedisConfig { @Primary @Bean public CacheManager cacheManager(RedisConnectionFactory redisConnectionFactory){ RedisCacheConfiguration redisCacheConfiguration = RedisCacheConfiguration.defaultCacheConfig(); redisCacheConfiguration = redisCacheConfiguration //设置缓存的默认超时时间:30分钟 .entryTtl(Duration.ofMinutes(30L)) //如果是空值,不缓存 .disableCachingNullValues() //设置key序列化器 .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(keySerializer())) //设置value序列化器 .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(valueSerializer())); return RedisCacheManager .builder(RedisCacheWriter.nonLockingRedisCacheWriter(redisConnectionFactory)) .cacheDefaults(redisCacheConfiguration) .build(); } /** * key序列化器 */ private RedisSerializer<String> keySerializer() { return new StringRedisSerializer(); } /** * value序列化器 */ private RedisSerializer<Object> valueSerializer() { return new GenericJackson2JsonRedisSerializer(); } }
逻辑代码:
@Api(description = "用户接口") @RestController @RequestMapping("/user") public class UserController { @Autowired private UserService userService; @ApiOperation("单个用户查询,按userid查用户信息") @RequestMapping(value = "/findById/{id}", method = RequestMethod.GET) public UserVO findById(@PathVariable int id) { User user = this.userService.findUserById(id); UserVO userVO = new UserVO(); BeanUtils.copyProperties(user, userVO); return userVO; } @ApiOperation("修改某条数据") @PostMapping(value = "/updateUser") public void updateUser(@RequestBody UserVO obj) { User user = new User(); BeanUtils.copyProperties(obj, user); userService.updateUser(user); } @ApiOperation("按id删除用户") @RequestMapping(value = "/del/{id}", method = RequestMethod.GET) public void deleteUser(@PathVariable int id) { this.userService.deleteUser(id); } }
Service
@Service @CacheConfig(cacheNames = { "user" }) public class UserService { private static final Logger LOGGER = LoggerFactory.getLogger(UserService.class); @Autowired private UserMapper userMapper; @Cacheable(key="#id") public User findUserById(Integer id){ return this.userMapper.selectByPrimaryKey(id); } @CachePut(key = "#obj.id") public User updateUser(User obj){ this.userMapper.updateByPrimaryKeySelective(obj); return this.userMapper.selectByPrimaryKey(obj.getId()); } @CacheEvict(key = "#id") public void deleteUser(Integer id){ User user=new User(); user.setId(id); user.setDeleted((byte)1); this.userMapper.updateByPrimaryKeySelective(user); } }
@Cacheable(key="#id") public User findUserById(Integer id){ return this.userMapper.selectByPrimaryKey(id); }
@CachePut(key = "#obj.id") public User updateUser(User obj){ this.userMapper.updateByPrimaryKeySelective(obj); return this.userMapper.selectByPrimaryKey(obj.getId()); }
@CacheEvict(key = "#id") public void deleteUser(Integer id){ User user=new User(); user.setId(id); user.setDeleted((byte)1); this.userMapper.updateByPrimaryKeySelective(user); }
像日常操作中,热点新闻阅读量、贴吧帖子阅读量、文章阅读量,只要用户查看了这些东西,其阅读量对应+1,大的并发量,一般不可能采用数据库来做计数器,通常都是用redis的incr命令来实现。
用途就是计数器,如果key不存在,那就将key的value值初始化为0,如果存在,则自动加1;
127.0.0.1:6379> incr article:100 (integer) 1 127.0.0.1:6379> incr article:100 (integer) 2 127.0.0.1:6379> incr article:100 (integer) 3 127.0.0.1:6379> incr article:100 (integer) 4 127.0.0.1:6379> get article:100 "4"
技术方案的缺陷:
需要频繁的修改redis,耗费CPU,高并发修改redis会导致 redisCPU 100%
@RestController @Slf4j public class ViewController { @Autowired private StringRedisTemplate stringRedisTemplate; @GetMapping(value = "/view") public void view(Integer id) { //redis key String key="article:"+id; //调用redis的increment计数器命令 long n=this.stringRedisTemplate.opsForValue().increment(key); log.info("key={},阅读量为{}",key, n); } }
大型分布式系统架构中,全局唯一ID生成器的机器需要实现高可用高QPS,不然整个系统就挂了;
技术思路:
代码:
ID生成器代码类:
@Service public class IdGenerator { @Autowired private StringRedisTemplate stringRedisTemplate; private static final String ID_KEY = "id:generator:product"; /** * 生成全局唯一id */ public Long incrementId() { long n=this.stringRedisTemplate.opsForValue().increment(ID_KEY); return n; } }
controller
@RestController @Slf4j @RequestMapping(value = "/pruduct") public class ProductController { @Autowired private IdGenerator idGenerator; @PostMapping(value = "/create") public void create(Product obj) { //步骤1:生成分布式id long id=this.idGenerator.incrementId(); //全局id,代替数据库的自增id obj.setId(id); //步骤2:取模,计算表名 //类似于海量的数据,例如淘宝一般是分为1024张表,这里为了演示方便,只分为8张表。 int table=(int)id % 8; String tablename="product_"+table; log.info("插入表名{},插入内容{}",tablename,obj); } }
Lua 是一个简洁、轻量、可扩展的脚本语言,它的特性有:
EVAL script numkeys key [key ...] arg [arg ...]
127.0.0.1:6379> EVAL "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 2 key1 key2 agan1 agna2 1) "key1" 2) "key2" 3) "agan1" 4) "agna2"
@GetMapping(value = "/updateuser") public void updateUser(Integer uid,String uname) { String key="user:"+uid; //优化点:第一次发送redis请求 String old=this.stringRedisTemplate.opsForValue().get(key); if(StringUtils.isEmpty(old)){ //优化点:第二次发送redis请求 this.stringRedisTemplate.opsForValue().set(key,uname); return; } if(old.equals(uname)){ log.info("{}不用修改", key); }else{ log.info("{}从{}修改为{}", key,old,uname); //优化点:第二次发送redis请求 this.stringRedisTemplate.opsForValue().set(key,uname); } }
以上代码,看似简单,但是在高并发的情况下,还是有一点性能瓶颈,在性能方面主要是发送了2次redis请求。 那如何优化呢?我们可以采用lua技术,把2次redis请求合成一次。
编写lua文件,并存储于resources/lua/compareAndSet.lua里;
-- 成功设置返回1 没设置返回0 -- 如果redis没找到,就直接写进去 if redis.call('get', KEYS[1]) == nil then redis.call('set', KEYS[1], ARGV[1]); return 1 end -- 如果旧值不等于新值,就把新值设置进去 if redis.call('get', KEYS[1]) ~= ARGV[1] then redis.call('set', KEYS[1], ARGV[1]); return 1 else return 0 end
创建lua脚本对象
@Configuration public class LuaConfiguration { @Bean public DefaultRedisScript<Long> compareAndSetScript() { DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(); redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/compareAndSet.lua"))); redisScript.setResultType(Long.class); return redisScript; } }
SpringBoot执行lua脚本
@GetMapping(value = "/updateuserlua") public void updateUserLua(Integer uid,String uname) { String key="user:"+uid; //设置redis的key List<String> keys = Arrays.asList(key); //执行lua脚本,execute方法有3个参数,第一个参数是lua脚本对象,第二个是key列表,第三个是lua的参数数组 Long n = this.stringRedisTemplate.execute(this.compareAndSetScript, keys, uname); if (n == 0) { log.info("{}不用修改", key); } else { log.info("{}修改为{}", key,uname); } }
网站黑客攻击通常就是通过并发死循环来请求接口,通常会请求两类接口;
针对某个接口,采用访问频率控制,当某个ip在短时间内频繁访问接口时,需要记录并识别出来,这种高并发请求,通常都是采用redis+lua来实现。
-- 为某个接口的请求ip设置计数器,例如 当ip 127.0.0.1请求商品接口时,key=product:127.0.0.1 local times = redis.call('incr',KEYS[1]) -- 当某个ip第一次请求时,为该ip的key设置超时时间。 if times == 1 then redis.call('expire',KEYS[1], ARGV[1]) end -- tonumber就是把某个字符串转换为数字 -- 例如 某个ip 30秒内,请求次数大于10,就返回0,反则 返回1 if times > tonumber(ARGV[2]) then return 0 end return 1
Redis-cli执行:
[root@node2 src]# ./redis-cli --eval limit.lua producapi:127.0.0.1 , 30 10 (integer) 0
创建lua脚本对象
@Bean public DefaultRedisScript<Long> limitScript() { DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(); redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/limit.lua"))); //设置返回值类型 redisScript.setResultType(Long.class); return redisScript; }
SpringBoot执行lua脚本
@GetMapping(value = "/productlist") public String productList(HttpServletRequest request) { //获取请求ip String ip = IpUtils.getIpAddr(request); //设置redis 的key List<String> keys = Arrays.asList("pruductAPI:" + ip); //执行lua脚本,execute方法有3个参数,第一个参数是lua脚本对象,第二个是key列表,第三个是lua的参数数组 //30代表30秒 ,10代表超过10次,也就是说同个ip 30秒内不能超过10次请求 Long n = this.stringRedisTemplate.execute(this.limitScript, keys, "30", "10"); String result=""; //非法请求 if (n == 0) { result= "非法请求"; } else { result= "返回商品列表"; } log.info("ip={}请求结果:{}", ip,result); return result; }
打开settings->File Encoding->勾选Transparent native-to-ascii conversion