batchLoader: 一个轻量级的数据分片组件

BatchLoader

一、简介

BatchLoader是一个轻量级的数据分片组件，针对大数据量的场景下，它提供了纵向与横向的分片能力，可批量、匀速的从数据源（MySQL、ES、RPC接口等）中查询并消费数据。该组件提供了游标模式（BatchLoader）与翻页模式（BatchPageLoader），前者更适用于解决深分页场景而后者则更多的用于分页RPC调用。组件还提供了横向分片模式（BatchLoaderGroupWrap），该模式适合分库分表场景下批量查询数据。组件设计之初是希望用户在批次查询数据后可以及时对数据进行消费以避免占用过多内存导致OOM，但组件也为用户提供了BatchLoaderResultWrap模式，可以返回全量的查询结果。

BatchLoader遵循了”约定大于配置，组合优于继承“的设计思想，采用了简单建造者、装饰器等设计模式进行开发，无任何外部依赖，轻量便捷、灵活易用，是循环数据读取场景下简化代码的优秀选择！

二、安装

1、下载到本地并安装

#下载项目到本地
git clone https://gitee.com/GiteeOrorda/batch-loader.git
#安装mvn依赖到本地仓库
cd batch-loader
mvn clean
mvn install  -Dmaven.test.skip=true

2、添加maven依赖

    <dependency>
        <groupId>com.lxd</groupId>
        <artifactId>batchLoader</artifactId>
        <version>1.0-SNAPSHOT</version>
    </dependency>

三、使用示例

BatchLoader示例

示例1：假设数据库中存在一张数据量较大的plan表，此时使用BatchLoader对该表进行全量遍历

*注：游标需确保唯一且有序

    @Test
    void testBatchLoader() {
        PlanCondition condition = new PlanCondition();
        BatchLoaderBase batchLoaderBase = 
            	 new BatchLoader<PlanCondition, PlanEntity,Long>() //声明BatchLoader
                .buildCondition(condition,new BatchInfo<>(6000), 50000)//构建查询条件
                .buildProcess((c,b) -> planDao.getPlanByCondtion(c, b)) //声明查询逻辑
            	//查询结果消费策略（本段代码仅打印每批查询结果的长度）
                .buildConsumer(ls -> System.out.println("批量查询到数据长度：" + ls.size()))
                .buildCursor(PlanEntity::getId)//寻找游标的方法
            	.ctrlRate(1000) //以毫秒为单位控制消费速率，这里意味着每1000ms执行一次查询
                .isCheckMaxCount(true);//是否校验数据上限
        //执行
        batchLoaderBase.doConsumer();
        System.out.println("查询总数：" + batchLoaderBase.getCurTotalCount());

    }

BatchLoader对象API介绍

1、buildCondition

构建查询入参、批次规则、最大数据，固定调用

参数	类型	是否必填	备注
condition	C	是	该对象是查询数据的入参
batchInfo	BatchInfo	否	该对象存在batch与cursor两个属性，前者控制每批数据查询的范围；后者控制游标的起始值，该类型与构建BatchLoader时的第三个泛型一致。如果不做传参程序默认batch为10000，cursor为null
maxCount	int	否	为防止程序过多循环并避免oom，用户可以设置查询数据的总上限，默认上限为1,000,000

2、buildProcess

该方法用于构建数据查询策略，固定调用

参数	类型	是否必填	备注
findStrategy	BiFunction<C, BatchInfo, List>	是

3、buildConsumer

该方法用于构建消费策略，每当查询出数据集合后，程序会调用该方法对查询出的结果集进行消费，固定调用

参数	类型	是否必填	备注
consumer	Consumer<List>	是

4、buildCursor

游标构建工厂，在每批数据查询结束后会调用该方法获取本次游标的终止值，固定调用

参数	类型	是否必填	备注
cursorFactory	Function<R, T>	是

5、ctrlRate

该方法用于控制消费速率，以避免程序执行速度过快导致资源压力过大，不固定调用，默认不控制查询以及消费速率

参数	类型	是否必填	备注
rate	long	是

6、isCheckMaxCount

是否校验数据查询上限，不固定调用，默认为true

参数	类型	是否必填	备注
isCheck	boolean	是

