c# - 并行任务与委托(delegate)

Question

我现在需要决定如何构建我的异步代码。我需要从我的函数 Func1() 中一次调用 20 个不同的 Web 服务，当它们都返回一个 xml 答案时，将所有结果加入一个大的 xml。

我考虑过使用 TPL 任务。像这样:

var task = Task.Factory.StartNew (call the web service1...);
var task2 = Task.Factory.StartNew (call the web service2...);
var task3 = Task.Factory.StartNew (call the web service3...);

task.WaitAll();

听起来不错还是有更好的方法来完成工作？

Answer 1

我们几个月前需要这样的东西，以同时处理多个远程 URL。我们通过从 SemaphoreSlim 派生我们自己的类来实现这一点类(class)。

你可以这样实现:

/// <summary>
    /// Can be used to process multiple URL's concurrently.
    /// </summary>
    public class ConcurrentUrlProcessor : SemaphoreSlim
    {
        private int initialCount;
        private int maxCount;
        private readonly HttpClient httpClient;

        /// <summary>
        /// Initializes a new instance of the <see cref="T:System.Threading.SemaphoreSlim" /> class, specifying the initial number of requests that can be granted concurrently.
        /// </summary>
        /// <param name="initialCount">The initial number of requests for the semaphore that can be granted concurrently.</param>
        public ConcurrentUrlProcessor(int initialCount)
            :base(initialCount)
        {
            this.initialCount = initialCount;
            this.maxCount = int.MaxValue;
            this.httpClient = new HttpClient();
        }
        /// <summary>
        /// Initializes a new instance of the <see cref="T:System.Threading.SemaphoreSlim" /> class, specifying the initial and maximum number of requests that can be granted concurrently.
        /// </summary>
        /// <param name="initialCount">The initial number of requests for the semaphore that can be granted concurrently.</param>
        /// <param name="maxCount">The maximum number of requests for the semaphore that can be granted concurrently.</param>
        public ConcurrentUrlProcessor(int initialCount, int maxCount)
            : base(initialCount, maxCount) 
        {
            this.initialCount = initialCount;
            this.maxCount = maxCount;
            this.httpClient = new HttpClient();
        }
        /// <summary>
        /// Starts the processing.
        /// </summary>
        /// <param name="urls">The urls.</param>
        /// <returns>Task{IEnumerable{XDocument}}.</returns>
        public virtual async Task<IEnumerable<XDocument>> StartProcessing(params string[] urls)
        {
            List<Task> tasks = new List<Task>();
            List<XDocument> documents = new List<XDocument>();

            SemaphoreSlim throttler = new SemaphoreSlim(initialCount, maxCount);
            foreach (string url in urls)
            {
                await throttler.WaitAsync();

                tasks.Add(Task.Run(async () =>
                {
                    try
                    {
                        string xml = await this.httpClient.GetStringAsync(url);

                        //move on to the next page if no xml is returned.
                        if (string.IsNullOrWhiteSpace(xml))
                            return;

                        var document = XDocument.Parse(xml);
                        documents.Add(document);
                    }
                    catch (Exception)
                    {
                        //TODO: log the error or do something with it.
                    }
                    finally
                    {
                        throttler.Release();
                    }
                }));
            }

            await Task.WhenAll(tasks);

            return documents;
        }
    }

以及相应的单元测试:

    [Test]
    public async void CanProcessMultipleUrlsTest()
    {
        string[] urls = new string[] 
        {
            "http://google.nl",
            "http://facebook.com",
            "http://linkedin.com",
            "http://twitter.com" 
        };

        IEnumerable<XDocument> documents = null;
        ConcurrentUrlProcessor processor = new ConcurrentUrlProcessor(100);

        documents = await processor.StartProcessing(urls);

        Assert.AreEqual(4, documents.Count());
    }