java中Jedis对redis内数据的批量获取- hscan,Pipeline的使用- 常用工具类-RedisUtil

指出问题

在使用redis的时候，配合jedis使用，但是发现jedis的keys* 或者mget都会造成redis阻塞，所以使用了redis的解决方案Pipeline（管道）的方式进行对redis内数据的获取。封装了以下工具类。需要自取。或者提供好的方法可以留言，我可以写进来。

初始化配置

先在application.yml配置redis基本信息

spring:
  redis:
    host: 127.0.0.1
    port: 6379
    password:
    database: 0
    timeout: 5000

创建jedis连接池

创建RedisConfig.java

@Configuration
@Slf4j
public class RedisConfig {
    @Autowired
    private RedisProperties redisProperties; //从配置文件获取redis连接配置

    private JedisPool jedisPool;

    @Autowired
    private JedisPoolConfig jedisPoolConfig;

    @Bean
    @Scope("prototype")
    /**
     * 多例模式返回redis客户端，但是在依赖注入时，建议不要使用
     * 因为jedis客户端是阻塞式IO的，而且线程不安全，
     * 所以在操作数据库的时候不要使用单例，建议从连接池获取
     */
    public redis.clients.jedis.Jedis jedis(){
        if (jedisPool!=null){
            return jedisPool.getResource();
        }else {
            return jedisPool().getResource();
        }
//        return jedisPool.getResource();
    }

    @Bean
    /**
     * 单例模式返回redis连接池
     * 依赖注入时，建议注入jedisPool，获取连接池，然后调用getResource方法获取jedis客户端
     * 使用完后，调用jedis.close()方法归还连接，节约资源
     */
    public JedisPool jedisPool(){
        synchronized (this){
            if(jedisPool==null){
                synchronized (this){
                    
                    if (StringUtils.isEmpty(redisProperties.getPassword())){
                        //无密码
                        jedisPool = new JedisPool(jedisPoolConfig,redisProperties.getHost(),redisProperties.getPort(),(int)redisProperties.getTimeout().toMillis());
                    }else {//有密码
                        jedisPool = new JedisPool(jedisPoolConfig,redisProperties.getHost(),redisProperties.getPort(),(int)redisProperties.getTimeout().toMillis(),redisProperties.getPassword());
                    }
                         
                }
            }
        }
        return jedisPool;
    }
    

    
    @Bean
    public JedisPoolConfig jedisPoolConfig(){
        // Jedis连接池配置
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        jedisPoolConfig.setMaxTotal(redisProperties.getLettuce().getPool().getMaxActive());
        // 最大空闲连接数, 默认8个
//        jedisPoolConfig.setMaxIdle(50);
        jedisPoolConfig.setMaxIdle(redisProperties.getLettuce().getPool().getMaxIdle());
        // 最大连接数, 默认8个
//        jedisPoolConfig.setMaxTotal(300);
        //最小空闲连接数, 默认0
        jedisPoolConfig.setMinIdle(redisProperties.getLettuce().getPool().getMinIdle());
//        jedisPoolConfig.setMinIdle(20);
        // 获取连接时的最大等待毫秒数(如果设置为阻塞时BlockWhenExhausted),如果超时就抛异常, 小于零:阻塞不确定的时间,  默认-1
        jedisPoolConfig.setMaxWaitMillis(redisProperties.getLettuce().getPool().getMaxWait().toMillis()); // 设置10秒


//        jedisPoolConfig.setMaxWaitMillis(3000); // 设置2秒
        //对拿到的connection进行validateObject校验
        
        jedisPoolConfig.setTestOnBorrow(true);
        return jedisPoolConfig;
    }
}

jedis工具类



@Component
@Slf4j
public class RedisUtil {
    @Autowired
    private JedisPool jedisPool;

    /**
     * findValueByKeyOfPipeline和 findValueByKeyOfPipelineMget 的区别在于 前者根据阈值获取多次 后者是直接一次性获取
     */


