public FileLoadContext load(FileBatch fileBatch, File rootDir, WeightController controller) {
        if (false == rootDir.exists()) {
            throw new LoadException(rootDir.getPath() + " is not exist");
        }
        FileLoadContext context = buildContext(fileBatch.getIdentity());
        context.setPrepareDatas(fileBatch.getFiles());
        boolean isDryRun = context.getPipeline().getParameters().isDryRun();
        try {
            // 复制成功的文件信息
            WeightBuckets<FileData> buckets = buildWeightBuckets(fileBatch.getIdentity(), fileBatch.getFiles());
            List<Long> weights = buckets.weights();
            controller.start(weights);
            // 处理数据
            for (int i = 0; i < weights.size(); i++) {
                Long weight = weights.get(i);
                controller.await(weight.intValue());
                if (logger.isInfoEnabled()) {
                    logger.debug("##start load for weight:{}\n", weight);
                }

                // 处理同一个weight下的数据
                List<FileData> items = buckets.getItems(weight);
                if (context.getPipeline().getParameters().isDryRun()) {
                    dryRun(context, items, rootDir);
                } else {
                    moveFiles(context, items, rootDir);
                }

                controller.single(weight.intValue());
                if (logger.isInfoEnabled()) {
                    logger.debug("##end load for weight:{}\n", weight);
                }
            }

            if (dump || isDryRun) {
                MDC.put(OtterConstants.splitPipelineLoadLogFileKey,
                        String.valueOf(fileBatch.getIdentity().getPipelineId()));
                logger.info(FileloadDumper.dumpContext("successed", context));
                MDC.remove(OtterConstants.splitPipelineLoadLogFileKey);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            if (dump || isDryRun) {
                MDC.put(OtterConstants.splitPipelineLoadLogFileKey,
                        String.valueOf(fileBatch.getIdentity().getPipelineId()));
                logger.info(FileloadDumper.dumpContext("error", context));
                MDC.remove(OtterConstants.splitPipelineLoadLogFileKey);
            }
        } catch (Exception e) {
            if (dump || isDryRun) {
                MDC.put(OtterConstants.splitPipelineLoadLogFileKey,
                        String.valueOf(fileBatch.getIdentity().getPipelineId()));
                logger.info(FileloadDumper.dumpContext("error", context));
                MDC.remove(OtterConstants.splitPipelineLoadLogFileKey);
            }
            throw new LoadException(e);
        } finally {

            // 不论是否移动成功，删除临时目录
            NioUtils.delete(rootDir, 3);
        }

        return context;
    }
public FileLoadContext load ( FileBatch fileBatch , File rootDir , WeightController controller ) { if ( false == rootDir . exists ( ) ) { throw new LoadException ( rootDir . getPath ( ) + " is not exist" ) ; } FileLoadContext context = buildContext ( fileBatch . getIdentity ( ) ) ; context . setPrepareDatas ( fileBatch . getFiles ( ) ) ; boolean isDryRun = context . getPipeline ( ) . getParameters ( ) . isDryRun ( ) ; try { WeightBuckets < FileData > buckets = buildWeightBuckets ( fileBatch . getIdentity ( ) , fileBatch . getFiles ( ) ) ; List < Long > weights = buckets . weights ( ) ; controller . start ( weights ) ; for ( int i = 0 ; i < weights . size ( ) ; i ++ ) { Long weight = weights . get ( i ) ; controller . await ( weight . intValue ( ) ) ; if ( logger . isInfoEnabled ( ) ) { logger . debug ( "##start load for weight:{}\n" , weight ) ; } List < FileData > items = buckets . getItems ( weight ) ; if ( context . getPipeline ( ) . getParameters ( ) . isDryRun ( ) ) { dryRun ( context , items , rootDir ) ; } else { moveFiles ( context , items , rootDir ) ; } controller . single ( weight . intValue ( ) ) ; if ( logger . isInfoEnabled ( ) ) { logger . debug ( "##end load for weight:{}\n" , weight ) ; } } if ( dump || isDryRun ) { MDC . put ( OtterConstants . splitPipelineLoadLogFileKey , String . valueOf ( fileBatch . getIdentity ( ) . getPipelineId ( ) ) ) ; logger . info ( FileloadDumper . dumpContext ( "successed" , context ) ) ; MDC . remove ( OtterConstants . splitPipelineLoadLogFileKey ) ; } } catch ( InterruptedException e ) { Thread . currentThread ( ) . interrupt ( ) ; if ( dump || isDryRun ) { MDC . put ( OtterConstants . splitPipelineLoadLogFileKey , String . valueOf ( fileBatch . getIdentity ( ) . getPipelineId ( ) ) ) ; logger . info ( FileloadDumper . dumpContext ( "error" , context ) ) ; MDC . remove ( OtterConstants . splitPipelineLoadLogFileKey ) ; } } catch ( Exception e ) { if ( dump || isDryRun ) { MDC . put ( OtterConstants . splitPipelineLoadLogFileKey , String . valueOf ( fileBatch . getIdentity ( ) . getPipelineId ( ) ) ) ; logger . info ( FileloadDumper . dumpContext ( "error" , context ) ) ; MDC . remove ( OtterConstants . splitPipelineLoadLogFileKey ) ; } throw new LoadException ( e ) ; } finally { NioUtils . delete ( rootDir , 3 ) ; } return context ; }
返回结果为已处理成功的记录