使用 TPL 和/或后台辅助角色处理间歇性 IO 作业

本文关键字：处理 作业 IO 角色 TPL 后台 使用 | 更新日期: 2023-09-27 18:06:40

我有一个与一些串行和USB设备交互的C#应用程序。我正在尝试找到一种以并行方式命令这些设备的好方法，以便可以同时控制许多设备。我将通过用户输入和脚本命令向设备启动命令。

我目前正在尝试找出一种干净的方法来"收集"并与我的 UI 线程并行运行命令。我有多个窗体，每个设备一个，每个窗体有几个可以发出命令的控件。

我目前看到它的方式是用户将单击一个按钮，该按钮将在表单中触发事件。另一个类让我们称之为CommandManager它将钩接到所有这些事件;每个事件传递必要的信息以形成要发送到设备的命令。

当事件由CommandManager处理时，它会形成命令，并将其添加到名为 DeviceCommandWorker 的子类化BackgroundWorker中的BlockingCollection<Command>，该在应用程序打开时启动。它所做的只是循环访问包含Task.Factory.StartNew()调用的代码块。

在StartNew块中，Take在等待输入命令的BlockingCollection上调用。一旦命令进入集合，Take就会返回，Task就会继续它的快乐方式。BackgroundWorker循环并重复该过程，直到它被取消。

// Event handler running on the UI Thread
public void ProcessCommand(DeviceCommand command)
{
    // I assume I cannot add to Commands (BlockingCollection) from here?
    DeviceCommandWorker.Instance.Commands.Add(command);
}
// ....
// BackgroundWorker started upon Application startup
// ...
public class DeviceCommandWorker : BackgroundWorker
{
    public static DeviceCommandWorker Instance { get { return lazyInstance.Value; } }
    private static readonly Lazy<DeviceCommandWorker> lazyInstance = new Lazy<DeviceCommandWorker>(() => new DeviceCommandWorker());
    public BlockingCollection<DeviceCommand> Commands { get; set; } 
    private DeviceCommandWorker()
    {
        WorkerSupportsCancellation = true;
        Commands = new BlockingCollection<DeviceCommand>();
    }
    protected override void OnDoWork(DoWorkEventArgs e)
    {
        while (!CancellationPending)
        {
            var commandTask = Task.Factory.StartNew(() =>
            {
                // If the BackGroundWorker is cancelled, how to esacpe this blocking call?
                DeviceCommand command = commandQueue.Take(); 
                DeviceCommandResult result;
                command.Process(out result);
                if(result == DeviceCommandResult.Error)
                    ; // How would I inform the UI of this result?
            });
        }
        e.Cancel = true;
    }
}

我的问题在上面的代码中陈述，但我会重申它们。

我