    /**
     * 根据(不完整)key值通过Pipeline的方式获取值
     * 先通过scan获取全部的key,再通过Pipeline获取全部的值
     * 根据countVPT 这个 阈值来控制 获取多次
     * (通过建立一个管道一次提交)
     * @param redisKey
     * @Auther erpangshou
     * @Date:2021/6/17 10:36
     **/
    public List findValueByKeyOfPipeline(String redisKey) {
        Jedis jedis = null;
        List values = new ArrayList<>();
        ArrayList>> responses = new ArrayList<>();
        int countVPT=ConstantCom.BATCH_SIZE;
        try {
            jedis = jedisPool.getResource();
            //从redis获取值刷新航迹信息
            List allKeys = findAllKeys(redisKey);
            if (Objects.isNull(allKeys) || allKeys.isEmpty()) {
                return values;
            }
            Pipeline pipelined = jedis.pipelined();
            ArrayList strings = new ArrayList<>();
            for (String key : allKeys) {
                strings.add(key);
                if (strings.size()==countVPT){
                    ArrayList subList = new ArrayList<>(strings);
                    String[] keys = subList.toArray(new String[subList.size()]);
                    Response> mget = pipelined.mget(keys);
                    responses.add(mget);
                    strings=new ArrayList<>();
                }
            }
            //最后的数据
            String[] keys = strings.toArray(new String[strings.size()]);
            Response> mget = pipelined.mget(keys);
            responses.add(mget);
            pipelined.sync();
            for (Response> respons : responses) {
                values.addAll(respons.get());
            }

        } catch (Exception e) {
            e.printStackTrace();
            log.error("redis查询异常：" + e.getMessage());
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
        return values;
    }

    /**
     * 根据(不完整)key值通过Pipeline的方式获取值
     * 先通过scan获取全部的key,再通过Pipeline获取全部的值
     * (通过Pipeline.mget直接获取)
     * @param redisKey
     * @Auther erpangshou
     * @Date:2021/6/17 10:36
     **/
    public List findValueByKeyOfPipelineMget(String redisKey) {
        Jedis jedis = null;
        List values = new ArrayList<>();
        ArrayList>> responses = new ArrayList<>();
        try {
            jedis = jedisPool.getResource();
            //从redis获取值刷新航迹信息
            List allKeys = findAllKeys(redisKey);
            if (Objects.isNull(allKeys) || allKeys.isEmpty()) {
                return values;
            }
//            allKeys=allKeys.subList(0,5000);
            String[] keys = allKeys.toArray(new String[allKeys.size()]);
            Pipeline pipelined = jedis.pipelined();
            Response> mget = pipelined.mget(keys);
            pipelined.sync();
            values=mget.get();
        } catch (Exception e) {
            log.error("redis查询异常：" + e.getMessage());
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
        return values;
    }


    /**
     * jedis的met 
     * @param redisKey
     * @Auther erpangshou
     * @return
     */
    public List mget(String redisKey) {
        Jedis jedis = null;
        List values = new ArrayList<>();
        try {
            jedis = jedisPool.getResource();
            //从redis获取值刷新航迹信息
            List allKeys = findAllKeys(redisKey);
            if (Objects.isNull(allKeys) || allKeys.isEmpty()) {
                return values;
            }
            String[] keys = allKeys.toArray(new String[allKeys.size()]);
            List mget1 = jedis.mget(keys);
            values=mget1;

        } catch (Exception e) {
            log.error("redis查询异常：" + e.getMessage());
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
        return values;
    }

    /**
     * 根据(不完整)key值通过Pipeline的方式获取值   --使用传入jedis的方式
     * 先通过scan获取全部的key,再通过Pipeline获取全部的值
     *
     * @param redisKey
     * @Auther erpangshou
     * @Date:2021/6/17 10:36
     **/
    public List findValueByKeyOfPipeline(String redisKey, Jedis jedis) {
        List values = new ArrayList<>();
        try {
            //从redis获取值刷新航迹信息
            List allKeys = findAllKeys(redisKey);
            if (allKeys == null || allKeys.isEmpty()) {
                return values;
            }
            //此处采用Pipeline，以提升性能
            Pipeline pipelined = jedis.pipelined();
            Response> mget = pipelined.mget(allKeys.toArray(new String[allKeys.size()]));
            pipelined.sync();
            values = mget.get();
        } catch (Exception e) {
            log.error("redis查询异常：" + e.getMessage());
        }
        return values;
    }

    /**
     * 根据多个完整(精确)key值通过Pipeline的方式获取值
     * 使用完整的key通过Pipeline获取全部的值
     *
     * @param allKeys
     * @Auther erpangshou
     * @Date 2021/6/17  10:33
     */
    public List findAllkeysByKeysOfPipeline(String[] allKeys) {
        Jedis jedis = null;
        List values = new ArrayList<>();
        int countVPT=ConstantCom.BATCH_SIZE;

        try {
            jedis = jedisPool.getResource();
            ArrayList>> responses = new ArrayList<>();
            //从redis获取值刷新航迹信息
            if (Objects.isNull(allKeys) || allKeys.length == 0) {
                return values;
            }
            //此处采用Pipeline，以提升性能
            Pipeline pipelined = jedis.pipelined();
            ArrayList strings = new ArrayList<>();
            for (String key : allKeys) {
                strings.add(key);
                if (strings.size()==countVPT){
                    ArrayList subList = new ArrayList<>(strings);
                    String[] keys = subList.toArray(new String[subList.size()]);
                    Response> mget = pipelined.mget(keys);
                    responses.add(mget);
                    strings=new ArrayList<>();
                }
            }
            //最后的数据
            String[] keys = strings.toArray(new String[strings.size()]);
            Response> mget = pipelined.mget(keys);
            responses.add(mget);
            pipelined.sync();

            for (Response> respons : responses) {
                values.addAll(respons.get());
            }

        } catch (Exception e) {
            log.error("redis查询异常：" + e.getMessage());
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
        return values;
    }



    /**
     * 根据多个完整(精确)key值通过Pipeline的方式获取值  --使用传入jedis的方式
     * 使用完整的key通过Pipeline获取全部的值
     *
     * @param redisKeys
     * @Auther erpangshou
     * @Date 2021/6/17  10:33
     */
    public List findAllkeysByKeysOfPipeline(String[] redisKeys, Jedis jedis) {
        List values = new ArrayList<>();
        try {
            //从redis获取值刷新航迹信息
            if (redisKeys == null || redisKeys.length == 0) {
                return values;
            }
            //此处采用Pipeline，以提升性能
            Pipeline pipelined = jedis.pipelined();
            Response> mget = pipelined.mget(redisKeys);
            pipelined.sync();
            values = mget.get();
        } catch (Exception e) {
            log.error("redis查询异常：" + e.getMessage());
        }
        return values;
    }

    /**
     * 根据keys 获取value
     *
     * @param pattern
     * @Auther erpangshou
     */
    public List findValueByKey(String pattern) {
        List values = null;
        List keys = findAllKeys(pattern);
        if (keys != null && keys.size() > 0) {
            String[] strings = keys.toArray(new String[keys.size()]);
            values = findAllkeysByKeysOfPipeline(strings);
        }
        return values;
    }

    /**
     * 根据keys 获取value  --使用传入jedis的方式
     *
     * @param pattern
     * @Auther erpangshou
     */
    public List findValueByKey(String pattern, Jedis jedis) {
        List values = null;
        List keys = findAllKeys(pattern);
        if (keys != null && keys.size() > 0) {
            String[] strings = keys.toArray(new String[keys.size()]);
            values = findAllkeysByKeysOfPipeline(strings, jedis);
        }
        return values;
    }

    /**
     * 根据keys 获取value 已map形式返回
     *
     * @param pattern
     * @Auther erpangshou
     */
    public List findValuetoMapByKey(String pattern, Jedis jedis) {
        List values = new ArrayList<>();
        ArrayList mapList = new ArrayList<>();
        try {
            List keys = findAllKeys(pattern);
            if (keys != null && keys.size() > 0) {
                String[] strings = keys.toArray(new String[keys.size()]);
                values = findAllkeysByKeysOfPipeline(strings);
            }

            if (ObjectUtil.isNotEmpty(values)) {
                for (String value : values) {
                    Map map = ((JSONObject) JSONObject.parse(value)).toJavaObject(Map.class);
                    mapList.add(map);
                }
            }

        } catch (Exception e) {
            jedis.close();
            e.printStackTrace();
        }
        return mapList;
    }

    /**
     * 因为Redis是单线程jedis.keys()方法会导致数据库阻塞，
     * 可能导致Redis其它业务无法操作，所有采用迭代模式(scan)获取数据
     * 并且keys()方法的时间复杂度为O(n)，scan()时间复杂度为O(1)
     */
    public List findAllKeys(String pattern) {
        Jedis jedis = null;
        List total = null;
        try {
            jedis = jedisPool.getResource();
            String cursor = String.valueOf(0);
            total = new ArrayList<>();
            ScanParams params = new ScanParams();
            params.match(pattern);
            do {
                params.count(ConstantCom.BATCH_SIZE);
                ScanResult result = jedis.scan(cursor, params);
                cursor = result.getStringCursor();
                total.addAll(result.getResult());
            } while (Integer.valueOf(cursor) > 0);
        } catch (NumberFormatException e) {
            e.printStackTrace();
            log.error("批量获取redis的key失败");
        } finally {
            if (ObjectUtil.isNotEmpty(jedis)) {
                jedis.close();
            }
        }
        return total;
    }

    /**
     * 获取全都的key --使用传入jedis的方式
     * 因为Redis是单线程jedis.keys()方法会导致数据库阻塞，
     * 可能导致Redis其它业务无法操作，所有采用迭代模式(scan)获取数据
     * 并且keys()方法的时间复杂度为O(n)，scan()时间复杂度为O(1)
     */
    public List findAllKeys(String pattern, Jedis jedis) {
        List total = null;
        try {
            String cursor = String.valueOf(0);
            total = new ArrayList<>();
            ScanParams params = new ScanParams();
            params.match(pattern);
            do {
                params.count(ConstantCom.BATCH_SIZE);
                ScanResult result = jedis.scan(cursor, params);
                cursor = result.getStringCursor();
                total.addAll(result.getResult());
            } while (Integer.valueOf(cursor) > 0);
        } catch (NumberFormatException e) {
            e.printStackTrace();
            log.error("批量获取redis的key失败");
        }
        return total;
    }

    /**
     * 取代jedis.keys(*)
     *
     * @param pattern
     * @return
     */
    public Set findAllKeysToSet(String pattern) {
        Jedis jedis = null;
        Set total = null;
        try {
            jedis = jedisPool.getResource();
            String cursor = String.valueOf(0);
            total = new HashSet() {
            };
            ScanParams params = new ScanParams();
            params.match(pattern);
            do {
                params.count(ConstantCom.BATCH_SIZE);
                ScanResult result = jedis.scan(cursor, params);
                cursor = result.getStringCursor();
                total.addAll(result.getResult());
            } while (Integer.valueOf(cursor) > 0);
        } catch (NumberFormatException e) {
            e.printStackTrace();
            log.error("批量获取redis的key失败");

        } finally {
            if (ObjectUtil.isNotEmpty(jedis)) {
                jedis.close();
            }
        }
        return total;
    }

    /**
     * 取代jedis.keys(*)  --使用传入jedis的方式
     * @param pattern
     * @return
     */
    public Set findAllKeysToSet(String pattern, Jedis jedis) {
        Set total = null;
        try {
            String cursor = String.valueOf(0);
            total = new HashSet();
            ScanParams params = new ScanParams();
            params.match(pattern);
            do {
                params.count(ConstantCom.BATCH_SIZE);
                ScanResult result = jedis.scan(cursor, params);
                cursor = result.getStringCursor();
                total.addAll(result.getResult());
            } while (Integer.valueOf(cursor) > 0);
        } catch (NumberFormatException e) {
            e.printStackTrace();
            log.error("批量获取redis的key失败");

        }
        return total;
    }

    /**
     * 实时数据取出，封装成指定的对象集合
     *
     * @param list        redis取出的value集合
     * @param entityClass 需要封装的对象类型
     * @param 
     * @return
     */
    public  List valueToClass(List list, Class entityClass) {
        List result = new ArrayList<>();
        for (String value : list) {
            Map map = ((JSONObject) JSONObject.parse(value)).toJavaObject(Map.class);
            Set keySet = map.keySet();
            // 将所有key转换为小写
            Map map1 = new HashMap<>();
            for (String set : keySet) {
                map1.put(set.toLowerCase(), map.get(set));
            }
            T instance = null;
            try {
                instance = entityClass.newInstance();
                BeanUtils.populate(instance, map1);
            } catch (InstantiationException e) {
                e.printStackTrace();
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            } catch (InvocationTargetException e) {
                e.printStackTrace();
            }
            result.add(instance);
        }
        return result;

    }

    public List> valueToMap(List list) {
        List> result = new ArrayList<>();
        for (String value : list) {
            if (org.apache.commons.lang3.StringUtils.isNotBlank(value)) {
                Map map = ((JSONObject) JSONObject.parse(value)).toJavaObject(Map.class);
                result.add(map);
            }
        }
        return result;
    }

    /**
     * @description redis常用方法封装get
     * @author erpangshou
     * @date 2021/6/11
     **/
    public String get(String key) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            return jedis.get(key);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (!Objects.isNull(jedis)) {
                jedis.close();
            }
        }
        return null;
    }

    /**
     * @description redis常用方法封装set
     * @author erpangshou
     * @date 2021/6/11
     **/
    public String set(String key, String value) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            return jedis.set(key, value);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (!Objects.isNull(jedis)) {
                jedis.close();
            }
        }
        return null;
    }
    

