Tpl实施问题

Question

Tpl Sample Click here to download

嗨， 我正在尝试在我的项目中实现 tpl。我使用 Web 客户端创建了 5 个并行 HTTP 调用。

我想要达到的目标。 如果 5 个并行调用中的任何一个返回包含 "First" 的字符串，则终止其余调用并继续调用返回 "First"。

我尝试过的：

我在上面附上了示例代码。我使用了以下谓词函数。

async Task<T> WhenAny<T>(IEnumerable<Task<T>> tasks, Func<T, bool> predicate)
    {
        var taskList = tasks.ToList();
        Task<T> completedTask = null;
        do
        {
            completedTask = await Task.WhenAny(taskList);
            taskList.Remove(completedTask);
        } while (!predicate(await completedTask) && taskList.Any());

        return completedTask == null ? default(T) : await completedTask;
    }

并调用如下：

 public async Task<string> methodname()
    {
        string sUrl = "https://abcd.com/test.php";
        Task<string> task1 = DownLoadData(sUrl);
        Task<string> task2 = DownLoadData(sUrl);
        Task<string> task3 = DownLoadData(sUrl);
        Task<string> task4 = DownLoadData(sUrl);
        Task<string> task5 = DownLoadData(sUrl); 
        var tasks = new[] { task1, task2, task3, task4, task5 };
        await WhenAny(tasks, t => t.Contains("First"));

        return "";


    }

但它不符合标准。请建议我在哪里遗漏了一些东西。任何帮助将不胜感激。

Answer 1

我假设您有一个要检查的 url 列表，而不是重复检查相同的 url（？！），但是在任何情况下都可以使用相同的方法....

我喜欢使用TPL Dataflow 来处理这类事情。它允许您将多个异步操作（例如下载字符串）与多个同步操作（例如检查子字符串“First”的内容）链接在一起，并提供各种“旋钮”来调整控制并行度、缓冲区大小等。我们还可以传递一个取消令牌来取消进一步的处理。所以首先让我们创建一个取消令牌源：

var cancellationSource = new CancellationTokenSource();

现在，我们的第一个“块”将是 TransformBlock<string,string>。这个块的工作是通过调用你的DownloadData 函数来获取 url 字符串并将其“转换”为内容字符串：

var downloadBlock = new TransformBlock<string,string>(
        async s => await DownloadData(s), 
        new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 10, //allow up to 10 simulteneous downloads
            BoundedCapacity = 100,       //allow the block to store up to 100 urls in a buffer
            CancellationToken = cancellationSource.Token 
        });

第二个块是TransformMany<string,string> 块。这种类型的块允许我们将字符串转换为字符串的集合。我们只是通过返回一个空集合来过滤掉任何不符合我们标准的内容字符串：

    var filterBlock = new TransformManyBlock<string,string>(
        s => {
                if (s.Contains("First")) 
                {
                    return new string[]{s};
                } else 
                {
                    return new string[]{};
                }
            }); // (we could also pass a cancellation token here)

最后一个块是您将对找到的内容字符串执行某些操作的地方（即您找到的第一个包含“First”的下载内容）。我们还 Cancel() 我们的取消令牌源，以防止我们在找到符合我们标准的网址后开始处理任何新网址：

var finalBlock = new ActionBlock<string>(
        s => {
                cancellationSource.Cancel();
                //do whatever you want with the content string here            
            });

我们现在需要做的就是连接块并输入 URL：

downloadBlock.LinkTo(
    filterBlock,
    new DataflowLinkOptions
    {
        PropagateCompletion = true // this flag ensures that if the first block completes ALL urls it signals to the nect block that no more will come
    }
);
filterBlock.LinkTo(
    finalBlock,
    new DataflowLinkOptions
    {
        PropagateCompletion = true,
        MaxMessages = 1         //we are only interested in the first message that arrives in our final block
    });

downloadBlock.Post("http://url.com/blah1");
downloadBlock.Post("http://url.com/blah2");
// etc...

downloadBlock.Complete(); // this signals that we've ran out of Urls to process
finalBlock.Completion.Wait(); //now wait for any messages we have 'in progress' to work through the pipeline