第四章节：流程核心元素——Process、Channel、Workflow的语法与实例

上一课我们把 Nextflow 的环境搭起来了。现在进入正题——流程的 三大核心元素：Process、Channel 和 Workflow。

说实话，我刚开始接触 Nextflow 时，觉得这三个概念有点像搭积木。Process 是积木块，Channel 是连接线，而 Workflow 就是那张搭建图纸。今天我们就把这套积木彻底拆开看看。嗯，我会尽量让每个概念都配上一个实战案例，这样你跟着写一遍，基本就掌握了。

1. Process：最小的执行单元

Process 封装了一段脚本或命令。它接收输入，执行计算，输出结果。说白了，它就是 一个可重复的执行环境。为什么强调“可重复”？因为我在项目中遇到过一个坑，写脚本时依赖了全局环境的某个工具版本，结果换台机器就崩了。而 Process 的 conda 或 container 指令就是专门解决这个问题的。

看一个最简单的例子：

process sayHello {
    input:
    val x

    output:
    stdout

    script:
    """
    echo "Hello $x!"
    """
}

每个 Process 有 4 个关键块：directives（可选）、input、output、script。其中 script 块里写的是 可执行的 shell 命令，变量用 $ 引用。

⚠️ 注意：script 块里的变量如果来自 Groovy（比如 $x），直接写就行了。如果想让 shell 本地的 $PATH 被识别，需要写成 \$PATH 或者用单引号包裹。

再来看一个实战一点儿的例子——用 STAR 做比对：

process STAR_ALIGN {
    tag { sample_id }
    publishDir "${params.outdir}/star", mode: 'copy'

    input:
    tuple val(sample_id), path(reads)

    output:
    tuple val(sample_id), path("${sample_id}.bam")

    script:
    """
    STAR --genomeDir ${genome_index} \
         --readFilesIn ${reads} \
         --outFileNamePrefix ${sample_id}_ \
         --runThreadN ${task.cpus}
    """
}

看到 tag 和 publishDir 了吗？这些是指令（directives）。我这里用 tag 给每个任务打上样本名，这样日志里一眼就能定位到哪个样本出了问题。这种小习惯，我建议你从一开始就养成。

💡 我个人的习惯是：每个 Process 除了 input/output，至少写上 tag 和 publishDir。前者让你能追踪，后者让审计时有结果可以查。虽然不写也能跑，但是“能用”和“好用”是两码事。

2. Channel：数据流动的管道

Channel 是 Nextflow 的血管。数据通过它在 Processes 之间传送。你可以把 Channel 理解成 一个异步队列，它或发送值、或发送文件。

有两种核心 Channel：value channel 和 queue channel。

类型 创建方式 特点 用途
value channel Channel.value() 永久可用，可被多个 Process 消费 全局参数、参考基因组路径
queue channel Channel.fromPath() 等 一次性消费，每个元素只能被一个 Process 取走 样本列表、测序文件

这个区别，用对了就是顺畅的管道，用错了就是数据饥饿。我曾经在项目的第二周就因为这个差异导致流程卡死——两个 Process 同时读同一个 queue channel，结果第二个 Process 永远等不到数据。

创建 Channel 的常见方式：

// 单值，可被多次使用
chr_ref = Channel.value("chr22.fa")

// 从文件列表创建，一次性消费
read_files = Channel.fromFilePairs("data/SRR*_{1,2}.fastq.gz")

// 直接发射数值
sample_ids = Channel.of("SRR1001", "SRR1002", "SRR1003")

还有 fromSRA——虽然我们下一章才详细介绍，但你可以记住这个名字。它直接从 SRA 数据库获取测序数据并自动下载，对重分析来说非常省事。

这里插一句：用好 Channel 的 组合算符，比如 join、combine、mix，能极大简化流程逻辑。用表格对比一下这三个常用算符：

算符 行为 示例场景
join 按 key 合并两个 Channel 的元素 将样本名与它的 read1/read2 文件进行配对
combine 笛卡尔积，每个元素都组合一次 每个样本都要与所有参考基因组做比对（比如不同数据库）
mix 直接合并成单一流的元素 合并不同来源的样本 ID

有意思的是，Channel 的每个元素在 Pipe 过程中会自动记录一些元信息（比如来源文件路径、发射时间），这些信息正是审计追踪的基础。嗯，这就是为什么我们说 Nextflow 天生具备可审计性。

3. Workflow：编排一切的剧本

Workflow 块定义流程的执行顺序。它就像一个剧本，告诉 Nextflow 什么时候让哪个 Process 上场、数据怎么传。

有两种写法：一种是 隐式调用（在同一个脚本中直接引用 Process 名），一种是 显式 Workflow 语义块。后者是现在推荐的方式：