    public String hget(String key, String field) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            return jedis.hget(key, field);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (!Objects.isNull(jedis)) {
                jedis.close();
            }
        }
        return null;
    }

    /**
     * 获取redis的hash值
     * 用scan方式模仿hgetAll
     * @param pattern key
     * @Auther erpangshou
     * @return
     */

    public Map hScan(String pattern) {
        long startTime = System.currentTimeMillis();
        Jedis jedis = null;

        Map result = new HashMap<>();
        List> results = new ArrayList>();
        try {
            jedis = jedisPool.getResource();
            String cursor = String.valueOf(0);
            ScanParams params = new ScanParams();
            params.count(ConstantCom.BATCH_SIZE);

            do {
                ScanResult> scanResult = jedis.hscan(pattern, cursor, params);
                cursor = scanResult.getStringCursor();
                scanResult.getResult().stream().forEach(entry->result.put(entry.getKey(),entry.getValue()));
            } while (Integer.valueOf(cursor) > 0);
        } catch (NumberFormatException e) {
            e.printStackTrace();
            log.error("批量获取redis的key失败");
        } finally {
            if (ObjectUtil.isNotEmpty(jedis)) {
                jedis.close();
            }
        }
        System.out.println("hscansize:"+result.size());
        System.out.println("hscantime:"+(System.currentTimeMillis()-startTime));
        return result;
    }

}
pipelinejedisHscanRedis

相关

ceryx+redis实现nginx动态路由功能、动态管理upstream

Go入门笔记-20 访问redis

为毛我的redis 非保护启动修改文件有问题呢？

org.springframework.data.redis.RedisConnectionFailureException

redis集群模式

hyperf 框架 redis 安装和使用

Laravel中Redis的使用

Win10安装Redis

redis哨兵原理

Redis高可用方案：Sentinel哨兵

那一天，我被Redis主从架构支配的恐惧

Redis | 第一部分：数据结构与对象 上篇《Redis设计与实现》

标签