[自研开源] 数据集成之分批传输 v0.7

作者：LIEN

2024-03-26
江苏
本文字数：2303 字
阅读完需：约 8 分钟

介绍

本篇基于数据集成之任务流程介绍任务分批传输的使用场景和配置操作。

使用场景

mydata 使用 API 方式集成数据，当一次请求或响应传输数据量较多时可能无法完成、或容易对服务端造成影响，因此需要分为多次处理；

例如常见的分页查询、导入大量数据时分批处理、集成对接时的全量同步等；

分批传输数据

业务系统与 mydata 集成时，在提供数据和消费数据这两个方向上分别实现分批传输；

提供数据

由 mydata 调用应用的 API 获取数据，通过配置分批参数实现一次任务内多次调用 API 获取完整数据，有以下两种基本的配置模式：

配置了 固定参数size=10、递增参数current 从 1 开始每次递增 1、每次间隔1 秒的任务；

配置了 递增参数start 从 1 开始每次递增 100、递增参数end 从 100 开始每次递增 100、每次间隔1 秒的任务；

执行过程如下代码，要点有：

通过 do-while 结构兼容单次和分批；
用lastProduceData记录上一次数据，用于和本次对比数据，若重复则结束，避免死循环（理论上很少有 2 次完全一样的数据）；
若分批有异常，则复用任务 3 次出错自动结束并发送邮件通知的功能；
执行完一次后，自动计算递增参数值；

// 提供数据case MdConstant.DATA_PRODUCER:    // 分批模式 记录上一次数据，用于对比两次数据，若重复 则结束，避免死循环    List<Map> lastProduceData = null;    do {        // 若启用分批，则将分批参数加入请求参数中        if (taskInfo.isBatch()) {            Map<String, Object> batchParam = jobBatchService.parseToMap(taskInfo);            Map<String, Object> reqParams = MapUtil.union(taskInfo.getReqParams(), batchParam);            taskInfo.setReqParams(reqParams);        }
        // 调用api 获取json        String json = ApiUtil.read(taskInfo);
        // 将json按字段映射 解析为业务数据        jobDataService.parseData(taskInfo, json);        // 若没有返回数据，则结束处理        if (CollUtil.isEmpty(taskInfo.getProduceDataList())) {            break;        }        // 对比上一次数据        if (lastProduceData != null) {            if (CollUtil.isEqualList(lastProduceData, taskInfo.getProduceDataList())) {                // 异常任务失败，邮件通知用户检查任务                throw new RuntimeException("分批获取数据异常，最后两次获取的数据相同！");            }        }        lastProduceData = taskInfo.getProduceDataList();
        // 根据条件过滤数据        jobDataFilterService.doFilter(taskInfo);
        // 保存业务数据        jobDataService.saveTaskData(taskInfo);
        // 更新环境变量        jobVarService.saveVarValue(taskInfo, json);
        // 递增分批参数        jobBatchService.incBatchParam(taskInfo);
        // 若启用分批，则等待间隔        if (taskInfo.isBatch()) {            ThreadUtil.sleep(taskInfo.getBatchInterval(), TimeUnit.SECONDS);        }    } while (taskInfo.isBatch());
    break;

复制代码

消费数据

由 mydata 通过 API 向应用发送数据，通过配置分批参数限制每次向 API 发送的数据量，从而减少数据查询量和请求处理时间；

如下图，配置了分批数量为 1000 的任务，分批参数为选填，mydata 将按 1000 为限制查询符合条件的数据，通过 API 请求发送给应用；

执行过程如下代码，要点有：

通过 do-while 结构兼容单次和分批；
自动管理分页参数，执行分页查询数据，发送给 API；
直到分页查询没有数据自动结束；

// 消费数据case MdConstant.DATA_CONSUMER:    String dataCode = taskInfo.getDataCode();    if (StrUtil.isEmpty(dataCode)) {        break;    }    List<BizDataFilter> filters = taskInfo.getDataFilters();    if (CollUtil.isNotEmpty(filters)) {        // 解析过滤条件值中的 自定义字符串        parseFilterValue(filters);        // 排除值为null的条件        filters = filters.stream().filter(filter -> filter.getValue() != null).collect(Collectors.toList());    }    int round = 0;    Long skip = null;    Integer limit = taskInfo.isBatch() ? taskInfo.getBatchSize() : null;    do {        if (taskInfo.isBatch()) {            skip = (long) round * taskInfo.getBatchSize();        }        // 根据过滤条件 查询数据        List<Map> dataList = bizDataDAO.list(MdUtil.getBizDbCode(taskInfo.getTenantId(), taskInfo.getProjectId(), taskInfo.getEnvId()), dataCode, filters, skip, limit);        if (CollUtil.isEmpty(dataList)) {            break;        }        taskInfo.setConsumeDataList(dataList);        // 根据字段映射转换为api参数        jobDataService.convertData(taskInfo);        // 调用api传输数据        ApiUtil.write(taskInfo);
        round++;        // 若启用分批，则等待间隔        if (taskInfo.isBatch()) {            ThreadUtil.sleep(taskInfo.getBatchInterval(), TimeUnit.SECONDS);        }    }    while (taskInfo.isBatch());    break;

复制代码

发布于: 刚刚阅读数: 4

原文链接:【http://xie.infoq.cn/article/6fce8373aa610d11787dc35d6】。文章转载请联系作者。

LIEN

关注

还未添加个人签名 2020-04-25 加入

还未添加个人简介

发布

暂无评论

创作场景