使用RedisTemplate批量存入数据，100万测试需要1分钟

1.需求

2.基本配置

3.核心代码

4.dao

5.mapper/*.xml

6.数据表

7.工具类

参考文档

https://blog.csdn.net/xiaoliu598906167/article/details/82218525

https://blog.csdn.net/fouling/article/details/98631144

https://blog.csdn.net/supersub000/article/details/80100016

1.需求

将mysql数据库中的上百万数据存入redis中存入的格式为hashMap的形式
即，（hash路径，key, obj）其中的obj为map 封装一条记录信息
并且这些数据在用的时候可以取得到。
难点，普通通过循环hset的方式比较慢，一万条数据需要6秒左右，一百万需要10分钟左右
优化方案：redisTemplate的管道api executePipelined()
注意，仍需要将数据进行分段向redis中存入（不然被卡死），比如可以分段从数据库取然后分段存，
但是最好是一下子取出来，再分段存入，本案例是分段取分段存。

2.基本配置

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

基本配置类（必须）

package com.example.studyspringboot.studyboot.utils.stuRedis;
import com.alibaba.fastjson.support.spring.GenericFastJsonRedisSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;@Configuration
public class RedisConfig {@SuppressWarnings("rawtypes")@Autowiredprivate RedisTemplate redisTemplate;/*** 解决redis插入中文乱码* @return*/@SuppressWarnings({ "unchecked", "rawtypes" })@Beanpublic RedisTemplate redisTemplateInit() {//设置序列化Key的实例化对象redisTemplate.setKeySerializer(new StringRedisSerializer());redisTemplate.setHashKeySerializer(new StringRedisSerializer());//设置序列化Value的实例化对象redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());redisTemplate.setHashValueSerializer(new GenericFastJsonRedisSerializer());return redisTemplate;}
}

配置文件


