Nextflow中实现按组非阻塞式处理并及时完成组内报告

要实现某车辆类型的所有step2完成后立即触发对应step3，核心是按车辆类型分组处理，避免使用.last()等待全局所有任务完成。下面是修改后的可运行示例：

修改后的完整代码 #

params.vehicles = [
    [type: 'car', id: 'car1'],
    [type: 'car', id: 'car2'],
    [type: 'truck', id: 'truck1'],
    [type: 'motorcycle', id: 'bike1'],
]

workflow {
    // 初始化车辆通道，格式为[类型, 车辆ID]
    Channel.fromList(params.vehicles)
        .map { veh -> [veh.type, veh.id] }
        .set { vehicle_ch }

    // step1：单个车辆处理
    step1(vehicle_ch)
    // step2：单个车辆处理，之后按类型分组
    step2(step1.out)
        .groupTuple(groupBy: 0)  // 按第一个元素（车辆类型）分组，收集该类型所有step2结果
        .map { type, step2_entries ->
            // 提取该类型所有step2的输出文件，整理为[type, 文件列表]
            def step2_files = step2_entries.collect { entry -> entry[2] }
            [type, step2_files]
        }
        .set { step2_grouped_ch }

    // step3：对应类型的所有step2完成后立即执行
    step3(step2_grouped_ch)
}

process step1 {
    input:
    tuple val(veh_type), val(veh_id)
    output:
    tuple val(veh_type), val(veh_id), path('step1_result.txt')
    script:
    """
    echo "Step1 processed ${veh_id} (${veh_type})" > step1_result.txt
    """
}

process step2 {
    input:
    tuple val(veh_type), val(veh_id), path(step1_file)
    output:
    tuple val(veh_type), val(veh_id), path('step2_result.txt')
    script:
    """
    echo "Step2 processed ${veh_id} (${veh_type})" > step2_result.txt
    cat ${step1_file} >> step2_result.txt
    """
}

process step3 {
    input:
    tuple val(veh_type), path(step2_files)
    output:
    path("${veh_type}_report.txt")
    script:
    """
    echo "=== Report for ${veh_type} ===" > ${veh_type}_report.txt
    cat ${step2_files} >> ${veh_type}_report.txt
    """
}

关键说明 #

1. groupTuple分组：groupTuple(groupBy: 0)会监听step2的输出通道，按车辆类型（第一个元素）自动分组。当某一类型的所有step2任务执行完成后，该分组的所有结果会被一次性推送到下游，无需等待其他类型的任务。
2. 避免全局阻塞：去掉.last()后，每个车辆类型的step3会独立触发——比如当所有car的step2完成后，car的step3会立刻启动，不用等truck或motorcycle的step2结束。
3. 动态适配分组：不管车辆类型是固定列表还是从外部文件动态读取，groupTuple都能自动识别所有出现的分组，无需提前配置类型数量。

运行这个工作流后，你会看到：car的两个step2完成后，car的report立即开始生成；truck的step2完成后，truck的report也会马上启动，完全独立于其他类型的任务进度。

内容的提问来源于stack exchange，提问作者user13388912