的第一个问题是我认为我不能从运行BackGroundWorker之外添加到BlockingCollection中？(通过阅读有关 TPL 的信息，添加时不应该有一个我可以锁定的同步对象吗？

假设我可以添加到集合中，那么当它被阻止时，是否有办法转义Take方法，特别是如果BackgroundWorker被取消，它不会永远被阻止吗？(可能在取消之前，我可以发送一个"自定义命令"，该命令只是创建一个不执行任何操作的任务，然后我将能够退出while循环(

最后，如何将命令执行中的成功或错误报告回 UI 线程？我已经看到了这个答案，这是正确的方向吗？

使用 TPL 和/或后台辅助角色处理间歇性 IO 作业

因此，经过更多的修补，看起来我最初得到了上面的想法对我来说效果很好。我不确定在做一些实际工作时它会如何表现，但到目前为止我很高兴。

所以我重新设计了代码。我发现没有Task.Factory.StartNew()之外的Take，我实际上只是尽可能快地完成任务，每个任务都等待从BlockingCollection中消耗。所以我把语句移到了循环之外。我还确认，要摆脱阻塞Take我需要发送某种特殊命令，以便我可以停止后台工作线程。最后(未显示(我计划使用 Control.Invoke 将任何失败返回到 UI 线程。

public Boolean StartCommandWorker()
{
    if (DeviceCommandWorker.Instance.IsBusy)
        return false;
    else
    {
        Console.Out.WriteLine("Called start command worker!");
        DeviceCommandWorker.Instance.RunWorkerAsync();
        return DeviceCommandWorker.Instance.IsBusy;
    }
}
public void StopCommandWorker()
{
    Console.Out.WriteLine("Called stop command worker!");
    ProcessCommand("QUIT");
    DeviceCommandWorker.Instance.CancelAsync();
}
public Boolean ProcessCommand(String command)
{
    DeviceCommandWorker.Instance.Commands.Add(command);
    Console.Out.WriteLine("Enqueued command '"" + command + "'" ThreadID: " + Thread.CurrentThread.ManagedThreadId);
    return true;
}
internal class DeviceCommandWorker : BackgroundWorker
{
    public static DeviceCommandWorker Instance { get { return lazyInstance.Value; } }
    private static readonly Lazy<DeviceCommandWorker> lazyInstance = new Lazy<DeviceCommandWorker>(() => new DeviceCommandWorker());
    public BlockingCollection<String> Commands { get; set; }
    private DeviceCommandWorker()
    {
        WorkerSupportsCancellation = true;
        Commands = new BlockingCollection<String>();
    }
    protected override void OnDoWork(DoWorkEventArgs e)
    {
        Console.Out.WriteLine("Background Worker Started ThreadID: " + Thread.CurrentThread.ManagedThreadId);
        while (!CancellationPending)
        {
            Console.Out.WriteLine("Waiting for command on ThreadID: " + Thread.CurrentThread.ManagedThreadId);
            String aString = Commands.Take();
            var commandTask = Task.Factory.StartNew(() =>
            {
                Console.Out.WriteLine("   Dequeued command '"" + aString + "'" ThreadID: " + Thread.CurrentThread.ManagedThreadId);
                if (aString.Equals("QUIT"))
                    Console.Out.WriteLine("   Quit worker called: " + aString);
            });
        }
        Console.Out.WriteLine("Background Worker: Stopped!");
        e.Cancel = true;
    }
}

下面是一些示例输出。我制作了一个小的 UI 窗体，可以用来启动、停止和发送命令。它只是发送一个随机的双精度作为命令。

Called start command worker!
Background Worker Started ThreadID: 17
Waiting for command on ThreadID: 17
Enqueued command "Hello" ThreadID: 10
Waiting for command on ThreadID: 17
   Dequeued command "Hello" ThreadID: 16
Enqueued command "0.0745" ThreadID: 10
Waiting for command on ThreadID: 17
   Dequeued command "0.0745" ThreadID: 12
Enqueued command "0.7043" ThreadID: 10
   Dequeued command "0.7043" ThreadID: 16
Waiting for command on ThreadID: 17
Called stop command worker!
Enqueued command "QUIT" ThreadID: 10
   Dequeued command "QUIT" ThreadID: 12
   Quit worker called: QUIT
Background Worker: Stopped!
Enqueued command "0.2646" ThreadID: 10
Enqueued command "0.1619" ThreadID: 10
Enqueued command "0.5570" ThreadID: 10
Enqueued command "0.4129" ThreadID: 10
Called start command worker!
Background Worker Started ThreadID: 12
Waiting for command on ThreadID: 12
Waiting for command on ThreadID: 12
Waiting for command on ThreadID: 12
Waiting for command on ThreadID: 12
Waiting for command on ThreadID: 12
   Dequeued command "0.2646" ThreadID: 16
   Dequeued command "0.1619" ThreadID: 16
   Dequeued command "0.5570" ThreadID: 16
   Dequeued command "0.4129" ThreadID: 16
Enqueued command "0.8368" ThreadID: 10
   Dequeued command "0.8368" ThreadID: 17
Waiting for command on ThreadID: 12
Called stop command worker!
Enqueued command "QUIT" ThreadID: 10
   Dequeued command "QUIT" ThreadID: 16
   Quit worker called: QUIT
Background Worker: Stopped!
相关文章：