1. 从零理解流水车间调度与NEH算法想象一下你是一个小工厂的车间主任面前有6个不同的工件比如零件A、零件B...零件F每个工件都需要依次经过3台不同的机器比如切割机、钻孔机、喷漆机进行加工。你的任务就是决定这6个工件按照什么顺序上生产线才能让所有工件都加工完所花的总时间最短。这就是经典的流水车间调度问题英文叫Flow Shop Scheduling Problem。这个问题听起来简单但实际是个“硬骨头”。随着工件和机器数量的增加可能的排序方案数量会爆炸式增长。对于6个工件就有720种排法如果是10个工件就有三百多万种。想靠人力或者简单的穷举法找出最优解几乎是不可能的。所以工程师和学者们想出了很多启发式算法它们就像经验丰富的老调度员虽然不能保证每次都找到绝对最好的方案但能在非常短的时间内找到一个非常不错的、接近最优的方案。NEH算法就是这些“老调度员”中的佼佼者全称是Nawaz-Enscore-Ham算法由三位学者在1983年提出。它在解决流水车间调度特别是最小化总完工时间这个目标上表现非常出色至今都是很多更复杂算法的基石。它的核心思想非常符合直觉先加工那些“大块头”、耗时长的工件。因为如果把这些耗时长的工件往后排它们会堵住生产线导致后面的机器空闲等待白白浪费时间。那么NEH具体是怎么做的呢它主要分两大步。第一步是排序计算每个工件在所有机器上的加工时间总和然后按照这个总时间从大到小排序得到一个初始的工件序列。这就像把一堆待处理的文件按照页数从多到少摞起来。第二步是逐步插入优化这是一个关键且巧妙的过程。算法不会一次性决定所有工件的顺序而是像搭积木一样从排序好的序列里每次取出一个工件尝试把它插入到当前已构建的部分序列的所有可能位置然后计算每种插入方式下的总加工时间这个时间在调度领域叫Makespan最后选择那个使总加工时间最短的插入位置。就这样一个一个地把所有工件都安排进序列里。举个例子假设排序后工件的顺序是[工件3 工件1 工件5 工件2 工件4]。NEH算法会先构建只包含[工件3]的序列。然后考虑工件1尝试把它插在[工件3]的前面或后面即得到[1,3]和[3,1]两个候选序列计算哪个的总时间短就选哪个。假设[3,1]更好那么当前序列就是[3,1]。接着取工件5尝试插入到[3,1]的三个可能位置最前、中间、最后即[5,3,1]、[3,5,1]、[3,1,5]再计算并选择最优的。如此反复直到所有工件都插入完毕。这个过程听起来是不是有点像我们整理扑克牌时的插入排序没错NEH算法的优化核心就是基于插入和评估。接下来我们就用C一步步把这个聪明的“调度员”从想法变成实实在在可以运行的代码。2. 搭建C项目框架与核心数据结构在动手写算法之前我们得先把“舞台”搭好。这个舞台主要包括两部分一是如何表示问题本身的数据二是设计好算法需要用到哪些“工具”函数。我们先创建一个简单的C项目比如一个名为neh_scheduler.cpp的文件。首先我们需要定义问题的规模。有多少个工件多少台机器这决定了我们数据结构的尺寸。我们定义两个全局变量num_jobs和num_machines。为了方便测试我们可以先写死一个例子比如6个工件3台机器。#include iostream #include vector #include algorithm #include climits // 用于INT_MAX using namespace std; // 问题规模 int num_jobs 6; // 工件数量 int num_machines 3; // 机器数量 // 加工时间矩阵times[i][j] 表示工件i在第j台机器上的加工时间 // 这里用一个二维vector方便动态调整大小 vectorvectorint processing_times { {19, 46, 65}, // 工件0的加工时间 {44, 63, 12}, // 工件1 {85, 56, 98}, // 工件2 {59, 68, 25}, // 工件3 {87, 66, 53}, // 工件4 {51, 4, 63} // 工件5 };这个processing_times矩阵就是我们问题的核心输入。每一行代表一个工件每一列代表一道工序一台机器。比如processing_times[2][1] 56表示工件2在机器1第二台机器上需要加工56个单位时间。接下来我们要设计算法运行过程中最重要的几个工具函数。第一个就是计算给定工件序列下的总完工时间。这是NEH算法的“裁判”每次插入尝试后都要靠这个函数来打分判断哪个位置更好。计算流水车间的总完工时间不能简单地把每个工件的时间加起来因为机器有先后顺序工件必须等前一道工序完成后才能进入下一台机器。我们需要模拟整个加工过程。通常使用一个二维数组completion_time[machine][job_position]来记录每台机器上每个工件的完成时间但更高效的方法是只记录上一行上一台机器和当前行的完成时间进行递推计算。我们来写这个核心的calculate_makespan函数/** * 计算给定工件序列的总完工时间Makespan * param sequence 工件序列存储的是工件的索引号 * return 总完工时间 */ int calculate_makespan(const vectorint sequence) { int job_count sequence.size(); // 创建两个数组分别记录当前机器和上一台机器的完成时间 vectorint current_machine_time(num_machines, 0); vectorint previous_machine_time(num_machines, 0); // 遍历序列中的每一个工件 for (int job_idx_in_seq 0; job_idx_in_seq job_count; job_idx_in_seq) { int job_id sequence[job_idx_in_seq]; // 当前要加工的工件号 // 遍历每一台机器 for (int machine 0; machine num_machines; machine) { if (machine 0) { // 第一台机器完成时间 上一工件在第一台机器的完成时间 当前工件在第一台机器的加工时间 current_machine_time[machine] previous_machine_time[machine] processing_times[job_id][machine]; } else { // 其他机器完成时间 max(当前工件在前一台机器的完成时间 上一工件在当前机器的完成时间) 当前加工时间 int ready_time current_machine_time[machine - 1]; // 工件在前一台机器完工的时间 int machine_available_time previous_machine_time[machine]; // 当前机器空闲的时间 current_machine_time[machine] max(ready_time, machine_available_time) processing_times[job_id][machine]; } } // 当前机器时间计算完毕将其赋给“上一机器时间”为下一个工件做准备 previous_machine_time current_machine_time; } // 最后一台机器的最后一个工件的完成时间就是总完工时间 return current_machine_time[num_machines - 1]; }我解释一下这个模拟过程。假设当前序列是[2, 0, 1]我们正在加工第二个工件job_id0。对于机器0它只需要等上一个工件job_id2在机器0上干完活previous_machine_time[0]然后自己再干19个单位时间。对于机器1情况就复杂点工件0必须先在机器0上干完活current_machine_time[0]同时机器1本身也必须等它干完上一个工件job_id2的活previous_machine_time[1]。这两个时间哪个晚工件0就从哪个时间点开始在机器1上加工。这个过程就像一条流水线工件是流动的“物料”机器是“工位”物料必须按顺序经过每个工位且每个工位一次只能处理一个物料。有了这个“裁判”函数我们还需要一个“操作员”函数负责把工件插入到序列的指定位置。C标准库的vector有insert函数但为了更清晰地展示过程我们可以自己写一个/** * 将一个新工件插入到序列的指定位置 * param original_seq 原始序列 * param job_to_insert 要插入的工件ID * param insert_position 插入位置0-based索引 * return 插入后的新序列 */ vectorint insert_job(const vectorint original_seq, int job_to_insert, int insert_position) { vectorint new_sequence original_seq; new_sequence.insert(new_sequence.begin() insert_position, job_to_insert); return new_sequence; }准备工作就绪舞台已经搭好。接下来我们就要请出主角NEH算法看看它如何运用这些工具一步步构建出优秀的调度方案。3. 实现NEH算法的第一步初始排序NEH算法的第一步非常直接就是给所有工件按“总工作量”排个座次。总工作量就是每个工件在所有机器上加工时间的总和。这个步骤背后的逻辑是优先安排那些加工时间长的“大块头”工件让它们早点进入生产线避免它们堵在后面造成后续机器的长时间空闲。我们来写这个初始化排序的函数initial_neh_sort/** * NEH第一步根据总加工时间对工件进行降序排序 * return 返回排序后的工件索引向量 */ vectorint initial_neh_sort() { // 创建一个向量存储每个工件的索引和总加工时间 vectorpairint, int job_total_time; // pair工件索引 总加工时间 for (int i 0; i num_jobs; i) { int total 0; for (int j 0; j num_machines; j) { total processing_times[i][j]; } job_total_time.push_back({i, total}); // 存储工件索引和它的总时间 } // 使用lambda表达式按照总加工时间从大到小排序 sort(job_total_time.begin(), job_total_time.end(), [](const pairint, int a, const pairint, int b) { return a.second b.second; // 降序排列 }); // 提取排序后的工件索引 vectorint sorted_jobs; for (const auto p : job_total_time) { sorted_jobs.push_back(p.first); } // 输出排序结果方便调试 cout 初始排序结果按总加工时间降序: ; for (int idx : sorted_jobs) { cout J idx ( job_total_time[idx].second ) ; } cout endl; return sorted_jobs; }让我们用之前的数据跑一下看看。计算每个工件的总时间工件0: 194665 130工件1: 446312 119工件2: 855698 239工件3: 596825 152工件4: 876653 206工件5: 51463 118按照从大到小排序结果应该是[工件2(239), 工件4(206), 工件3(152), 工件0(130), 工件1(119), 工件5(118)]。注意我们的索引是从0开始的所以对应的索引序列是[2, 4, 3, 0, 1, 5]。这个排序结果就是NEH算法后续操作的“候选列表”。算法会从这个列表里按顺序一个一个地取出工件安排进最终的调度序列里。你可能会问为什么不直接用这个排序当最终结果因为流水线调度不是简单的单个机器排序工件之间的相互影响很复杂。一个总时间长的工件如果它的加工时间集中在后面的机器而另一个总时间稍短的工件如果它的加工时间集中在开头的机器那么先安排后者可能反而能让生产线更流畅。所以我们需要第二步的插入优化来精细调整。这里有个小细节原始NEH论文和一些实现中排序后就直接用这个列表了。但在我们接下来的实现中我们会严格遵循“取出-插入”的流程。现在我们有了一个有序的工件列表下一步就是最核心的迭代插入过程了。4. 实现NEH算法的核心迭代插入优化这是NEH算法最精彩的部分也是它比简单排序强大得多的原因。它不像一些贪心算法那样做一个决定就再也不回头。NEH的插入过程是逐步构造且局部最优的每加入一个新工件它都会重新审视当前已构建的局部序列为这个新工件找到一个最好的位置从而保证在每一步当前的部分序列都是针对已考虑工件的最优或近似最优排列。我们来详细拆解这个过程。假设经过初始排序我们得到的列表是L [J2, J4, J3, J0, J1, J5]。算法从一个空序列S开始。第一步处理第一个工件J2。当前部分序列S为空。将J2插入空序列只有一种可能S [J2]。计算makespan([J2])。这个值就是J2自己在三台机器上加工的总时间即855698239注意由于机器间存在等待实际总完工时间就是各工序时间之和因为只有一个工件。记录当前最优序列为[J2]。第二步处理第二个工件J4。当前部分序列S [J2]。将J4插入到[J2]的所有可能位置。对于长度为1的序列有2个插入点在J2之前或在J2之后。候选1:[J4, J2]候选2:[J2, J4]分别计算两个候选序列的总完工时间makespan。假设makespan([J4, J2]) 320假设makespan([J2, J4]) 310选择makespan更小的那个序列作为新的S。这里我们选择[J2, J4]因为310 320。第三步处理第三个工件J3。当前部分序列S [J2, J4]长度为2。将J3插入到[J2, J4]的所有可能位置。对于长度为2的序列有3个插入点最前、中间、最后。候选1:[J3, J2, J4]候选2:[J2, J3, J4]候选3:[J2, J4, J3]分别计算三个候选的makespan。假设结果分别是350, 340, 360选择makespan最小的[J2, J3, J4]作为新的S。后续步骤重复这个过程每次从列表L中取出下一个工件将其插入当前最优部分序列S的所有可能位置评估并选择makespan最小的排列作为新的S直到列表L中的所有工件都被处理完毕。现在我们把这一过程用C实现出来。我们将创建一个函数neh_algorithm它调用之前写的排序函数和计算函数。/** * 完整的NEH算法实现 * return 返回找到的最优或近似最优工件序列 */ vectorint neh_algorithm() { // 步骤1获取初始排序列表 vectorint job_list initial_neh_sort(); // 步骤2初始化当前最优部分序列 vectorint current_sequence; current_sequence.push_back(job_list[0]); // 先放入第一个工件 cout 开始NEH迭代插入优化过程 endl; cout 初始序列: [J current_sequence[0] ] endl; cout Makespan: calculate_makespan(current_sequence) endl; cout ---------------------------------------- endl; // 步骤3迭代插入剩余工件 for (int i 1; i job_list.size(); i) { int new_job job_list[i]; // 本次要插入的新工件 int best_position 0; int best_makespan INT_MAX; // 初始化为最大整数 vectorint best_candidate; cout 处理新工件 J new_job endl; cout 当前序列: [; for (int idx : current_sequence) cout J idx ; cout ] endl; // 尝试将新工件插入到所有可能的位置 (0 到 current_sequence.size()) for (int pos 0; pos current_sequence.size(); pos) { // 生成候选序列 vectorint candidate current_sequence; candidate.insert(candidate.begin() pos, new_job); // 计算该候选序列的makespan int current_makespan calculate_makespan(candidate); cout 尝试插入位置 pos : [; for (int idx : candidate) cout J idx ; cout ] - Makespan current_makespan; // 判断是否优于当前找到的最佳位置 if (current_makespan best_makespan) { best_makespan current_makespan; best_position pos; best_candidate candidate; cout -- 新的最佳位置; } cout endl; } // 更新当前最优序列为本次找到的最佳候选 current_sequence best_candidate; cout **选定位置 best_position 更新后序列: [; for (int idx : current_sequence) cout J idx ; cout ]; cout Makespan best_makespan endl; cout ---------------------------------------- endl; } cout NEH算法执行完毕 endl; cout 最终调度序列: [; for (int idx : current_sequence) cout J idx ; cout ] endl; cout 预估总完工时间: calculate_makespan(current_sequence) endl; return current_sequence; }让我们结合之前的例子模拟一下这个函数的输出。当处理第二个工件J4时循环会输出类似这样的信息处理新工件 J4 当前序列: [J2 ] 尝试插入位置 0: [J4 J2 ] - Makespan 320 尝试插入位置 1: [J2 J4 ] - Makespan 310 -- 新的最佳位置 **选定位置 1更新后序列: [J2 J4 ] Makespan 310这个过程非常清晰你可以看到算法在每一步是如何做决策的。它没有复杂的数学公式就是朴素的“尝试-比较-选择”但正是这种逐步优化的策略使得它在实际应用中非常有效。你可以尝试修改processing_times矩阵里的数字观察最终序列和完工时间的变化感受不同加工时间分布对调度结果的影响。5. 代码整合、测试与性能思考现在我们把所有代码片段整合到一个完整的、可以编译运行的C程序中。我们还会添加一个简单的main函数来驱动整个流程并允许用户输入自定义的加工时间矩阵而不是使用硬编码的数据。#include iostream #include vector #include algorithm #include climits using namespace std; // 全局变量 int num_jobs; int num_machines; vectorvectorint processing_times; // 函数声明 int calculate_makespan(const vectorint sequence); vectorint initial_neh_sort(); vectorint neh_algorithm(); void print_matrix(const vectorvectorint matrix); int main() { cout NEH算法流水车间调度求解器 endl; // 用户输入问题规模 cout 请输入工件数量: ; cin num_jobs; cout 请输入机器数量: ; cin num_machines; // 初始化并输入加工时间矩阵 processing_times.resize(num_jobs, vectorint(num_machines)); cout \n请输入加工时间矩阵 ( num_jobs 行 x num_machines 列): endl; for (int i 0; i num_jobs; i) { cout 工件 i 的加工时间用空格分隔: ; for (int j 0; j num_machines; j) { cin processing_times[i][j]; } } cout \n您输入的加工时间矩阵为 endl; print_matrix(processing_times); // 执行NEH算法 vectorint best_sequence neh_algorithm(); // 输出最终结果 cout \n 最终调度方案 endl; cout 最优工件序列: ; for (int job_id : best_sequence) { cout J job_id ; } cout endl; cout 总完工时间 (Makespan): calculate_makespan(best_sequence) endl; return 0; } // 计算Makespan的函数实现同上略 int calculate_makespan(const vectorint sequence) { int job_count sequence.size(); vectorint current_machine_time(num_machines, 0); vectorint previous_machine_time(num_machines, 0); for (int job_idx_in_seq 0; job_idx_in_seq job_count; job_idx_in_seq) { int job_id sequence[job_idx_in_seq]; for (int machine 0; machine num_machines; machine) { if (machine 0) { current_machine_time[machine] previous_machine_time[machine] processing_times[job_id][machine]; } else { int ready_time current_machine_time[machine - 1]; int machine_available_time previous_machine_time[machine]; current_machine_time[machine] max(ready_time, machine_available_time) processing_times[job_id][machine]; } } previous_machine_time current_machine_time; } return current_machine_time[num_machines - 1]; } // 初始排序函数实现同上略 vectorint initial_neh_sort() { vectorpairint, int job_total_time; for (int i 0; i num_jobs; i) { int total 0; for (int j 0; j num_machines; j) { total processing_times[i][j]; } job_total_time.push_back({i, total}); } sort(job_total_time.begin(), job_total_time.end(), [](const pairint, int a, const pairint, int b) { return a.second b.second; }); vectorint sorted_jobs; for (const auto p : job_total_time) { sorted_jobs.push_back(p.first); } return sorted_jobs; } // NEH主算法函数实现同上略 vectorint neh_algorithm() { vectorint job_list initial_neh_sort(); vectorint current_sequence; current_sequence.push_back(job_list[0]); for (int i 1; i job_list.size(); i) { int new_job job_list[i]; int best_position 0; int best_makespan INT_MAX; vectorint best_candidate; for (int pos 0; pos current_sequence.size(); pos) { vectorint candidate current_sequence; candidate.insert(candidate.begin() pos, new_job); int current_makespan calculate_makespan(candidate); if (current_makespan best_makespan) { best_makespan current_makespan; best_position pos; best_candidate candidate; } } current_sequence best_candidate; } return current_sequence; } // 辅助函数打印矩阵 void print_matrix(const vectorvectorint matrix) { for (const auto row : matrix) { for (int val : row) { cout val \t; } cout endl; } }你可以将这段代码复制到一个C文件中比如neh_flowshop.cpp使用g或你喜欢的IDE进行编译运行。运行后程序会提示你输入工件数、机器数和具体的加工时间。输入我们一直使用的例子6, 3, 以及那6行时间数据看看最终输出的序列是否和你手动推导的一致。关于性能的思考我们的实现是清晰易懂的但在处理大规模问题时比如几百个工件效率会成为瓶颈。瓶颈主要在于calculate_makespan函数被调用了太多次。在插入第k个工件时我们需要计算k1个候选序列的makespan每个计算需要O(k*m)的时间m是机器数。因此整个NEH算法的时间复杂度大约是O(n³ * m)其中n是工件数。在实际的工业级应用中有几种常见的优化手段加速Makespan计算利用插入操作只影响局部这一特性设计增量式计算makespan的方法避免每次都对整个序列重新计算。这需要维护更多的状态信息。使用更快的评估方法对于某些特定类型的流水车间如置换流水车间存在一些下界或近似公式可以快速评估插入位置的好坏从而减少需要精确计算makespan的次数。并行计算尝试不同插入位置的makespan计算是相互独立的可以很容易地用多线程并行处理。尽管有这些优化NEH算法本身作为一个启发式算法其核心价值在于用合理的计算时间获得高质量的解。它常常作为更复杂元启发式算法如遗传算法、迭代贪婪算法的初始化步骤或局部搜索算子。自己动手实现这个基础版本能让你深刻理解流水车间调度的复杂性和启发式算法的设计哲学。下次当你需要安排一系列有前后依赖关系的任务时或许NEH算法的思想能给你带来启发。