7、doConsumer

调用该方法执行组件，固定调用，无入参

8、getCurTotalCount

执行doConsumer后调用该方法可查询数据总量，非固定调用，无入参

BatchLoaderGroupWrap示例

示例1：分表查询

使用BatchLoaderGroupWrap对plan0~plan9这十张表进行全量遍历

    @Test
    void testBatchLoaderGroupForSharding() {
        PlanCondition condition = new PlanCondition();
        BatchLoaderGroupWrap<PlanCondition, PlanEntity, Long> planGroupWrap = new BatchLoaderGroupWrap<PlanCondition, PlanEntity, Long>()
                .buildCondition(condition,new BatchInfo<>(6000))
                .buildProcess((c, b) -> planDao.getPlanShardingByCondtion(c, b))
                .buildConsumer(ls -> System.out.println("批量查询到数据长度：" + ls.size()))
                .buildCursor(PlanEntity::getId)
                .isCheckMaxCount(false)
                .buildGroupWrap();
        List<String> tables = IntStream.range(0, 10).boxed().map(t -> SHARD_TABLE + t).collect(Collectors.toList());
        planGroupWrap.doConsumerByGroup(tables, (ls,con) -> con.setTableName(ls.get(0)), 1);
        int curTotalCount = planGroupWrap.getCurTotalCount();
        System.out.println(curTotalCount);
    }

示例2：分片查询，解决in子句索引区分度较低问题

假设plan表中存在varchar类型的week字段，该字段存储了从第1周~~第1000周的字符串，当查询第1周~~第600周的数据时我们会因为in后的条件过多导致索引区分度较低进而影响查询性能。此时我们就可以使用BatchLoaderGroupWrap对查询条件进行分片，批量查询

    @Test
    void testBatchLoaderGroup() {
        List<String> weekList = IntStream.range(1, 601)
            .mapToObj(p -> "第#周".replace("#",String.valueOf(p)))
            .collect(Collectors.toList());
        
        PlanCondition condition = new PlanCondition();
        condition.setWeekList(weekList);
        BatchLoaderGroupWrap<PlanCondition, PlanEntity, Long> planGroupWrap = 
            new BatchLoaderGroupWrap<PlanCondition, PlanEntity, Long>()
                .buildCondition(condition)
                .buildProcess((c, b) -> planDao.getPlanByCondtion(c, b))
                .buildConsumer(ls -> System.out.println("批量查询到数据长度：" + ls.size()))
                .buildCursor(PlanEntity::getId)
                .isCheckMaxCount(true)
                .buildGroupWrap();//调用BatchLoaderGroupWrap装饰器的固定写法

        planGroupWrap.doConsumerByGroup(weekList, (ls,con) -> con.setWeekList(ls), 20);
        int curTotalCount = planGroupWrap.getCurTotalCount();

        System.out.println(curTotalCount);
    }

BatchLoaderGroupWrap对象API介绍

buildCondition、buildProcess、buildConsumer、buildCursor、ctrlRate、getCurTotalCount方法参考BatchLoader对象用法

1、buildGroupWrap

该方法为BatchLoaderGroupWrap构建固定写法，无入参，固定调用。

2、doConsumerByGroup

构建分片查询集合数据并执行分片查询，固定调用

参数	类型	是否必填	备注
list	List	是	需要分片的集合数据
conditionChangeStrategy	BiConsumer<List, C>	是	构建分片查询条件的策略
groupCount	int	否	每个分片的元素个数，默认值为50

*注：上文示例以及SQL脚本请参考BatchLoader案例项目loaderDemo：

https://gitee.com/GiteeOrorda/loader-demo

ororda / batchLoader

BatchLoader

一、简介

二、安装

三、使用示例

BatchLoader示例

BatchLoaderGroupWrap示例

简介

发行版

贡献者

近期动态

ororda / batchLoader .gitee-modal { width: 500px !important; }

BatchLoader

一、简介

二、安装

三、使用示例

BatchLoader示例

BatchLoaderGroupWrap示例

简介

发行版

开源评估指数源自 OSS-Compass 评估体系，评估体系围绕以下三个维度对项目展开评估：

贡献者

近期动态

搜索帮助

ororda / batchLoader