Mybatis Plus中的流式查询案例

@Select("select * from t_xxx t ${ew.customSqlSegment}")
@Options(resultSetType = ResultSetType.FORWARD_ONLY, fetchSize = 1000)
@ResultType(ClearReconDiffAbnormalDO.class)
void listByStream(@Param(Constants.WRAPPER) Wrapper<Model> wrapper, ResultHandler<Model> resultHandler);

public class FetchByStream extends AbstractMethod {
    private static final String METHOD = "fetchByStream";
    @Override
    public MappedStatement injectMappedStatement(Class<?> mapperClass, Class<?> modelClass, TableInfo tableInfo) {
        String sqlFormat = "<script>\nSELECT %s FROM %s %s %s\n</script>";
        String sql = String.format(sqlFormat, sqlSelectColumns(tableInfo, true),
                tableInfo.getTableName(), sqlWhereEntityWrapper(true, tableInfo),
                sqlComment());
        SqlSource sqlSource = languageDriver.createSqlSource(configuration, sql, modelClass);
        String statementName = mapperClass.getName() + DOT + METHOD;
        if (configuration.hasStatement(statementName, false)) {
            logger.warn(LEFT_SQ_BRACKET + statementName + "] Has been loaded by XML or SqlProvider or Mybatis's Annotation, so ignoring this injection for [" + getClass() + RIGHT_SQ_BRACKET);
            return null;
        }
        /* 缓存逻辑处理 */
        return builderAssistant.addMappedStatement(METHOD, sqlSource, StatementType.PREPARED, SqlCommandType.SELECT,
                Integer.MIN_VALUE, null, null, null, null, modelClass,
                ResultSetType.FORWARD_ONLY, true, true, false, null, null, null,
                configuration.getDatabaseId(), languageDriver, null);
    }
}

void fetchByStream(@Param(Constants.WRAPPER) Wrapper<T> wrapper, ResultHandler<T> handler);

package com.unionpay.dao.db2;
 
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.unionpay.entity.TblMallOrder;
import org.apache.ibatis.annotations.*;
import org.apache.ibatis.mapping.ResultSetType;
import org.apache.ibatis.session.ResultHandler;
 
/**
 * (TblMallOrder)表数据库访问层
 *
 * @author liudong
 * @since 2020-09-15 17:07:13
 */
@Mapper
public interface TblMallOrderDao extends BaseMapper<TblMallOrder> {
    @Select("${sql}")
    @Options(resultSetType = ResultSetType.FORWARD_ONLY, fetchSize = 1000)
    @ResultType(TblMallOrder.class)
    void dynamicSelectLargeData1(@Param("sql") String sql, ResultHandler<TblMallOrder> handler);
 
    @Select("${sql}")
    @Options(resultSetType = ResultSetType.FORWARD_ONLY, fetchSize = 1000)
    @ResultType(Map.class)
    void dynamicSelectLargeData2(@Param("sql") String sql, ResultHandler<Map> handler);
}

@RestController
public class TestSearchLargeData {
    // 这是每批处理的大小
    private final static int BATCH_SIZE = 1000;
    private int size;
    // 存储每批数据的临时容器
    private List<TblMallOrder> mallOrders;
 
    @Autowired
    private TblMallOrderDao tblMallOrderDao;
 
    @GetMapping("/getLargeData1")
    public void getLargeData1() {
        String sql = "select * from t_mall_order";
        tblMallOrderDao.dynamicSelectLargeData1(sql, new ResultHandler<TblMallOrder>() {
            @Override
            public void handleResult(ResultContext<? extends TblMallOrder> resultContext) {
                TblMallOrder tblMallOrder = resultContext.getResultObject();
                System.out.println(tblMallOrder);
            }
        });
    }
 
    @GetMapping("/getLargeData2")
    public void getLargeData2() {
        String sql = "select * from t_mall_order";
        tblMallOrderDao.dynamicSelectLargeData1(sql, new ResultHandler<TblMallOrder>() {
            @Override
            public void handleResult(ResultContext<? extends TblMallOrder> resultContext) {
                TblMallOrder tblMallOrder = resultContext.getResultObject();
                System.out.println(tblMallOrder);
                // 你可以看自己的项目需要分批进行处理或者单个处理，这里以分批处理为例
                mallOrders.add(tblMallOrder);
                size++;
                if (size == BATCH_SIZE) {
                    handle();
                }
            }
        });
        //用来完成最后一批数据处理
        handle();
    }
    /**
     * 数据处理
     */
    private void handle(){
        try{
            // 在这里可以对你获取到的批量结果数据进行需要的业务处理
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            // 处理完每批数据后后将临时清空
            size = 0;
            mallOrders.clear();
        }
    }
}