ElasticSearch 通用化批量数据同步器封装设计

一、设计需要满足的功能场景
二、约定
三、问题
四、功能封装
- 1、接口函数定义
- 2、常量定义
- 3、多线程分片处理器（Fork/Join 框架任务切分与处理）
- 4、分页分片接口
- 5、分页分片实现

一、设计需要满足的功能场景

ES 现在也是很多公司会用到的工具了，用来做数据的快速查询，那么不可避免的就会遇到需要将数据同步到 ES 的情况，为了做封装我们需要考虑到几个方面的问题。

数据量：数据同步那么自然涉及到同步的数据量问题。通常的做法是把所有的数据查询到缓存中，然后再依次丢到 ES 里。但是假如数据量极大的情况就会把内存撑爆。因此设计上我们需要考虑分页分片同步，每次查个500条就同步500条，用时间换空间。
效率问题：既然分页分片，每个分片任务是互不影响的，因此可以考虑用多线程来提高处理效率。
通用性：因为需要考虑通用性，因此分页接口需要调用方传入，作为规范约定，分页返回的数据必须与ES字段一致

二、约定

规范约定，分页返回的数据必须与ES字段一致，目的是为了保证数据的正确性

至于存储工具的封装，参考：ElasticSearch 通用化批量存储封装设计

三、问题

1.如何知道分页是否全部处理完成了？

这里采用一种假设法，假设我们一页500条数据，每批5个线程处理，那么我们每批的线程共会处理线程 500*5 = 2500 条数据，假如有 10001 条数据要处理，那么之前的 4 批，每次都会处理满 2500 条，仅最后一次只处理 1 条数据。因此当我们发现处理数据的数量无法达到最大值时，说明数据处理完。

2.我们如何做到每批 5 个线程，每个线程处理 500 条这种操作？并且还要统计没批处理的数据量？

我们可以使用 Fork/Join 框架来实现

四、功能封装 1、接口函数定义

分页接口函数：用来传分页查询接口，PageResultDTO 自定义的分页返回值，Pageable 分页参数

@FunctionalInterface
public interface PageFunction {
    PageResultDTO page(Pageable pageable);
}

ES操作接口函数： ES 操作接口函数

@FunctionalInterface
public interface EsOperationFunction {
    void handle(List data);
}

2、常量定义

public class PageShardingConstants {
    
    public static final int MAX_SHARDING_NUM = 10;

    
    public static final int MAX_SHARDING_SIZE = 1000;

    
    public static final int DEFAULT_SHARDING_NUM = 4;

    
    public static final int DEFAULT_SHARDING_SIZE = 500;
}

3、多线程分片处理器（Fork/Join 框架任务切分与处理）

我们用 Fork/Join 框架来作为分片处理器，Fork/Join 框架的好处在于在多线程处理中，可以帮我们实工作密取，空闲的线程会从繁忙线程的任务队列中，从任务队列后面获取一个任务去执行，以保证线程的利用率。

处理器的返回值为处理的数据总条数

public class PageShardingHandle extends RecursiveTask {

    private static final long serialVersionUID = 416664792640153410L;

    
    private boolean cutTask;

    
    private Integer page;

    
    private Integer size;

    
    private Integer remainShardNum;

    
    private PageFunction pageFunction;

    
    private EsOperationFunction esOperationFunction;


    public PageShardingHandle(Integer page, Integer size, Integer remainShardNum, PageFunction pageFunction,
        EsOperationFunction esOperationFunction) {
        this.cutTask = Boolean.FALSE;
        this.page = page;
        this.size = size;
        this.remainShardNum = remainShardNum;
        this.pageFunction = pageFunction;
        this.esOperationFunction = esOperationFunction;
    }

    public PageShardingHandle(boolean cutTask, Integer page, Integer size, Integer remainShardNum,
        PageFunction pageFunction, EsOperationFunction esOperationFunction) {
        this.page = page;
        this.size = size;
        this.remainShardNum = remainShardNum;
        this.cutTask = cutTask;
        this.pageFunction = pageFunction;
        this.esOperationFunction = esOperationFunction;
    }

