使用C# Channel实现工位流水线调度系统

在现代制造业中，流水线生产需要精确的工位协作。本文将介绍如何使用C#的Channel实现一个高效的工位流水线调度系统。

1、首先我们准备一个工位接口

 public interface IWorkstation{string WorkName { get; }Task StartAsync(CancellationToken cancellationToken);Task StopAsync();WorkstationStatus Status { get; }}public enum WorkstationStatus{Idle,Running,Paused,Faulted}

接着我们实现基类
基类封装了通用的启动、停止和状态管理逻辑：

  public abstract class WorkstationBase : IWorkstation, INotifyPropertyChanged{public virtual string WorkName { get; }private WorkstationStatus _status;private CancellationTokenSource _cts;private static readonly ITangdaoLogger Logger = TangdaoLogger.Get(typeof(WorkstationBase));public WorkstationStatus Status{get => _status;protected set{_status = value;OnPropertyChanged();}}public event PropertyChangedEventHandler PropertyChanged;public async Task StartAsync(CancellationToken parentToken){if (Status == WorkstationStatus.Running)return;Status = WorkstationStatus.Running;_cts = CancellationTokenSource.CreateLinkedTokenSource(parentToken);try{await ExecuteWorkAsync(_cts.Token);}catch (OperationCanceledException){Status = WorkstationStatus.Idle;}catch (Exception ex){Status = WorkstationStatus.Faulted;// 记录错误}}public async Task StopAsync(){_cts?.Cancel();await CleanupAsync();Status = WorkstationStatus.Idle;}protected abstract Task ExecuteWorkAsync(CancellationToken token);protected virtual Task CleanupAsync() => Task.CompletedTask;protected virtual void OnPropertyChanged([CallerMemberName] string name = null){PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(name));}}

我们现在可以实现流水线步骤了，假设我们的流水线只有4个工位，上料，预校，切割、下料
我们可以实现

 public class LoadWorkstation : WorkstationBase{private static long waferId = 0;public new string WorkName => "上料工位";private static readonly ITangdaoLogger Logger = TangdaoLogger.Get(typeof(LoadWorkstation));protected override async Task ExecuteWorkAsync(CancellationToken token){while (!token.IsCancellationRequested){Logger.WriteLocal($"生成产品: {wafer.CellId}");await Task.Delay(500, token);}}}

类似的其他工位也可以这样写
接着我们启动开始按钮前需要一个Task[]来全部启动

 public class WorkstationManager{private readonly List<IWorkstation> _workstations;private CancellationTokenSource _globalCts;public IReadOnlyCollection<IWorkstation> Workstations => _workstations.AsReadOnly();public WorkstationManager(){_workstations = new List<IWorkstation>{new LoadWorkstation(),new PreWorkstation(),new CutWorkstation(),new UnLoadWorkstation(),// 其他工位...};}public async Task StartAllAsync(){await StopAllAsync();_globalCts = new CancellationTokenSource();var startTasks = _workstations.Select(ws => ws.StartAsync(_globalCts.Token));await Task.WhenAll(startTasks);}public async Task StopAllAsync(){_globalCts?.Cancel();var stopTasks = _workstations.Select(ws => ws.StopAsync());await Task.WhenAll(stopTasks);}}

我们的ViewModel对应的开始按钮可以写

 public async void ExecuteStart(){CurrentStatus = RunStatus.Running;WorkstationManager workstationManager = new WorkstationManager();await workstationManager.StartAllAsync();}

各工位独立运行，缺乏协调
产品处理顺序无法保证
无法形成真正的"流水线"
资源竞争可能导致死锁
所有的产片都是并行的，不能保证切割顺序，此时我们需要用到Channel
我们新建一个传输器的实体类，用来演示传输的数据

 public class WaferMessage{public string CellId { get; }public DateTime EnterTime { get; }public WaferMessage(string cellId){CellId = cellId;EnterTime = DateTime.Now;}}

接着使用Channel新建一个状态机

 public static class ProductionLine{// 创建工位间的通道public static readonly Channel<WaferMessage> LoadToPre = Channel.CreateUnbounded<WaferMessage>();public static readonly Channel<WaferMessage> PreToCut = Channel.CreateUnbounded<WaferMessage>();public static readonly Channel<WaferMessage> CutToUnload = Channel.CreateUnbounded<WaferMessage>();}

我们修改刚才上料的代码

 public class LoadWorkstation : WorkstationBase{private static long waferId = 0;public new string WorkName => "上料工位";private static readonly ITangdaoLogger Logger = TangdaoLogger.Get(typeof(LoadWorkstation));protected override async Task ExecuteWorkAsync(CancellationToken token){while (!token.IsCancellationRequested){// 生成新产品，确保ID递增var id = Interlocked.Increment(ref waferId);var wafer = new WaferMessage($"Wafer-{id:D4}");Logger.WriteLocal($"生成产品: {wafer.CellId}");await Task.Delay(500, token);// 传递给预校工位await ProductionLine.LoadToPre.Writer.WriteAsync(wafer, token);Logger.WriteLocal($"{wafer.CellId} 完成上料 准备 预校");}}}

接着预校工位代码

 public class PreWorkstation : WorkstationBase{public new string WorkName => "预校工位";private static readonly ITangdaoLogger Logger = TangdaoLogger.Get(typeof(PreWorkstation));protected override async Task ExecuteWorkAsync(CancellationToken token){while (!token.IsCancellationRequested){// 从上料工位获取产品var wafer = await ProductionLine.LoadToPre.Reader.ReadAsync(token);Logger.WriteLocal($"{wafer.CellId} 开始预校");await Task.Delay(800, token);// 传递给切割工位await ProductionLine.PreToCut.Writer.WriteAsync(wafer, token);Logger.WriteLocal($"{wafer.CellId} 完成预校 准备 切割");}}}

切割工位代码

 public class CutWorkstation : WorkstationBase{public new string WorkName => "切割工位";private static readonly ITangdaoLogger Logger = TangdaoLogger.Get(typeof(CutWorkstation));protected override async Task ExecuteWorkAsync(CancellationToken token){while (!token.IsCancellationRequested){var wafer = await ProductionLine.PreToCut.Reader.ReadAsync(token);Logger.WriteLocal($"{wafer.CellId} 开始切割");await Task.Delay(800, token);// 传递给切割工位await ProductionLine.CutToUnload.Writer.WriteAsync(wafer, token);Logger.WriteLocal($"{wafer.CellId} 完成切割 → 下料");}}}

下料工位代码

 public class UnLoadWorkstation : WorkstationBase{private static readonly ITangdaoLogger Logger = TangdaoLogger.Get(typeof(UnLoadWorkstation));public new string WorkName => "下料工位";protected override async Task ExecuteWorkAsync(CancellationToken token){while (!token.IsCancellationRequested){var wafer = await ProductionLine.CutToUnload.Reader.ReadAsync(token);Logger.WriteLocal($"[下料] 开始: {wafer.CellId}");await Task.Delay(800, token); // 下料时间var totalTime = DateTime.Now - wafer.EnterTime;Logger.WriteLocal($"[下料] {wafer.CellId} ✓ 完成! 总耗时: {totalTime.TotalSeconds:F1}秒");}}}