多线程之futureTask(future,callable)实例,jdbc数据多线程查询

朴灿烈づ我的快乐病毒、 2022-06-07 20:38 276阅读 0赞

最近遇到一个这样的功能要求。在查询数据时，由于查询的数据量比较大，一次查询（一条SQL语句中包含太多的条件）查询起来很慢，大约要10S左右才能查询出来，这样体验太不好了，需要进行优化。今天想了想，打算采用在后端把一条SQL进行拆分，拆分成为多条SQL语句，再抛给多个线程去分别查询，查询完毕后，再进行汇总的这个思路来实现。

由于会用到多线程，JAVA中对于多线程感觉就两种。一种是常用的runnable还有一种是callable。

主要区别是runnable无返回值，callable有返回值。考虑到SQL查询会有结果，需要把查询的结果拿来进行汇总，故决定采用callable来进行实现。

与Callable相关的类：

Future是Callable返回的结果

FutureTask是对Future的一个封装，让运行Future更方便点.

实现的代码主要如下：

/**
         * @param keyList 键值list
         * @return 结果, 多线程查询并汇总
         */
        public List<String> getValueListByKeyListSharding(List<String> keyList) {
            long startTime = System.currentTimeMillis();
            log.info("MYSQL查询mul开始时间:" + startTime);
            //为提高响应速度，每次分片查询并汇总
            final int shardingCount = 600;//每600条为一个分片
            if (keyList.size() > shardingCount) {
                Map<Integer, List<String>> resultMap = new TreeMap<Integer, List<String>>();
                List<Future<Map<Integer, List<String>>>> futureTaskList = new ArrayList<>();
                final ExecutorService executorService = Executors.newFixedThreadPool(10);
                for (int i = 0; i <= keyList.size() / shardingCount; i++) {
                    //每次都得到一个subList
                    List<String> subList = new ArrayList<>();
                    if ((i + 1) * shardingCount > keyList.size()) {
                        subList = keyList.subList(i * shardingCount, keyList.size());
                    } else {
                        subList = keyList.subList(i * shardingCount, shardingCount * (i + 1));
                    }
                    //拿着这个subList去DB进行查询,多线程
                    final List<String> finalSubList = subList;
                    final int finalI = i;
                    Callable<Map<Integer, List<String>>> queryCall = new Callable<Map<Integer, List<String>>>() {
                        @Override
                        public Map<Integer, List<String>> call() throws Exception {
                            Map<Integer, List<String>> subResultMap = new TreeMap<Integer, List<String>>();
                            subResultMap.put(finalI, getValueListByKeyList(new CopyOnWriteArrayList<String>(finalSubList)));
                            return subResultMap;
                        }
                    };
                    //添加到task中,执行查询
                    FutureTask<Map<Integer, List<String>>> futureTask = new FutureTask<Map<Integer, List<String>>>(queryCall);
                    futureTaskList.add(futureTask);
                    executorService.submit(futureTask);//执行查询
                }
                //待所有查询完毕后，再进行顺序汇总
                for (Future<Map<Integer, List<String>>> futureTask : futureTaskList) {
                    try {
                        Map<Integer, List<String>> integerListMap = futureTask.get();
                        resultMap.putAll(integerListMap);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                //进行结果汇总
                List<String> resultList = new ArrayList<>();
                Set<Integer> mapKeys = resultMap.keySet();
                for (Integer mapItem : mapKeys) {
                    resultList.addAll(resultMap.get(mapItem));
                }
                long endTime = System.currentTimeMillis();
                log.info("本次mul数据查询(mysql)耗时:" + (endTime - startTime));
                log.info("本次mul数据查询(mysql)结束时间:" + endTime);
                executorService.shutdown();//关闭线程池
                return resultList;
            } else {
                return getValueListByKeyList(keyList);
    //            return getValueListByKeyList(new CopyOnWriteArrayList<String>(keyList));
            }
        }

其中getValueListByKeyList是一个进行数据查询的具体方法

/**
         * @param keyList 键值list
         * @return valueList, 如果没有这个值或这条记录，也需要返回一个空串在list中
         */
        public List<String> getValueListByKeyList(List<String> keyList) {
            long startTime = System.currentTimeMillis();
            log.info("MYSQL查询开始时间:" + startTime);
            String sql = "SELECT `key`,`value` FROM  tb_super where `KEY` in (:ids)";
    
            JdbcTemplate jdbcTemplate = getJdbcTemplate();
            NamedParameterJdbcTemplate namedParameterJdbcTemplate = new NamedParameterJdbcTemplate(jdbcTemplate);
            MapSqlParameterSource parameters = new MapSqlParameterSource();
            parameters.addValue("ids", keyList);
            List<Map<String, String>> list = namedParameterJdbcTemplate.queryForList(sql, parameters);
            List<String> resultList = new ArrayList<>();
            //<editor-fold desc="通过循环迭代遍历，获取出对应的数据">
    //        Iterator<String> resultIterator = keyList.iterator();
    //        while (resultIterator.hasNext()) {
    //            String keyItem = resultIterator.next();
    //            boolean hasValue = false;
    //            Iterator<Map<String, String>> mapIterator = list.iterator();
    //            MARKER:
    //            while (mapIterator.hasNext()) {
    //                Map<String, String> mapItem = mapIterator.next();
    //                String key = mapItem.get("key") == null ? "" : mapItem.get("key");
    //                String value = mapItem.get("value") == null ? "" : mapItem.get("value");
    //                if (keyItem.equals(key)) {//说明这条KEY有查询结果
    //                    resultList.add(value);
    //                    hasValue = true;
    //                    //在两个集合中都要去掉这个已经查找到的数据节点
    //                    resultIterator.remove();
    //                    mapIterator.remove();
    //                    continue MARKER;//跳出，进行外层循环
    //                }
    //            }
    //            if (!hasValue) {
    //                resultList.add("");//说明没有查询结果，则赋空值
    //            }
    //        }
            //</editor-fold>
    
    
            //<editor-fold desc="全部遍历方式">
            for (String keyItem : keyList) {
                boolean hasValue = false;
                MARKER:
                for (Map<String, String> listItem : list) {
                    String key = listItem.get("key") == null ? "" : listItem.get("key");
                    String value = listItem.get("value") == null ? "" : listItem.get("value");
                    if (keyItem.equals(key)) {//说明这条KEY有查询结果
                        resultList.add(value);
                        hasValue = true;
                        continue MARKER;//跳出，进行外层循环
                    }
                }
                if (!hasValue) {
                    resultList.add("");//说明没有查询结果，则赋空值
                }
            }
            //</editor-fold>
    
            long endTime = System.currentTimeMillis();
            log.info("本次数据查询(mysql)耗时:" + (endTime - startTime));
            log.info("本次数据查询(mysql)结束时间:" + endTime);
            return resultList;
        }

查询相同的数据，以前没有拆分时，需要10S左右，现在查询时，只需要3S左右就可以查完了。相比以前，查询还是有了很大的提高。