流式查询1. mybatis的游标Cursor，分页大数据查询

mybatis,cursor · 浏览次数 : 34

小编点评

流式查询是一种数据库查询方式，它允许应用程序从数据库中一次获取一条记录，而不是一次性获取所有记录。这种方式可以降低内存使用，特别是在处理大量数据时。流式查询的优点在于它可以减少内存占用，提高查询效率。在MyBatis Plus中，游标分页是一种常用的流式查询实现方式。使用游标分页，可以在mybatis-config.xml或application.yml中添加相应的配置。例如： ```xml mybatis-plus: configuration: settings: useCursorFetch: true ``` 在Mapper接口中，定义一个返回Cursor类型的方法，例如： ```java public interface CustomerLoyCardMapper { Cursor getCardCursorByBatchIdOrCardNum(String batchId, String startingNumber, String endingNumber, String cardRange); } ``` 实现类需要实现该方法，例如： ```java @Override public Cursor getCardCursorByBatchIdOrCardNum(String batchId, String startingNumber, String endingNumber, String cardRange) { // 实现方法逻辑 } ``` SQL语句应该使用`resultType`和`resultSetType`指定结果类型和游标类型，例如： ```xml ``` 使用游标分页时，需要注意以下几点： 1. 确保数据库支持游标分页功能。 2. 根据实际情况设置合适的分页大小。 3. 及时关闭游标，释放资源。 4. 进行性能监控和测试，确保游标分页带来的性能提升符合预期。 5. 合理使用缓存，提高查询效率。 6. 关注MyBatis Plus的最新版本特性和改进。总之，流式查询和游标分页是一种提高查询效率和降低内存占用的有效方式，但在使用时需要注意数据库支持、分页大小、连接管理等方面的问题。

正文

流式查询
流式查询指的是查询成功后不是返回一个集合而是返回一个迭代器，应用可以通过迭代器每次取一条查询结果。流式查询的好处是能够降低内存使用。例如我们想要从数据库取 1000 万条记录而又没有足够的内存时，就不得不分页查询。

而分页查询就需要我们按照顺序查询并设置一个参数来记录当前进度并在下次查询时将进度作为参数传入。

（比如按id升序查询，记录每次查询结果的最大id, 下次查询将这个最大id传入只查询大于这个id的），否则就会出现深度分页的情况。而查询效率取决于表设计，如果设计的不好，那么每次查询都会是一次单独的低效查询。
而流式查询不需要自己记录进度（数据库来记录），且即使表设计的较差或在sql比较复杂，也仅仅只需要一次低效查询。
流式查询的过程当中，要保证数据库连接是保持打开状态的，否则就会流关闭。
MyBaits通过游标Cursor实现了流式查询。 MyBaits Plus基于Mybais, 自然也是支持的。

如何使用
写一个获取流的Mapper
不需要其他配置，像平常我们写查询一样在Mapper定义查询，并将返回结果设为Cursor即可实现一个流式查询。
sql也正常按照时间查询条件写，不需要加limit之类的参数。

1. 配置游标分页

在MyBatis Plus中，使用游标分页需要在mybatis-config.xml或application.yml 中添加配置:

mybatis-plus:
configuration:
settings:
useCursorFetch: true

    @Transactional
    @Async("asyncServiceExecutor")
    public void asyncSaveAndUpdateCardPool(String buId, CardStatusReqVO cardReqVO) {
        try (ShardingCtx s = ShardingCtx.setShardingValue(buId)) {
            log.info("asyncSaveAndUpdateCardPool: {}, {}", buId, cardReqVO);

            try (Cursor<CardUpdateVO> cursor = memberCardRepository.getCardCursorByBatchIdOrCardNum(
                    cardReqVO.getBatchId(),
                    cardReqVO.getStartingNumber(),
                    cardReqVO.getEndingNumber(),
                    cardReqVO.getCardRange())) {

                final int BATCH_SIZE = 1000;
                AtomicBoolean isFirstIteration = new AtomicBoolean(true);
                CopyOnWriteArrayList<CardUpdateVO> updateVOList = new CopyOnWriteArrayList<>();

                for (CardUpdateVO cardUpdateVO : cursor) {
                    updateVOList.add(cardUpdateVO);
                    if (updateVOList.size() == BATCH_SIZE) {
                        batchUpdateCardStatusAsync(buId, updateVOList, cardReqVO, isFirstIteration);
                        updateVOList.clear();
                    }
                }

                if (!updateVOList.isEmpty()) {
                    batchUpdateCardStatusAsync(buId, updateVOList, cardReqVO, isFirstIteration);
                }
            } catch (Exception e) {
                log.error("Cursor process error: ", e);
            }
        }
        log.info("Finished asyncSaveAndUpdateCardPool for cardReqVO: {}, {}", buId, cardReqVO);
    }

