使用Java（Springboot）操作Redis

你的名字 2022-05-31 04:26 317阅读 0赞

1、 redis简介   
redis是Nosql数据库中使用较为广泛的非关系型内存数据库，redis内部是一个key-value存储系统。它支持存储的value类型相对更多，包括string(字符串)、list(链表)、set(集合)、zset(sorted set –有序集合)和hash（哈希类型，类似于Java中的map）。Redis基于内存运行并支持持久化的NoSQL数据库，是当前最热门的NoSql数据库之一，也被人们称为数据结构服务器。   
2、 互联网时代背景下大机遇，什么要使用Nosql？   
1） 当数据量的总大小一个机器放不下时。   
2） 数据索引一个机器的内存放不下时。   
3） 访问量（读写混合）一个实例放不下时。

单机时代模型

![SouthEast][]

如果每次存储成千上万条数据，这样很会导致mysql的性能很差，存储以及读取速度很慢，然后就演变成缓存+mysql+垂直拆分的方式。   
![SouthEast 1][]

Cache作为中间缓存   
将所有的数据先保存到缓存中，然后再存入mysql中，减小数据库压力，提高效率。   
但是当数据再次增加到又一个量级，上面的方式也不能满足需求，由于数据库的写入压力增加，Memcached只能缓解数据库的读取压力。读写集中在一个数据库上让数据库不堪重负，大部分网站开始使用主从复制技术来达到读写分离，以提高读写性能和读库的可扩展性。Mysql的master-slave模式成为这个时候的网站标配了。

![SouthEast 2][]

主从分离模式   
在redis的高速缓存，MySQL的主从复制，读写分离的基础之上，这时MySQL主库的写压力开始出现瓶颈，而数据量的持续猛增，由于MyISAM使用表锁，在高并发下会出现严重的锁问题，大量的高并发MySQL应用开始使用InnoDB引擎代替MyISAM。

![SouthEast 3][]

分表分库模式   
将变化小的、业务相关的放在一个数据库，变化多的，不相关的数据放在一个数据库。   
3、 nosql数据库的优势   
1）易扩展   
这些类型的数据存储不需要固定的模式，无需多余的操作就可以进行横向的扩展。相对于关系型数据库可以减少表和字段特别多的情况。也无型之间在架构的层面上带来了可扩展的能力   
2）大数据量提高性能   
3）多样灵活的数据模型   
在nosql中不仅可以存储String，hash，set、Zset等数据类型，还可以保存javaBean以及多种复杂的数据类型。   
4、 NoSql的应用   
1） 大数据时代淘宝、微信、以及微博等都广泛的使用了redis数据库，将一些固定不变的数据例如学校，区域等固定的信息保存在关系型数据库中。然后对于经常变化的数据例如淘宝每个节日都会有比较热门的搜索显示在搜索框，当节日过去关键字自动删除，为了便于管理，可以将这些数据保存在redis数据库中，并设置过期时间，到达时间就自动删除。   
2）为了缓解数据库压力，微博首先将发送的微博保存到redis数据库，自己可以立即查看到，然后将内存中的数据同步到关系型数据库。

下面通过例子，介绍利用Java操作Redis的例子。在springboot中操作Redis可以通过Jedis或者通过RedisTemplate来实现，因为springboot会自动扫描生成RedisTemplate对象，并且使用RedisTemplate操作Redis单节点或者Redis集群都可以使用同样的接口，这里使用RedisTemplate来讲解。

1. 首先访问https://start.spring.io，生成springboot项目，依赖选择web和redis，选择Generate project。

![SouthEast 4][]

2. 我们这里整合了**swagger-ui**，可以方便你测试后台restful api接口。在pom文件中添加如下依赖

<dependency>
                <groupId>io.springfox</groupId>
                <artifactId>springfox-swagger2</artifactId>
                <version>${swagger.version}</version>
            </dependency>
            <dependency>
                <groupId>io.springfox</groupId>
                <artifactId>springfox-swagger-ui</artifactId>
                <version>${swagger.version}</version>
            </dependency>

###### 修改application.properties文件 ######

######  ######

