当前位置：首页 > article >正文

大批量查询方案简记(Mybatis流式查询)

article 2024/11/15 3:04:54

Mybatis的流式查询

摘要:

介绍使用mybatis流式查询解决大数据量查询问题.

1 业务背景

开发中遇到一个业务,说起来也很无奈:公司用的数据库MySQL,一张表里只保留了一个月的数据,但是数据量竟然高达2000W还要多,然后用户有个需求也很恶心,为了完成这个业务我需要定时任务每一个月跑一次,每次取出上一个月的历史数据,做一次计算,将需求所需要的结果数据保存到另一张表里.那么问题就是我不得不一次性去出这两千多万条数据去计算.

2 分析

既然要从MySQL里面取出这2000多万的数据,那普通的查询肯定不行了,一次性查出来先不说速度慢得问题,光内存估计就会塞爆掉.

分批查询呢? 我之前有写过一篇超300W数据的导入导出的文章,里面介绍了关于使用SQL的批处理加上事务控制以及分批读取的方式解决的,但是这个就不适用于这种业务场景,因为数据量实在太大了

如果分批读取每次查询20万你需要一百多次的查询才能将数据读出来,先不收内存够不够用,时间上就已经浪费很长时间了.

那么分析到这里,我要解决的主要问题其实就两个:

1 内存不要爆掉;

2 时间不能慢,定时任务每个月执行一次,最慢一次处理(查询,数据处理,入库等操作)要保持在15分钟以内)

3 解决方式

这里经过分析和查阅资料我找到了两种解决思路:

1 开线程查询每个线程处理一定量的数据;

2 使用游标查询;

3 使用流式查询;

首先第一种解决方案其实就是普通的分批查询只不过换成了并行方案,我们可以在SQL中使用limit 关键字,然后配合线程池以及CountDownLatch使用,处理结果最终合并每条线成的结果,

优点: 速度比单线程快,如果配合SQL的批处理这样速度不会慢.

缺点: 开销大,对数据库的压力和服务器配置的要求比较高,然后还有就是2000W还要多的数据线程数开多少合适呢?开的少了,效果不明显,开的多了可能会造成MySQL的连接数占满,导致其他查询操作阻塞.

其次第二种解决方案这种其实就是使用fetchSize控制一次读取的条数;这种方式其实也很快

优点：游标查询可以避免内存溢出的问题，主要原因会使用临时空间存储数据。

缺点：1 存在大两IO读取写入操作，在此过程中可能会引起其他业务的写入抖动；

2 磁盘空间会飙升，如果临时空间写入的表数据非常大，可能会导致磁盘空间被打满，正常情况下临时空间的数据会在读取结束或者客户端发器关闭对ResultSet操作的时候被MySQL回收。

3 客户端查询会有很长的等待时间，等待SQL响应；这个时间段内主要是MySQL在准备临时空间数据；

4 当临时空间数据准备完成之后，SQL开始响应，这时IO就可以从写入转变为读取，这时候网络响应也会开始飙升，客户端发生抖动。

最后第三种解决方案 这种流式查询，MySQL会将结果通过输出流进行结果的输出，就是向本地内核的缓存区（socket buffer）输出数据，再通过TCP链路传输将数据传送给JDBC对应的内核缓存区。

优点：速度快，不会内存溢出；

缺点：1 开启流式查询每次只能读取一行数据（注意：所以在我们处理数据的时候要尽量快，这样效率才会更高）

2 如果MySQL 服务一直向JDBC 对应的缓冲区输送数据，如果客户端Socket缓冲区满了就会导致MySQL服务阻塞；

3 相对游标流式查询对数据库的影响时间会更长一些；

4 高度依赖网络，流式查询数据量超大，处理业务复杂的情况下会有更大的概率导致网络阻塞。

5 Java会不断进行GC，数据量较大会可能会抛出GC次数过多的异常，导致业务异常停止。（使用MyBatis操作可能会出现这种情况）

4 Mybatis-plus流式查询代码示例

Service 层和 Mapper层代码示例如下

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void highFactorCalculation(String startTime, String endTime) {
        radarFactorDataMapper.selectDataByStream(result -> {
            // coding ...
        }, startTime, endTime);
        // coding ...
    }

	 /**
     * 流式获取指定时间内的数据
     *
     * @param startTime 开始时间
     * @param endTime   结束时间
     */
    @Options(resultSetType = ResultSetType.FORWARD_ONLY, fetchSize = Integer.MIN_VALUE)
    @ResultType(Pojo.class)
    void selectDataByStream(ResultHandler<Pojo> handler, @Param("startTime") String startTime, @Param("endTime") String endTime);