workflow {
    read_ch = Channel.fromFilePairs(params.reads)
    trimmed_ch = FASTQC_TRIM(read_ch)
    aligned_ch = STAR_ALIGN(trimmed_ch)
    QUANTIFICATION(aligned_ch)
}

这段代码，我相信你一眼就能看懂它的逻辑。为什么？因为 workflow 层抽象了执行细节，暴露的是 数据流与控制流。我个人在给团队做 Code Review 时，大多数时间花在 workflow 层级——只要这里对了，底层 Process 出问题的概率很小。

Workflow 支持的条件判断和参数化：

workflow {
    read_ch = Channel.fromSRA("SRP251580")

    if (params.use_alignment) {
        aligned_ch = ALIGNMENT(read_ch)
        QUANT_ALIGN(aligned_ch)
    } else {
        QUANT_PSEUDO(read_ch)
    }
}

这个例子展示了 语义化分支。你可以根据参数决定走比对路线还是 pseudo-alignment 路线，而外部调用者只需要改 params.use_alignment。这就把一个复杂流程变成了可配置的模块。

🔑 Workflow 的输入参数可以是 Channel，也可以是纯值。官方推荐把流程的“入口数据”统一设计成 Channel，这样便于下游处理、复用和测试。

4. 三者是如何协同的？

我们现在把 Process、Channel、Workflow 放在一起看一个完整的 RNA-seq 片段：

nextflow.enable.dsl=2

params.reads = "data/SRR*_{1,2}.fastq.gz"
params.outdir = "./results"

process FASTQC {
    publishDir "${params.outdir}/fastqc", mode: 'copy'

    input:
    tuple val(sample_id), path(reads)

    output:
    path("*_fastqc.html")

    script:
    """
    fastqc -t ${task.cpus} ${reads}
    """
}

process SALMON_INDEX {
    publishDir "${params.outdir}/salmon_index", mode: 'copy'

    input:
    path transcriptome

    output:
    path("salmon_index")

    script:
    """
    salmon index -t ${transcriptome} -i salmon_index -p ${task.cpus}
    """
}

workflow {
    // Channel 创建
    read_ch = Channel.fromFilePairs(params.reads)

    // Process 调用 → 数据通过 Channel 传送
    FASTQC(read_ch)
}

注意这里隐含了一个重要概念：Process 的依赖关系。Nextflow 不需要你显式写 after 或 wait，它通过 Channel 的输入输出 自动推断出执行顺序。这就是声明式编程的魔力。

我们绘制一张图来直观理解这三者的协作关系：

流程核心元素：Process・Channel・Workflow 协作架构 Workflow（编排层） 定义流程图 控制分支与参数 自动推断执行顺序 审计追踪根节点 Channel（数据管道） value / queue 类型 fromFilePairs / fromSRA join / combine / mix 记录元信息 Process（执行单元） input → output script 块包裹命令 tag + publishDir conda / container 下发 输入 输出 👁️ 三者通过 DSL2 自动组合：Workflow 统揽全局，Channel 负责搬运，Process 专心计算

看到这张图，你应该能体会到：Workfow 不会直接调用代码，它只负责“编排”Channel 和 Process 的关系。每一个 Process 的输出 Channel 就是下一个 Process 的输入 Channel——这样一条链，Nextflow 自动解析成 DAG（有向无环图）。

💡 我建议你在写第一个完整流程前，先在纸上把 Channel 的流向画出来。这个习惯让我在后面调试时节省了大量时间。说实话，大多数流程跑失败的原因就两个：Channel 数据格式对不上，或者 Process 脚本路径写死了。

5. 避坑指南与我的经验

  • Process 传入的文件别用字符串拼接路径——用 path() 声明式包装。我曾遇到过用户明明传了 /data/reads.fastq.gz，但在 script 里用 $reads 却带上了工作目录的相对路径，找不到文件报了错。就是因为没把 input 声明为 path 类型。
  • Channel 不要重复消费——queue channel 被消费掉后就不存在了。如果多个 Process 共享同一份复杂数据，请用 Channel.value() 或 .view() 先看清数据流。
  • Workflow 内的变量作用域——Workflow 块内的变量只在块内可见。如果你希望调试或记录某个中间结果，建议显式赋值给一个 Channel。

现在我回想起来，第一次写 Nextflow 流程时，把 Process 写得特别重，反而忽略了 Workflow 的简洁性。后来重构时，我把核心逻辑拆成了标准 Process，用 Channel 连接它们，整个流程的审计路径就变得特别清晰——每一站的数据从哪来、到哪去，一目了然。

这就是为什么我们说 Process、Channel、Workflow 是 Nextflow 的三大基本功。它们不止是语法问题，更是架构思路。一旦你把这三个概念理解透了，接下来的高级话题（比如模块化、元数据传播、资源管理）就顺理成章了。

← 上一章 📖 返回目录 下一章 →