    @Override
    protected Long compute() {
        // 总条数
        Long total = 0L;
        // 已经任务进行分片，处理任务
        if (this.cutTask) {
            // 查询分页的数据（接口函数的处理）
            PageResultDTO result = this.pageFunction.page(BasePageable.of(this.page, this.size));
            // ES处理
            if (result.getTotal() > 0) {
                this.esOperationFunction.handle(result.getResult());
            }
            total = result.getTotal();
        } else {
            // 未分片，需要进行分片
            List handles = new ArrayList<>();
            while (this.remainShardNum > 0) {
                // page：传递下一页
                // remainShardNum：剩余分片数量减少
                // cutTask：任务为已分片任务
                handles.add(new PageShardingHandle(Boolean.TRUE, this.page++, this.size, --this.remainShardNum,
                    this.pageFunction, this.esOperationFunction, this.context));
                // 提交分片任务
                invokeAll(handles);
            }
            // 接收处理结果返回值，结果中汇总总条数
            for (PageShardingHandle handle : handles) {
                total += handle.join();
            }
        }
        return total;
    }
}

4、分页分片接口

public interface PageShardingApi {
    
    void syncEsByTotal(Integer total, PageFunction pageFunction, EsIndexEnum esIndexEnum);

    
    void syncEsByTotal(Integer size, Integer total, PageFunction pageFunction, EsIndexEnum esIndexEnum);

    
    void syncEs(PageFunction pageFunction, EsIndexEnum esIndexEnum);

    
    void syncEs(Integer size, PageFunction pageFunction, EsIndexEnum esIndexEnum);

    
    void syncEs(Integer size, Integer shardNum, PageFunction pageFunction, EsIndexEnum esIndexEnum);
}

5、分页分片实现

EsIndexEnum 与 EsOperationApi 参考：ElasticSearch 通用化批量存储封装设计

@Service
public class PageShardingApiImpl implements PageShardingApi {

    @Resource
    private EsOperationApi esOperationApi;

    @Override
    public void syncEsByTotal(Integer total, PageFunction pageFunction, EsIndexEnum esIndexEnum) {
        this.syncEsByTotal(null, total, pageFunction, esIndexEnum);
    }

    @Override
    public void syncEsByTotal(Integer size, Integer total, PageFunction pageFunction, EsIndexEnum esIndexEnum) {
        // 如果发现没有总条数或者总条数为 0，不进行后续操作
        if (total == null || total == 0) {
            return;
        }
        if (size == null) {
            size = PageShardingConstants.DEFAULT_SHARDING_SIZE;
        }
        Integer shardNum = total / size + (total % size == 0 ? 0 : 1);
        this.syncEs(size, shardNum, pageFunction, esIndexEnum);
    }

    @Override
    public void syncEs(PageFunction pageFunction, EsIndexEnum esIndexEnum) {
        this.syncEs(null, null, pageFunction, esIndexEnum);
    }

    @Override
    public void syncEs(Integer size, PageFunction pageFunction, EsIndexEnum esIndexEnum) {
        this.syncEs(size, null, pageFunction, esIndexEnum);
    }

    @Override
    public void syncEs(Integer size, Integer shardNum, PageFunction pageFunction, EsIndexEnum esIndexEnum) {
        // 分片的数量不能大于最大数量
        if (shardNum == null) {
            shardNum = PageShardingConstants.DEFAULT_SHARDING_NUM;
        }
        if (size == null) {
            size = PageShardingConstants.DEFAULT_SHARDING_SIZE;
        }
        if (shardNum > PageShardingConstants.MAX_SHARDING_NUM) {
            shardNum = PageShardingConstants.MAX_SHARDING_NUM;
        }
        if (size > PageShardingConstants.MAX_SHARDING_SIZE) {
            size = PageShardingConstants.MAX_SHARDING_SIZE;
        }
        // 每次分片的最大查询到的数量：分片数量 * 每个分片的条数
        Integer maxPageSize = size * shardNum;

        ForkJoinPool forkJoinPool = new ForkJoinPool(shardNum);
        ForkJoinTask futureTask;
        for (int i = 1;; i++) {
            futureTask = forkJoinPool
                .submit(new PageShardingHandle(i, size, shardNum, pageFunction,
                    (data) -> this.esOperationApi.save(data, esIndexEnum),
                    RecruitContext.getAll()));
            // 如果此处分页处理的数据没有达到最大数量，说明分页见底
            if (futureTask.join() < maxPageSize) {
                break;
            }
        }
    }
}

ElasticSearch 通用化批量数据同步器封装设计

Java相关栏目本月热门文章