#for single node
    spring.redis.host=127.0.0.1
    spring.redis.port=6379
    spring.redis.database=0
    spring.redis.pool.min-idle=5
    spring.redis.pool.max-idle=10
    
    #for redis cluster
    #spring.redis.cluster.nodes[0]=172.17.0.61:6379
    #spring.redis.cluster.nodes[1]=172.17.0.61:6380
    #spring.redis.cluster.nodes[2]=172.17.0.61:6381
    #spring.redis.cluster.nodes[3]=172.17.0.61:6382
    #spring.redis.cluster.nodes[4]=172.17.0.61:6383
    #spring.redis.cluster.nodes[5]=172.17.0.61:6384
    #spring.redis.cluster.nodes[6]=172.17.0.61:6385
    #spring.redis.cluster.nodes[7]=172.17.0.61:6386
    #spring.redis.cluster.nodes[8]=172.17.0.61:6387
    #spring.redis.cluster.nodes[9]=172.17.0.61:6388
    #spring.redis.cluster.nodes[10]=172.17.0.61:6389
    #spring.redis.cluster.nodes[11]=172.17.0.61:6390
    #spring.redis.cluster.nodes[12]=172.17.0.61:6391
    #spring.redis.cluster.nodes[13]=172.17.0.61:6392
    #spring.redis.cluster.nodes[14]=172.17.0.61:6393

3.  RedisTemplate介绍

spring 封装了 RedisTemplate 对象来进行对redis的各种操作，它支持所有的 redis 原生的 api。

RedisTemplate中定义了5中数据结构的操作。

redisTemplate.opsForValue();//操作字符串
    redisTemplate.opsForHash();//操作hash
    redisTemplate.opsForList();//操作list
    redisTemplate.opsForSet();//操作set
    redisTemplate.opsForZSet();//操作有序set

spring-data-redis的序列化类有下面这几个:

*  GenericToStringSerializer: 可以将任何对象泛化为字符串并序列化
 *  Jackson2JsonRedisSerializer: 跟JacksonJsonRedisSerializer实际上是一样的
 *  JacksonJsonRedisSerializer: 序列化object对象为json字符串
 *  JdkSerializationRedisSerializer: 序列化java对象
 *  StringRedisSerializer: 简单的字符串序列化

具体可以参考这篇文章， [关于Spring Data redis几种对象序列化的比较][Spring Data redis]。

我们这里选择Key使用StringRedisSerialize，Value使用Jackson2JsonRedisSerializer的方式。避免Key被序列化后无法通过Redis客户端访问的问题。

@Component
    public class RedisUtils {
     @Autowired
     @Qualifier("redisTemplate")
     RedisTemplate template;
    
    
     @Autowired
     ApplicationContextProvider provider;
    
    
     private Logger log = LoggerFactory.getLogger(this.getClass());
    
    
     @Bean
     RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
     MessageListenerAdapter listenerAdapter) {
    
    
     RedisMessageListenerContainer container = new RedisMessageListenerContainer();
     container.setConnectionFactory(connectionFactory);
     // container.addMessageListener(listenerAdapter, new PatternTopic("chat"));
    
    
     return container;
     }
    
    
     @Bean
     MessageListenerAdapter listenerAdapter(Receiver receiver) {
     return new MessageListenerAdapter(receiver, "receiveMessage");
     }
    
    
     @Bean
     Receiver receiver(CountDownLatch latch) {
     return new Receiver(latch);
     }
    
    
     @Bean
     CountDownLatch latch() {
     return new CountDownLatch(1);
     }
    
    
     @Bean
     StringRedisTemplate stringTemplate(RedisConnectionFactory connectionFactory) {
     return new StringRedisTemplate(connectionFactory);
     }
    
    
     @PostConstruct
     public void init() {
     RedisSerializer<String> stringSerializer = new StringRedisSerializer();
     Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(
     Object.class);
     ObjectMapper om = new ObjectMapper();
     om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
     om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
     jackson2JsonRedisSerializer.setObjectMapper(om);
     template.setKeySerializer(stringSerializer);
     template.setValueSerializer(jackson2JsonRedisSerializer);
     template.setHashKeySerializer(stringSerializer);
     template.setHashValueSerializer(jackson2JsonRedisSerializer);
     template.afterPropertiesSet();
     }
    }

