[大数据]waterfallProcess

 CodeUtil.waterfallProcess(new WaterfallProcessInterface<FrontendCategoryEntity, Long>() {
                @Override
                public List<FrontendCategoryEntity> getPage(Long sortParam, int size) {
                    return repo.getVirtualAllWaterfall(sortParam, size);
                }

                @Override
                public void process(List<FrontendCategoryEntity> pageList) {
                    processVirtual(pageList);
                }

                @Override
                public int getSize() {
                    return 100;
                }

                @Override
                public Long getNextParam(FrontendCategoryEntity last) {
                    return last.getId();
                }

                @Override
                public Long getFirstParam() {
                    return 0L;
                }
            });

LambdaQueryWrapper<FrontendCategoryDo> wrapper = new LambdaQueryWrapper<>();

            wrapper.eq(FrontendCategoryDo::getIsDelete, SystemConst.SCALE_VALUE0)
                    .eq(FrontendCategoryDo::getIsAll, SystemConst.SCALE_VALUE1)
                    .eq(FrontendCategoryDo::getLevel, SystemConst.SCALE_VALUE3)
                    .eq(FrontendCategoryDo::getName, "全部")
                    .gt(FrontendCategoryDo::getId, startId)
                    .orderByAsc(FrontendCategoryDo::getId)
                    .last("limit " + count);
     List<FrontendCategoryEntity> frontendCategoryEntities = frontendCategoryMapper.selectList(wrapper).stream().map(converter::do2Entity).collect(Collectors.toList());

import java.util.List;

public interface WaterfallProcessInterface<T, P> {

    /**
     * 查询分页数据
     *
     * @param sortParam
     * @param size
     * @return
     */
    List<T> getPage(P sortParam, int size) throws Exception;

    /**
     * 处理
     *
     * @param pageList
     */
    void process(List<T> pageList);

    /**
     * 分页大小
     *
     * @return
     */
    int getSize();

    /**
     * 获取下一次查询参数标记
     *
     * @param last
     * @return
     */
    P getNextParam(T last);

    /**
     * 获取首次参数
     *
     * @return
     */
    P getFirstParam();
}

public class CodeUtil {

    private CodeUtil() {
    }

    /**
     * 注: 多个动态参数可包装对象
     *
     * @param waterfallProcessInterface
     * @param <T>                       查询的对象
     * @param <P>                       动态参数类型
     */
    public static <T, P> long waterfallProcess(WaterfallProcessInterface<T, P> waterfallProcessInterface) throws Exception {
        long count = 0;
        int size = waterfallProcessInterface.getSize();
        List<T> page = waterfallProcessInterface.getPage(waterfallProcessInterface.getFirstParam(), size);
        while (!CollectionUtils.isEmpty(page)) {
            if (Thread.currentThread().isInterrupted()) {
                throw new InterruptedException();
            }

            count += page.size();
            waterfallProcessInterface.process(page);
            if (page.size() < size) {
                break;
            }

            T last = page.get(page.size() - 1);
            page = waterfallProcessInterface.getPage(waterfallProcessInterface.getNextParam(last), size);
        }
        return count;
    }
}