4、进程与通道：process基本结构、channel定义与传递
走到第四章，咱们终于要动手写真正的 Nextflow 核心了。前面三章搭建了环境、理解了数据流的概念，现在我们把这两样东西串起来——进程和通道，就是 Nextflow 世界里最基础的两个构件。
我个人习惯把进程比作「工人」，通道比作「传送带」。工人只管干活，传送带负责把原材料送到工人手里，再把成品运走。你想想看，只要把这两样定义清楚，整个管道就跑起来了。
4.1 进程的基本结构：一个工人的自白
一个 Nextflow 进程，说白了就是一段可执行的脚本，外加一堆声明。它长什么样？来看一个最精简的例子：
process SAY_HELLO {
input:
val greeting
output:
val message
script:
"""
echo "$greeting, world!"
"""
}这里有个细节我刚开始学的时候踩过坑——进程名必须全大写。不是可选的，是强制规范。Nextflow 的解析器看到小写进程名会直接报错，别问我怎么知道的（苦笑）。
一个完整进程结构包含五个区段：
| 区段 | 作用 | 典型写法 |
|---|---|---|
directives |
控制资源配置、标签、执行策略 | cpus 4、memory 8.GB、tag "my_process" |
input |
声明从通道接收什么类型的数据 | val x、path file、env VAR |
output |
声明进程产生什么数据发送到通道 | val result、path "*.bam" |
script |
实际执行的命令/脚本块 | 双引号内任意 Shell 命令 |
stub |
快速测试时用于替代真实脚本的模拟块 | 新版本特性，调试利器 |
嗯，这里要注意，输入输出声明是有顺序要求的。输入必须写在输出前面，不然解析器会报语法错误。我在项目中遇到过有人因为这个顺序问题排查了半天，结果就是调换一下位置的事。
4.2 通道的定义：传送带怎么搭？
通道是 Nextflow 数据流的血管。没有通道，进程之间的数据就没办法传递。通道的核心作用就三件事：创建、转换、连接。
创建通道最简单的方式有两种：
// 从值创建——一次性发送一个值
ch_greeting = Channel.of("hello")
// 从列表创建——逐个发送每个元素
ch_samples = Channel.fromPath("data/*.fastq")
// 工厂方法创建——更灵活
ch_pairs = Channel.fromFilePairs("data/*_{1,2}.fastq")
为什么会用这么多种创建方式？因为数据源的性质不同。一个配置文件里的参数，可能是单个值；一批测序数据，那就是文件列表。你想想看，用同一种方法处理所有情况，反而更麻烦。
.of() 或 .value()，后者用 .fromPath() 或 .fromFilePairs()。这个习惯能让管道更清晰。
4.3 通道的类型：你真的了解传送带吗？
通道根据数据发送方式，分为两大阵营：队列通道和值通道。
| 类型 | 行为 | 适用场景 | 注意点 |
|---|---|---|---|
QueueChannel |
一次性发送，消费后释放 | 文件列表、样本列表、分批数据 | 只能被消费一次，重复使用需重新绑定 |
ValueChannel |
始终保留最后一个值，可重复读取 | 参数、配置、常量 | 必须显式创建，用 .value() |
我曾经犯过一个经典错误：把一个值通道当成队列通道用，结果进程重复执行了无数次。后来才明白，值通道的 .value() 创建的是「单例」，任何时候读取都是同一个值；而队列通道的 .of() 或 .fromPath() 创建的是「流」，读取一个就少一个。
如果你不确定当前通道是什么类型，有个简单方法：在 script 块里打印它，看输出的是单个值还是列表。嗯，有时候老的调试手段反而最管用。
4.4 通道的传递：怎么把数据送到进程手里？
进程通过 input 声明来消费通道。最核心的规则是：进程的输入声明必须和通道发送的数据类型匹配。
// 定义两个通道
ch_sample_names = Channel.of("sample_A", "sample_B")
ch_fastq_files = Channel.fromPath("data/*.fastq")
// 进程消费
process QC_READS {
input:
val sample_name
path fastq
script:
"""
fastqc -o results/ $fastq
"""
}
// 连接——把通道传给进程
workflow {
QC_READS(ch_sample_names, ch_fastq_files)
}这里有个关键点：传递顺序和声明顺序必须一致。上面例子中，第一个通道 ch_sample_names 给第一个输入 sample_name，第二个通道给第二个输入 fastq。这个顺序一旦搞反，轻则数据错位，重则进程直接失败。
我记得有一次帮同事调试，他坚持说「不可能传错」，结果一检查，他把 path 类型和 val 类型颠倒了。Nextflow 虽然会做类型检查，但有些情况下（比如文件路径也是字符串）它可能不会报错，只是结果异常。所以我的习惯是：在进程开头加一行 echo "input: $sample_name" 做热身检查，跑一次就能看出是不是传对了。
4.5 通道操作符：传送带上的分拣器
实际项目中，很少直接拿原始通道就开干。通常需要对通道做各种变换——过滤、合并、分支、分割。Nextflow 提供了一套完整操作符，就像传送带上的分拣机器人。
// 过滤——只保留满足条件的元素
ch_good = ch_all.filter { it =~ /^sample_/ }
// 映射——对每个元素做变换
ch_renamed = ch_good.map { name -> "${name}_processed" }
// 合并——把多个通道合成一个
ch_combined = ch_a.mix(ch_b)
// 分组——按规则把元素分到不同进程
ch_grouped = ch_fastq.groupTuple()
// 分割——把一个集合元素拆成单个
ch_single = ch_list.flatten()
这里有个实战技巧：能不写循环就不要写循环。很多初学者习惯在 script 块里写 for 循环处理多个文件，其实用操作符就能优雅地搞定。你想想看，.map 和 .filter 写出来不仅代码短，而且语义明确——看名字就知道它在干什么。
4.6 避坑指南：我踩过的几个坑
坑一：通道重复消费。队列通道被消费一次后就空了，同一个通道不能传给两个进程。如果你需要多个进程共用同一批数据，必须显式复制或使用 .buffer() 操作符。
// 错误示范——ch_data 只够用一次
process A { input: val data; script: ... }
process B { input: val data; script: ... }
workflow {
A(ch_data)
B(ch_data) // 这里 ch_data 已经是空的了！
}
// 正确做法——用 buffer 复制
workflow {
ch_copy = ch_data.buffer()
A(ch_data)
B(ch_copy)
}坑二：值通道当队列用。如果你想让一个值可以被多个进程共享，直接用 .value() 创建值通道。它天然支持重复读取。
坑三：进程输出没绑定到通道。有些新手以为进程执行完，结果自然就流向下一个进程了。不对，你得在 output 里显式声明，才能把结果发到通道里。
4.7 知识结构总览：一图看懂进程与通道
4.8 实战片段：三句话总结
- 进程定义「做什么」——写好 input、output、script，注意类型匹配和顺序。
- 通道定义「传什么」——选对创建方式（队列/值），善用操作符做数据预处理。
- 连接就是「谁和谁」——workflow 块里把通道传进进程，一条管道就通了。