4. 为了避免每个service都去生成RedisTemplate，我包装了一个RedisUtils对象，在对象里面实现了所有Redis相关操作。

4.1 设置key/value值，并且可以设置超时时间。

public void setValue(String key, Object val) {
    		template.opsForValue().set(key, val);
    	}
    
    	public void setValue(String key, Object val, int time, TimeUnit unit) {
    		template.opsForValue().set(key, val, time, unit);
    	}

4.2 获取key/value值

public Object getValue(String key) {
    		return template.opsForValue().get(key);
    	}

4.3 同时设置多个key/value

public void multiSet(Map<String, Object> map) {
    		template.opsForValue().multiSet(map);
    	}

4.4 同时获取多个key/value

public List<Object> multiGet(Collection<String> keys) {
    		return template.opsForValue().multiGet(keys);
    	}

4.5 操作取号器

public long incr(String key, long delta) {
    		return template.opsForValue().increment(key, delta);
    	}

4.6 操作队列，在高并发的情况下，如果同时上千个线程同时操作数据库，数据库很可能会因此而宕机。这个时候我们可以利用redis队列，进行排队依次写入数据库。利用lpush和rpop，可以形成一个queue。

public void lpush(String key, String value) {
    		template.opsForList().leftPush(key, value);
    	}
    
    	public List<Object> range(String key, int start, int end) {
    		return template.opsForList().range(key, start, end);
    	}
    
    	public Object rpop(String key) {
    		return template.opsForList().rightPop(key);
    	}

4.7 有时候我们不光希望保存key/value，也希望缓存某个对象，比如用户所有数据到redis。我们可以使用Redis的Hash数据结构来缓存。

public void setHash(String key, Map<String, Object> map) {
    		template.opsForHash().putAll(key, map);
    	}
    
    	public Object getHash(String key, String prop) {
    		return template.opsForHash().get(key, prop);
    	}
    
    	public Map getHashAll(String key) {
    		Map map = new HashMap();
    		map.put("keys", template.opsForHash().keys(key));
    		map.put("vals", template.opsForHash().values(key));
    		return map;
    	}

4.8 如果我们需要一个小型的消息中间件，可以选择redis的订阅/发布来实现。

public void subscribe(String channel) {
    		RedisMessageListenerContainer container = provider.getBean(RedisMessageListenerContainer.class);
    		MessageListenerAdapter listenerAdapter = provider.getBean(MessageListenerAdapter.class);
    		container.addMessageListener(listenerAdapter, new PatternTopic(channel));
    	}
    
    	public void publish(String channel, String message) throws InterruptedException {
    		CountDownLatch latch = provider.getBean(CountDownLatch.class);
    
    		log.info("Sending message...");
    		template.convertAndSend(channel, message);
    
    		latch.await();
    	}

5. 生成一个TemplateController作为我们的启动Controller，用它来调用RedisUtils里面实现的功能。

