通知ActionBlock完成到另一个类。

Question

我写了一个实现TPL的方法 ActionBlock，它将做一个函数来找到我要Posting到块的元素的XPath。我正在从一个实时应用程序中发送元素（每当用户发现一个元素时，就将其发布到块中）。现在，我想在调用另一个类的保存按钮时检查该块是否完成。其逻辑是如果 ActionBlok 是完成的，当我们点击保存按钮时，元素会用一些保存逻辑来保存，否则会显示一个尚未准备好的消息框。第一次，这个想法是有效的，但是从第二个元素开始，这个 Actionblock 是不接受任何。我使用的是 block.Complete() 和 block.Completion.Wait() 的保存检查。现在我可以显示我正在调用codelogic。

1. 一旦创建一个元素，发布它 ActionBlock 级别1

......
jobQ.Enqueue(familarizedElement);
......

2. 内部 ActionBlock 班级2

public class TPLCommonDataFlow
{
    private ActionBlock<Element> _jobs;
    public static bool TPLFlowStatus = false;
    public static string JobStatus = string.Empty;
    public TPLCommonDataFlow()
    {
        var executionDataflowBlockOptions = new ExecutionDataflowBlockOptions()
        {
            MaxDegreeOfParallelism = 2,
        };

        _jobs = new ActionBlock<Element>((job) =>
        {
            Thread.Sleep(5);
            File.WriteAllText("C:\\Temp\\A.txt", "Started" + _jobs.InputCount.ToString());
            JobStatus = "started";
            Familiarization.FindAndUpdateXPath(job);
            File.WriteAllText("C:\\Temp\\A.txt", "Finished");
        }, executionDataflowBlockOptions);
        _jobs.Complete();


        //Wait for all messages to propagate through the network
        _jobs.Completion.Wait();

        //Checking all jobs are completed or not, if completed changing the boo, value.
        if (_jobs.InputCount == 0)
        {
            TPLFlowStatus = true;
            JobStatus = "stoped";
        }
    }
}

3. 对于保存，我检查了TPLFlowStatus的boolean值 级别3

if (class2.TPLFlowStatus == true)
{
    //Save Logic
}
else
{
    //Message Box showing Not Ready
}

现在，我想在每次工作完成时检查保存逻辑中的每个元素是否完成。如果有两个元素的块在 Queue 和一个完成，然后 MessageBox 需要在按下保存按钮后弹出。如果在块中全部完成，则需要进入保存逻辑。

Answer 1

你看到的问题是，当一个块完成后，它将不再接受新的消息，你已经告诉块你已经完成了发送消息。要开始发送新的一批消息，你可以不调用complete并以另一种方式跟踪批次完成来保持块的活力，或者你可以直接重置块。像这样一个简单的worker可能就是你要找的。

public class DataflowWorker
{
    private ActionBlock<string> _jobs;

    private void BuildJobsHandler()
    {
        _jobs = new ActionBlock<string>(x => Console.WriteLine(x) /*Do your work in the action block*/);
    }

    public async Task Enque(string element)
    {
        await _jobs.SendAsync(element);
    }

    public async Task CompleteJobsAndSave()
    {
        _jobs.Complete();
        await _jobs.Completion;
        //Save data when complete
        BuildJobsHandler();
    }
}