📚 课程目录第3章子工作流定义与调用

子工作流定义与调用：把重复逻辑装进“黑盒子”

说实话，我最早用 Nextflow 写流程时，也干过把同一段代码复制粘贴好几遍的蠢事。那时候觉得“不就几行嘛，改改参数就好”。结果后来需求一变更，我得把每个副本都改一遍——改漏一个就出 bug，排查起来特别痛苦。

后来我才意识到，Nextflow 早就给了我们一个利器：子工作流。说白了，就是把一段可复用的流程逻辑封装成一个“黑盒子”，然后在主流程里反复调用。你传进去不同的输入，它给你吐出对应的结果，内部怎么折腾你都不用管。

子工作流到底是什么？

子工作流是用 workflow 关键字定义的一段独立流程。它有自己的入口参数、执行逻辑和输出结果。你可以把它理解成编程里的“函数”——只不过这个函数里跑的是流程而不是普通代码。

举个例子，我在一个 RNA-seq 项目里，需要频繁对样本做质控。质控的步骤基本一样：跑 FastQC、汇总 MultiQC、输出报告。之前我就在每个样本的处理流程里都写一遍这些步骤，后来用子工作流一封装，清爽多了。

核心思路： 子工作流 = 可复用的流程模块。参数驱动，输入输出明确，内部实现对外透明。

如何定义子工作流？

定义子工作流的语法其实和写主流程差不多。你用 workflow 关键字给它起个名字，然后在花括号里写它要执行的步骤。关键是要声明清楚输入参数和输出结果。

我个人习惯把子工作流单独放在一个模块文件里，比如 modules/quality_control.nf，这样主流程看起来更干净。

看一个完整的定义例子：

// modules/quality_control.nf

workflow quality_control {
    take:
    reads_ch  // 输入的 reads 通道

    main:
    FASTQC(reads_ch)
    MULTIQC(FASTQC.out.zip.collect())
    emit:
    multiqc_report = MULTIQC.out.report
}

这里有几个要点你得注意：

take: 声明子工作流需要哪些输入。可以是通道，也可以是普通值。
main: 是流程执行的主体。你在这里调用进程、连接通道。
emit: 定义子工作流的输出。外面调用的时候，通过这些名字来获取结果。

小技巧： 子工作流的 take 和 emit 就是它的“接口”。接口设计得越清晰，别人用起来越顺手。我一般会在 emit 里把所有重要的中间结果都暴露出来，方便主流程灵活使用。

在主流程中调用子工作流

定义好了子工作流，怎么在主线里调用它呢？很简单——就像调用一个函数一样。你需要先 include 它，然后传参、收结果。

看一个完整的主流程例子：

// main.nf
include { quality_control } from './modules/quality_control.nf'

workflow {
    // 准备输入数据
    read_pairs_ch = Channel.fromFilePairs(params.reads)

    // 调用子工作流，只传 reads 部分
    qc_results = quality_control(read_pairs_ch)

    // 使用子工作流的输出
    qc_results.multiqc_report.view()
}

调用的时候，子工作流的 take 对应你传入的参数，emit 里的名字作为属性挂在返回对象上。这个设计很直观——你传进去啥，它返回啥，清清楚楚。

避坑指南： 我曾经犯过一个低级错误——子工作流只定义了一个输出，但我在主流程里用了另一个名字去取。结果流程跑完啥也没拿到。记住，emit 里的名字就是你在主流程里访问的“钥匙”，两个地方必须严格匹配。

子工作流的嵌套与复用

子工作流之间也能互相调用。你可以把一个子工作流当作另一个子工作流的组建块。这种嵌套设计在复杂项目中特别有用。

比如我做过一个项目，先有一个基础的“样本处理”子工作流，然后在它之上构建了“肿瘤-正常配对分析”子工作流。基础子工作流只管比对、排序、标记重复；上层子工作流调用它，再额外做变异检测和注释。

看一个嵌套调用的代码骨架：

// modules/sample_processing.nf
workflow sample_processing {
    take:
    sample_ch

    main:
    BWA_MEM(sample_ch)
    SORT_SAM(BWA_MEM.out.bam)
    MARK_DUPS(SORT_SAM.out.sorted_bam)

    emit:
    dedup_bam = MARK_DUPS.out.bam
}

// modules/tumor_normal.nf
include { sample_processing } from './sample_processing.nf'

workflow tumor_normal_analysis {
    take:
    tumor_ch
    normal_ch

    main:
    tumor_result = sample_processing(tumor_ch)
    normal_result = sample_processing(normal_ch)

    // 然后执行配对分析
    MUTECT2(tumor_result.dedup_bam, normal_result.dedup_bam)

    emit:
    vcf = MUTECT2.out.vcf
}

这种嵌套结构会让你的流程层次特别清晰。每个子工作流只关心自己的责任，上层只关心怎么组合结果。

子工作流的最佳实践

用到现在，我总结了几条经验，分享给你：

实践要点	具体做法
模块化文件	每个子工作流单独放到 `modules/` 目录下的一个文件中
接口最小化	`take` 只放必要参数，`emit` 暴露需要的输出，不要多给
命名清晰	子工作流名称用动词开头，如 `run_qc`、`process_sample`
参数传递	尽量用通道传数据，少用全局变量，保证隔离性
错误处理	子工作流内部可以用 `errorStrategy` 控制失败行为，不要在外部依赖它

注意： 子工作流内部定义的进程或变量，默认不会污染主流程的命名空间。这是好事，但如果你不小心在子工作流里用了 params 全局变量，那就又打破了隔离。我建议：子工作流里尽量少用 params，数据全走 take 传进来。

一张图理解子工作流机制

下面这张图展示了子工作流定义、调用的完整数据流。核心就是：主流程准备数据→调用子工作流→子工作流内部自己跑→返回结果给主流程。

一些坑和我的建议

嗯，这里要提醒你几个容易踩的坑：

通道复用问题：同一个通道如果被多个子工作流消费，记得用 .clone() 或者 .buffer() 复制。我当初没注意，结果第二个子工作流接到的数据是空的……排查了半天。
子工作流内的进程名冲突：不同子工作流里如果用相同名字的进程，Nextflow 会报错。我建议给进程名加前缀或者用模块命名空间隔离。
过度嵌套：虽然子工作流可以无限嵌套，但超过三层之后，调试起来就头疼了。保持扁平化，每个子工作流职责单一。

子工作流是 Nextflow 里性价比最高的功能之一。你花半小时封装好，后面每次调用都能省下好几倍的重复劳动。而且代码可读性、可维护性直接提升一个档次。