接口：
Cursor<CardUpdateVO> getCardCursorByBatchIdOrCardNum(String batchId, String startingNumber, String endingNumber, String cardRange);

实现类：
 @Override
 public Cursor<CardUpdateVO> getCardCursorByBatchIdOrCardNum(String batchId, String startingNumber, String endingNumber, String cardRange) {
     return customerLoyCardMapper.getCardCursorByBatchIdOrCardNum(batchId, startingNumber, endingNumber, cardRange);
 }
    
mapper:
Cursor<CardUpdateVO> getCardCursorByBatchIdOrCardNum(@Param("batchId")String batchId,
                                                         @Param("startingNumber")String startingNumber,
                                                         @Param("endingNumber")String endingNumber,
                                                         @Param("cardRange")String cardRange);
sql:
<select id="getCardCursorByBatchIdOrCardNum" resultType="com.aswatson.csc.member.req.CardUpdateVO" resultSetType="FORWARD_ONLY" fetchSize = "500">
        SELECT M.BU_ID as buId,
        M.MEMBER_ID as memberId,
        M.PROGRAM_ID as programId,
        C.CARD_ID as cardId,
        C.CARD_NUM as cardNum,
        C.VISIBLE_CARD as visibleCard,
        C.CARD_TYPE_CD as cardTypeCd,
        C.STATUS_CD as statusCd,
        C.ACTIVE_FLAG as activeFlag
        FROM CUSTOMER_LOY_CARD C, CUSTOMER_LOY_MEMBER M
        WHERE C.MEMBER_ID = M.MEMBER_ID
        <if test="startingNumber != null and endingNumber != null and cardRange == 1 ">
            AND C.VISIBLE_CARD BETWEEN #{startingNumber} AND #{endingNumber}
        </if>
        <if test="batchId != null and batchId != '' and cardRange == 2 ">
            AND C.BATCH_ID = #{batchId}
        </if>
    </select>

需要注意的点：

获取一个打开状态的连接。

上面提过，使用流式查询需要保证数据库连接是保持打开状态。而正常情况下我们使用mybaits执行一次查询，连接都会被关闭或在重置。因此我们需要一些方法来保持连接。

使用事务：事务执行完毕之前连接会一直保持因此，我们可以来使用事务来保持连接。

这是最简单的方法，但是需要注意的是，由于cursor在遍历结束的方法末尾后会主动关闭连接。因此：

1. 方法内事务是正常的，在cursor查询前和遍历后的的数据操作依然是一个事务。从这里可以猜测，连接并不在在遍历完立即关闭，而是采用了类似AOP的手段在方法末尾关闭。
2. 如果你使用了数据库连接池，那么池中的这个连接会被关掉，这可能并不会导致你的连接池数量减少，因为连接池的连接数据依赖于各自的规则，
当有连接断开，连接池会根据自己的策略对做相应的处理，比如重新建立，因此不需要过分关注这个，需要注意的是这个行为可能带来的影响，
比如druid连接池，druid在事务执行完毕后会进行连接的清理。但是这个连接在myDbTableMapper遍历完成时已经关闭，
就会导致日志打印java.sql.SQLException: No operations allowed after statement closed。这个问题不会带来数据影响，
但是依然属于一个错误，见github-druid-issues，在1.2.10， druid将这个日志改为debug级别进行了临时的屏蔽。其他连接池尚未进行测试。

循环获取数据
Cursor继承了迭代器，可以通过Cursor获取他的Iterator, 或者直接使用for循环来获取数据。
需要注意的是，Cursor在查询一瞬间数据就固化了，如果你先查询，再更新，再遍历，即使在同一个事务里面，获得的结果也是未更新的数据。

关闭流
cursor在遍历结束后会主动关闭连接。如果未遍历结束中途退出，可以调用cursor的close方法关闭连接。推荐finaly内总是调用close或者try(resoure)来保证连接总是被正常关闭。

在使用MyBatis Plus的游标分页时，有一些注意事项和建议：

数据库支持：游标分页依赖数据库的游标支持，因此确保数据库支持游标分页功能。

分页大小：需要根据实际情况设置合适的分页大小，过小可能导致频繁查询，过大可能失去游标分页的优势。

及时关闭游标：使用Cursor时，确保在处理完数据后及时关闭游标，释放资源。

性能监控：对于大数据量的场景，建议进行性能监控和测试，确保游标分页带来的性能提升符合预期。

合理使用缓存：需要根据实际情况考虑是否使用缓存，以及如何合理使用缓存，以提高查询效率。

版本更新： MyBatis Plus的版本可能会更新，建议关注最新版本的特性和改进，以获取更好的支持和性能。
————————————————