#MYSQL链接
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.url=jdbc:mysql://localhost:3306/redistest?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai
spring.datasource.username=root
spring.datasource.password=lps123
# AOP
# spring.aop.auto=true
# 默认使用cglib动态代理
# 强制使用jdk代理（不要这样做）
#spring.aop.proxy-target-class=false#mybatis配置
mybatis.mapper-locations = classpath:/mapper/*.xml
#mybatis.type-aliases-package=com.customerNoPlatform.entity#redis配置
#Redis服务器地址
spring.redis.host=127.0.0.1
#Redis服务器连接端口
spring.redis.port=6379
#Redis数据库索引（默认为0）
spring.redis.database=0
#连接池最大连接数（使用负值表示没有限制）
spring.redis.jedis.pool.max-active=50
#连接池最大阻塞等待时间（使用负值表示没有限制）
spring.redis.jedis.pool.max-wait=3000
#连接池中的最大空闲连接
spring.redis.jedis.pool.max-idle=20
#连接池中的最小空闲连接
spring.redis.jedis.pool.min-idle=2
#连接超时时间（毫秒）
spring.redis.timeout=5000

3.核心代码

package com.example.studyspringboot.studyboot.utils.stuRedis;import com.alibaba.fastjson.JSON;
import com.example.studyspringboot.studyboot.dao.UserDao;
import org.apache.catalina.Pipeline;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;import javax.annotation.Resource;
import java.util.List;
import java.util.Map;@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest
public class Main {@Autowiredprivate RedisTemplate redisTemplate;@Autowiredprivate UserDao userDao;@Autowired RedisUtil redisUtil;/*** 读取缓存数据*/@Testpublic void get() {//  redisUtils.get("908687_name_2726||908687_sex_2726||908687_hobby_2726908687_height_2726");Map<String,Object> value = (Map<String,Object>)redisUtil.hget("nbs:hello:leaf","908687_name_2726||908687_sex_2726||908687_hobby_2726908687_height_2726");System.out.println(JSON.toJSONString(value));System.out.println(value.get("name"));}@Testpublic void method2() {CustomersToRedis("nbs:hello:leaf",0);}/*** @Author lps* @Description //TODO* @Date 2020-07-25 15:21:44 星期六* @Param [path, stepSize] pathhash值的路径，stepSize 步长多少数据查一次* @return boolean**/public boolean CustomersToRedis(String path,int stepSize) {long zero =  System.currentTimeMillis();//数据总量int total = userDao.getCount();if(total==0){return false;}stepSize=stepSize==0?100000:stepSize;//分步int step =(total%stepSize)==0?(total/stepSize):(total/stepSize+1);long startTime;//开始时间long selectTime;//查询花费时间long endTime;//结束时间long saveTime;//存入redis花费时间long dis;for(int j = 1;j<=step;j++){System.out.println("######第"+j+"次操作######");startTime= System.currentTimeMillis();//查出stepSize条数据List<Map<String,Object>> res=userDao.getLimitUser((j-1)*stepSize,stepSize);selectTime = System.currentTimeMillis();System.out.println("查询花费时间："+(selectTime-startTime)+"ms");//将这些数据存入redishandleSave(path, res);saveTime = System.currentTimeMillis();System.out.println("存入redis话费时间："+(saveTime-selectTime)+"ms");endTime= System.currentTimeMillis();dis= endTime-startTime;System.out.println("每十万条数据花费时间："+dis+"ms");System.out.println();//换行}long last =  System.currentTimeMillis();System.out.println("总共花费时间"+(last-zero)+"ms");return true;}//RedisTemplate使用PipeLine//使用管道存储public void handleSave(String path,List<Map<String, Object>> maps) {List resultList = redisTemplate.executePipelined(new RedisCallback<String>() {@Overridepublic String doInRedis(RedisConnection connection) throws DataAccessException {connection.openPipeline();String tempKey="";for(Map<String,Object> map:maps){tempKey=map.get("name")+"||"+map.get("sex")+"||"+map.get("hobby")+map.get("height");connection.hSet(path.getBytes(),tempKey.getBytes(),redisTemplate.getValueSerializer().serialize(map));}return null;}});System.out.println("长度："+resultList.size());}
}

4.dao

package com.example.studyspringboot.studyboot.dao;import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;import java.util.List;
import java.util.Map;@Repository
public interface UserDao {//分页查询List<Map<String,Object>> getLimitUser(@Param("st") int st, @Param("ed") int ed);//总条数Integer getCount();
}