@Controller
    public class TemplateController {
    	private Logger log = LoggerFactory.getLogger(this.getClass());
    	
    	@Autowired
    	RedisUtils utils;
    	
        @ApiOperation(value = "操作字符串/整数/浮点数", notes = "字符串", httpMethod = "GET")
        @GetMapping(value = "/setString")
        @ResponseBody
        public String setString(@ApiParam(value = "存入key", required = true) @RequestParam String key,
        		 @ApiParam(value = "存入value", required = true) @RequestParam String value) {
        	utils.setValue(key,value,100,TimeUnit.SECONDS);
        	String ret = utils.getValue(key).toString();
        	return ret;
        }
    
        @ApiOperation(value = "一次性设置多组数据，字符串/整数/浮点数", notes = "多组数据", httpMethod = "POST")
        @PostMapping(value = "/multiSet")
        @ResponseBody
        public List<Object> multiSet(@ApiParam(value = "存入key/value", required = true) @RequestBody Map keys) {
        	utils.multiSet(keys);
        	List<Object> ret = utils.multiGet(keys.keySet());
        	return ret;
        }
    
        @ApiOperation(value = "取号器 增量", notes = "取号器", httpMethod = "GET")
        @GetMapping(value = "/incr")
        @ResponseBody
        public long incr(@ApiParam(value = "key", required = true) @RequestParam String key,
        						@ApiParam(value = "增量 整数", required = true) @RequestParam long delta) {
        	return utils.incr(key, delta);
        }
    
        @ApiOperation(value = "插入队列", notes = "插入队列", httpMethod = "GET")
        @GetMapping(value = "/lpush")
        @ResponseBody
        public List<Object> lpush(@ApiParam(value = "key", required = true) @RequestParam String key,
        						@ApiParam(value = "内容", required = true) @RequestParam String content) {
        	utils.lpush(key, content);
        	return utils.range(key, 0, -1);
        }
    
        @ApiOperation(value = "推出队列", notes = "推出队列", httpMethod = "GET")
        @GetMapping(value = "/rpop")
        @ResponseBody
        public Object rpop(@ApiParam(value = "key", required = true) @RequestParam String key) {
        	return utils.rpop(key);
        }
    
        @ApiOperation(value = "Hashset", notes = "Hashset", httpMethod = "GET")
        @GetMapping(value = "/setHash")
        @ResponseBody
        public void setHash(@ApiParam(value = "key", required = true) @RequestParam String key,
        		@ApiParam(value = "Json字符串", required = true) @RequestParam String json) {
        	Map map = (Map)JSON.parse(json);
        	utils.setHash(key, map);
        }
    
        @ApiOperation(value = "HashGetAll", notes = "HashGetAll", httpMethod = "GET")
        @GetMapping(value = "/getHashAll")
        @ResponseBody
        public Map getHashAll(@ApiParam(value = "key", required = true) @RequestParam String key) {
        	return utils.getHashAll(key);
        }
    
        @ApiOperation(value = "HashGet", notes = "HashGet", httpMethod = "GET")
        @GetMapping(value = "/getHash")
        @ResponseBody
        public Object getHash(@ApiParam(value = "key", required = true) @RequestParam String key,
        					@ApiParam(value = "prop", required = true) @RequestParam String prop) {
        	
        	Object ret = utils.getHash(key, prop);
        	log.info("key:" + key + " prop:" + prop + " val:" + ret.toString());
        	return ret;
        }
    
        @ApiOperation(value = "订阅", notes = "subscribe", httpMethod = "GET")
        @GetMapping(value = "/subscribe")
        @ResponseBody
        public void subscribe(@ApiParam(value = "渠道", required = true) @RequestParam String channel) throws InterruptedException {   	
        	utils.subscribe(channel);
        }
    
        @ApiOperation(value = "发布", notes = "publish", httpMethod = "GET")
        @GetMapping(value = "/publish")
        @ResponseBody
        public void publish(@ApiParam(value = "渠道", required = true) @RequestParam String channel,
        		@ApiParam(value = "消息", required = true) @RequestParam String message) throws InterruptedException {   	
        	utils.publish(channel, message);
        }
    }

6. 运行springboot项目，访问http://localhost:8080/swagger-ui.html\#/，测试所有完成的功能。  
![SouthEast 5][]

所有代码可以到码云上找到，[springboot学习redis][springboot_redis]。

未解决问题：在redis集群环境，订阅的时候要报连接redis node失败。

[SouthEast]: /images/20220531/3dea7f56767f4319925510f995773801.png
[SouthEast 1]: /images/20220531/6d9fbc0864a340db9f1277a86c258d6c.png
[SouthEast 2]: /images/20220531/d9aeb1f1563d45b4b26429164ad833b5.png
[SouthEast 3]: /images/20220531/eccc1dcd51ec4ed199302b0ac47ceb9b.png
[SouthEast 4]: /images/20220531/3b0e414536d848b4a212feb3c6c1e773.png
[Spring Data redis]: http://blog.csdn.net/hotdust/article/details/51832926
[SouthEast 5]: /images/20220531/58557a08eaf3439b825afdc746655004.png
[springboot_redis]: https://gitee.com/39627020/StudyRedis