采集数据的任务-数据的保存

在上一章中讲解了项目中责任链的数据的计算节点的执行过程，从本章节开始会详细讲解责任链中采集节点的下一个节点：数据的保存节点执行逻辑

计算节点

org.javaup.chain.impl.SaveDataHandlerChainHandler

@Component
public class SaveDataHandlerChainHandler extends AbstractDataChainHandler {
    
    @Autowired
    private SaveDataHandler saveDataHandler;
    
    @Override
    protected void handler(final TotalParamTransfers totalParamTransfers) {
        saveDataHandler.saveHandle(totalParamTransfers);
    }
    
    @Override
    protected Integer order() {
        return 3;
    }
}

计算节点中逻辑都交给了数据保存处理器SaveDataHandler来处理

数据保存处理器

org.javaup.handler.core.save.SaveDataHandler#saveHandle

/**
 * 保存数据处理
 * @param totalParamTransfers 总参数传输对象
 */
public void saveHandle(TotalParamTransfers totalParamTransfers){
    List<Map<String, Object>> resultDataList = totalParamTransfers.getResultDataList();
    if (CollectionUtil.isEmpty(resultDataList)) {
        return;
    }
    //利用规则id查询数据保存配置信息
    DataSave dataSave = dataSaveService.getDataSaveByRuleId(totalParamTransfers.getParamTransfers().getRuleId());
    if (Objects.isNull(dataSave)) {
        return;
    }
    //查询规则
    Rule rule = ruleService.getById(totalParamTransfers.getParamTransfers().getRuleId());
    if (Objects.isNull(rule)) {
        return;
    }
    Long parentVideoTypeId = totalParamTransfers.getParamTransfers().getDimensionTransfers().getParentVideoTypeId();
    Long videoTypeId = totalParamTransfers.getParamTransfers().getDimensionTransfers().getVideoTypeId();
    if (Objects.nonNull(parentVideoTypeId) && Objects.nonNull(videoTypeId)) {
        redisCache.addForSet(RedisKeyBuild.createRedisKey(RedisKeyManage.VIDEO_TYPE,parentVideoTypeId),videoTypeId);
    }

    //要保存的数据源名称
    String dataSourceName = dataSave.getDataSourceName();
    //要保存的表前缀
    String createTablePrefix = dataSave.getCreateTablePrefix();
    //统计的表名,以业务维度是video为例：
    //如果业务维度是video，时间维度是day，那么表名：d_report_video_data_video_stats
    //如果业务维度是video，时间维度不是day，那么表名：d_report_video_data_video_stats_range
    String tableName = DataSourceFunc.createTableName(createTablePrefix, rule.getRuleName(), 
            totalParamTransfers.getParamTransfers().getVideoDimensionType(), 
            totalParamTransfers.getParamTransfers().getRequestTime().getDateType());


    //删除表数据语句
    String deleteTableDataSql = null;
    //删除参数
    Map<String, Object> params = null;
    if (centerProperties.getCenterDeleteStats().equals(CenterProperties.EXECUTE_CENTER_DELETE_STATS)) {
        deleteTableDataSql = DataSourceFunc.createDeleteTableDataSql(tableName,
                totalParamTransfers.getParamTransfers().getVideoDimensionType());
        params = totalParamTransfers.assemblyParams();
    }
    //插入表数据语句
    String insertTableDataSql = DataSourceFunc.createInsertTableDataSql(tableName, resultDataList);
    //执行删除和插入操作
    agilityDataHandler.execute(dataSourceName, deleteTableDataSql, insertTableDataSql, params, resultDataList);
}

执行流程

1. 读取结果集并进行验证

List<Map<String, Object>> resultDataList = totalParamTransfers.getResultDataList();
if (CollectionUtil.isEmpty(resultDataList)) {
    return;
}

• 从 totalParamTransfers 取出 resultDataList，如果为空则直接返回，整个保存流程不继续。

2. 读取保存配置与规则并校验

//利用规则id查询数据保存配置信息
DataSave dataSave = dataSaveService.getDataSaveByRuleId(totalParamTransfers.getParamTransfers().getRuleId());
if (Objects.isNull(dataSave)) {
    return;
}
//查询规则
Rule rule = ruleService.getById(totalParamTransfers.getParamTransfers().getRuleId());
if (Objects.isNull(rule)) {
    return;
}

• 通过规则 ID 查询 DataSave（拿到目标数据源名和表前缀）。为空直接返回。

• 通过规则 ID 查询 Rule（至少需要 ruleName 参与表名拼接）。为空直接返回。

• 配置结构：

@TableName("d_data_save")
public class DataSave extends BaseTableData implements Serializable {
    @TableId(value = "id", type = IdType.AUTO)
    private Long id;
    private Long ruleId;
    private String dataSourceName;        // 数据源名称
    private String createTablePrefix;     // 表前缀
}

3. 维护 Redis 中的分类关系

Long parentVideoTypeId = totalParamTransfers.getParamTransfers().getDimensionTransfers().getParentVideoTypeId();
Long videoTypeId = totalParamTransfers.getParamTransfers().getDimensionTransfers().getVideoTypeId();
if (Objects.nonNull(parentVideoTypeId) && Objects.nonNull(videoTypeId)) {
    redisCache.addForSet(RedisKeyBuild.createRedisKey(RedisKeyManage.VIDEO_TYPE,parentVideoTypeId),videoTypeId);
}

若 parentVideoTypeId 与 videoTypeId 均非空，则将 videoTypeId 添加到以 parentVideoTypeId 为命名的的集合键中（用于维度层级关系维护）。

4. 计算目标统计表名

//要保存的数据源名称
String dataSourceName = dataSave.getDataSourceName();
//要保存的表前缀
String createTablePrefix = dataSave.getCreateTablePrefix();
//统计的表名,以业务维度是video为例：
//如果业务维度是video，时间维度是day，那么表名：d_report_video_data_video_stats
//如果业务维度是video，时间维度不是day，那么表名：d_report_video_data_video_stats_range
String tableName = DataSourceFunc.createTableName(createTablePrefix, rule.getRuleName(), 
        totalParamTransfers.getParamTransfers().getVideoDimensionType(), 
        totalParamTransfers.getParamTransfers().getRequestTime().getDateType());

• 表名来源： 保存配置的前缀 + 规则英文名 + 下划线 + 业务维度值 + 下划线 + 时间分区后缀。
• 规则如下： 若时间维度是 DAY，使用 stats；否则使用 stats_range。

👉 点击解锁全部付费内容