**5.mapper/*.xml**

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
<mapper namespace="com.example.studyspringboot.studyboot.dao.UserDao"><select id="getLimitUser"  resultType="java.util.Map">select * from user limit #{st},#{ed};</select><select id="getCount"  resultType="java.lang.Integer">select COUNT(*)cnt  from user;</select></mapper>

部分结果展示

redis存储结果

后台打印输出

可以看出查询花费由于是本机时间还不算长，如果是远程连接时间可能要长得多。

每次存10万，100万的数据大概需要56秒，如果不算链接查询的时间可能40秒就可以了。

6.数据表

7.工具类

package com.example.studyspringboot.studyboot.utils.stuRedis;import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;@Component
@Slf4j
public class RedisUtil {//锁名称public static final String LOCK_PREFIX = "redis_lock";//加锁失效时间，毫秒public static final int LOCK_EXPIRE = 300; // ms@Resourceprivate RedisTemplate<String, Object> redisTemplate;public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {this.redisTemplate = redisTemplate;}//=============================common============================  /*** 指定缓存失效时间** @param key  键* @param time 时间(秒)* @return*/public boolean expire(String key, long time) {try {if (time > 0) {redisTemplate.expire(key, time, TimeUnit.SECONDS);}log.info("指定redis缓存失效时间成功，key:{},time:{}",key,time);return true;} catch (Exception e) {e.printStackTrace();log.info("指定redis缓存失效时间失败，key:{},time:{}",key,time);return false;}}/*** 根据key 获取过期时间** @param key 键 不能为null* @return 时间(秒) 返回0代表为永久有效*/public long getExpire(String key) {long time=redisTemplate.getExpire(key, TimeUnit.SECONDS);log.info("根据key获取redis缓存失效时间，key:{},time:{}",key,time);return time;}/*** 判断key是否存在** @param key 键* @return true 存在 false不存在*/public boolean hasKey(String key) {try {boolean flag=redisTemplate.hasKey(key);log.info("判断redis是否包含key，key:{},return:{}",key,flag);return flag;} catch (Exception e) {e.printStackTrace();return false;}}/*** 删除缓存** @param key 可以传一个值 或多个*/@SuppressWarnings("unchecked")public void del(String... key) {if (key != null && key.length > 0) {if (key.length == 1) {boolean flag=redisTemplate.delete(key[0]);log.info("从redis删除key，key:{},return:{}",key,flag);} else {long count=redisTemplate.delete(CollectionUtils.arrayToList(key));log.info("从redis删除key，key:{},return:{}",key,count);}}}//============================String=============================  /*** 普通缓存获取** @param key 键* @return 值*/public Object get(String key) {if(key==null) {log.info("从redis获取key，key:{},return:{}",key,null);return null;}else {Object object=redisTemplate.opsForValue().get(key);log.info("从redis获取key，key:{},return:{}",key,object);return object;}}/*** 普通缓存放入** @param key   键* @param value 值* @return true成功 false失败*/public boolean set(String key, Object value) {try {redisTemplate.opsForValue().set(key, value);log.info("保存键值到redis成功，key:{},value:{}",key,value);return true;} catch (Exception e) {e.printStackTrace();log.info("保存键值到redis失败，key:{},value:{}",key,value);return false;}}/*** 普通缓存放入并设置时间** @param key   键* @param value 值* @param time  时间(秒) time要大于0 如果time小于等于0 将设置无限期* @return true成功 false 失败*/public boolean set(String key, Object value, long time) {try {if (time > 0) {redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);log.info("保存键值到redis成功并设定失效时间，key:{},value:{},time:{},",key,value,time);} else {set(key, value);}return true;} catch (Exception e) {e.printStackTrace();return false;}}//================================Map=================================  /*** HashGet** @param key  键 不能为null* @param item 项 不能为null* @return 值*/public Object hget(String key, String item) {return redisTemplate.opsForHash().get(key, item);}/*** 获取hashKey对应的所有键值** @param key 键* @return 对应的多个键值*/public Map<Object, Object> hmget(String key) {return redisTemplate.opsForHash().entries(key);}/*** HashSet** @param key 键* @param map 对应多个键值* @return true 成功 false 失败*/public boolean hmset(String key, Map<String, Object> map) {try {redisTemplate.opsForHash().putAll(key, map);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** HashSet 并设置时间** @param key  键* @param map  对应多个键值* @param time 时间(秒)* @return true成功 false失败*/public boolean hmset(String key, Map<String, Object> map, long time) {try {redisTemplate.opsForHash().putAll(key, map);if (time > 0) {expire(key, time);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 向一张hash表中放入数据,如果不存在将创建** @param key   键* @param item  项* @param value 值* @return true 成功 false失败*/public boolean hset(String key, String item, Object value) {try {redisTemplate.opsForHash().put(key, item, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 向一张hash表中放入数据,如果不存在将创建** @param key   键* @param item  项* @param value 值* @param time  时间(秒)  注意:如果已存在的hash表有时间,这里将会替换原有的时间* @return true 成功 false失败*/public boolean hset(String key, String item, Object value, long time) {try {redisTemplate.opsForHash().put(key, item, value);if (time > 0) {expire(key, time);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 删除hash表中的值** @param key  键 不能为null* @param item 项 可以使多个 不能为null*/public void hdel(String key, Object... item) {redisTemplate.opsForHash().delete(key, item);}/*** 判断hash表中是否有该项的值** @param key  键 不能为null* @param item 项 不能为null* @return true 存在 false不存在*/public boolean hHasKey(String key, String item) {return redisTemplate.opsForHash().hasKey(key, item);}//============================set=============================  /*** 根据key获取Set中的所有值** @param key 键* @return*/public Set<Object> sGet(String key) {try {return redisTemplate.opsForSet().members(key);} catch (Exception e) {e.printStackTrace();return null;}}/*** 根据value从一个set中查询,是否存在** @param key   键* @param value 值* @return true 存在 false不存在*/public boolean sHasKey(String key, Object value) {try {return redisTemplate.opsForSet().isMember(key, value);} catch (Exception e) {e.printStackTrace();return false;}}/*** 将数据放入set缓存** @param key    键* @param values 值 可以是多个* @return 成功个数*/public long sSet(String key, Object... values) {try {return redisTemplate.opsForSet().add(key, values);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 将set数据放入缓存** @param key    键* @param time   时间(秒)* @param values 值 可以是多个* @return 成功个数*/public long sSetAndTime(String key, long time, Object... values) {try {Long count = redisTemplate.opsForSet().add(key, values);if (time > 0) expire(key, time);return count;} catch (Exception e) {e.printStackTrace();return 0;}}/*** 获取set缓存的长度** @param key 键* @return*/public long sGetSetSize(String key) {try {return redisTemplate.opsForSet().size(key);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 移除值为value的** @param key    键* @param values 值 可以是多个* @return 移除的个数*/public long setRemove(String key, Object... values) {try {Long count = redisTemplate.opsForSet().remove(key, values);return count;} catch (Exception e) {e.printStackTrace();return 0;}}//===============================list=================================  /*** 获取list缓存的内容** @param key   键* @param start 开始* @param end   结束  0 到 -1代表所有值* @return*/public List<Object> lGet(String key, long start, long end) {try {return redisTemplate.opsForList().range(key, start, end);} catch (Exception e) {e.printStackTrace();return null;}}/*** 获取list缓存的长度** @param key 键* @return*/public long lGetListSize(String key) {try {return redisTemplate.opsForList().size(key);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 通过索引 获取list中的值** @param key   键* @param index 索引  index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推* @return*/public Object lGetIndex(String key, long index) {try {return redisTemplate.opsForList().index(key, index);} catch (Exception e) {e.printStackTrace();return null;}}/*** 将list放入缓存** @param key   键* @param value 值* @param time  时间(秒)* @return*/public boolean lSet(String key, Object value) {try {redisTemplate.opsForList().rightPush(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 将list放入缓存** @param key   键* @param value 值* @param time  时间(秒)* @return*/public boolean lSet(String key, Object value, long time) {try {redisTemplate.opsForList().rightPush(key, value);if (time > 0) expire(key, time);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 将list放入缓存** @param key   键* @param value 值* @param time  时间(秒)* @return*/public boolean lSet(String key, List<Object> value) {try {redisTemplate.opsForList().rightPushAll(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 将list放入缓存** @param key   键* @param value 值* @param time  时间(秒)* @return*/public boolean lSet(String key, List<Object> value, long time) {try {redisTemplate.opsForList().rightPushAll(key, value);if (time > 0) expire(key, time);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 根据索引修改list中的某条数据** @param key   键* @param index 索引* @param value 值* @return*/public boolean lUpdateIndex(String key, long index, Object value) {try {redisTemplate.opsForList().set(key, index, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 移除N个值为value** @param key   键* @param count 移除多少个* @param value 值* @return 移除的个数*/public long lRemove(String key, long count, Object value) {try {Long remove = redisTemplate.opsForList().remove(key, count, value);return remove;} catch (Exception e) {e.printStackTrace();return 0;}}/***  分布式锁** @param key key值* @return 是否获取到*/@SuppressWarnings("unchecked")public boolean lock(String key){String lock = LOCK_PREFIX + key;return (Boolean) redisTemplate.execute((RedisCallback) connection -> {long expireAt = System.currentTimeMillis() + LOCK_EXPIRE + 1;Boolean acquire = connection.setNX(lock.getBytes(), String.valueOf(expireAt).getBytes());if (acquire) {return true;} else {byte[] value = connection.get(lock.getBytes());if (Objects.nonNull(value) && value.length > 0) {long expireTime = Long.parseLong(new String(value));// 如果锁已经过期if (expireTime < System.currentTimeMillis()) {// 重新加锁，防止死锁byte[] oldValue = connection.getSet(lock.getBytes(), String.valueOf(System.currentTimeMillis() + LOCK_EXPIRE + 1).getBytes());return Long.parseLong(new String(oldValue)) < System.currentTimeMillis();}}}return false;});}/*** 删除锁** @param key*/public void delete(String key) {redisTemplate.delete(key);}
}

8.一次查询，分批插入的效率对比

核心代码

    @Testpublic void method3() {long startTime = System.currentTimeMillis();int total = userDao.getCount();List<Map<String,Object>> resList = userDao.getLimitUser(0,total);long selectTime = System.currentTimeMillis();System.out.println("查询数据花费时间："+(selectTime-startTime)+"ms");int len = resList.size();//实际数据总量(等于)totalint stepSize=100000;//步长为10万；String path="nbs:hello:leaf";int cnt=1;long startRedisTime;long endRedisTime= -System.currentTimeMillis();List<Map<String,Object>> tempList = new ArrayList<>();for(int i=0;i<len;i++){tempList.add(resList.get(i));if(i%stepSize==0&&i!=0){System.out.println("第"+cnt+"次向redis中存入数据");startRedisTime = System.currentTimeMillis();handleSave(path,tempList);tempList.clear();//清除内存cnt++;endRedisTime = System.currentTimeMillis();System.out.println("存入redi花费时间："+( endRedisTime-startRedisTime)+"ms");}}//存在不能恰好整除的，需要额外处理if(tempList.size()>0){startRedisTime = System.currentTimeMillis();handleSave(path,tempList);endRedisTime = System.currentTimeMillis();tempList.clear();//清除内存System.out.println("存入redi花费时间："+( endRedisTime-startRedisTime)+"ms");}System.out.println("总共花费的时间为："+(endRedisTime-startTime)+"ms");}

运行结果

查询数据花费时间：17521ms
第1次向redis中存入数据
长度：100001
存入redi花费时间：6233ms
第2次向redis中存入数据
长度：100000
存入redi花费时间：4107ms
第3次向redis中存入数据
长度：100000
存入redi花费时间：3579ms
第4次向redis中存入数据
长度：100000
存入redi花费时间：3406ms
第5次向redis中存入数据
长度：100000
存入redi花费时间：3384ms
第6次向redis中存入数据
长度：100000
存入redi花费时间：3468ms
第7次向redis中存入数据
长度：100000
存入redi花费时间：3382ms
第8次向redis中存入数据
长度：100000
存入redi花费时间：3359ms
第9次向redis中存入数据
长度：100000
存入redi花费时间：3742ms
第10次向redis中存入数据
长度：100000
存入redi花费时间：4318ms
长度：24
存入redi花费时间：6ms
总共花费的时间为：56535ms

可能由于都是本地连接，而不是远程连接。所以看不出来差别。

（1）值得注意的是大量数据如果一次性从数据库读到内存中，操作起来程序就会崩溃。比如一次性读入300万的数据就无法继续进行存储操作了。所以这个一次性读入，分批存的想法经过测试是行不通的。

（2）另外随着数据量的增加，即便是分批从数据库读，再分批写写入redis中，效率也越来也低。100万数据需要1分钟。

300百万数据却要4分钟。如果数据进一步增加的话可能就无法胜任了。具体如下：