c# - 异步但不并行

Question

我有一个定义异步操作的契约(Contract)，但我是从未设置为异步的区域调用它的。我想同时进行多个调用，所以我从这个开始:

var tasks = inputs.Select(input => service.GetResult(input));
var results = tasks.WhenAll(tasks).Result;

我认为这会并行启动所有调用，然后在第二行等待。但是，查看目标服务的日志记录，我发现调用是连续传入的。

我找到了 this article这显示了一种类似的调用方法，并解释说它不一定并行运行，所以我只是将其切换为直接并行调用以进行测试:

var results = new ConcurrentBag<Result>();
Parallel.ForEach(inputs, input => results.Add(service.GetResult(input).Result));

这按预期工作 - 我可以看到调用是并行到达服务的。

所以，这让我想到了两个问题:

1) 使用选项 2 的缺点是什么？
2) 如何让选项 1 正常工作？

这里有几个服务可以重现这个问题。例如，使用 WCFTestClient 和四个整数(1、2、3、4)的列表调用 ClientService。 (运行时可能需要更改端口。)

目标服务:

using System.Diagnostics;
using System.ServiceModel;
using System.Threading.Tasks;

namespace AsyncNotParallel
{
    [ServiceBehavior(InstanceContextMode = InstanceContextMode.PerCall, ConcurrencyMode = ConcurrencyMode.Single)]
    public class TargetService : ITargetService
    {
        public async Task<int> GetResult(int input)
        {
            Trace.WriteLine($"In:  {input}");
            await Task.Delay(1000); // Do stuff.
            Trace.WriteLine($"Out: {input}");
            return input;
        }
    }
    [ServiceContract]
    public interface ITargetService
    {
        [OperationContract]
        Task<int> GetResult(int input);
    }
}

客户服务:

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.ServiceModel;
using System.Threading.Tasks;

namespace AsyncNotParallel
{
    [ServiceBehavior(InstanceContextMode = InstanceContextMode.PerCall, ConcurrencyMode = ConcurrencyMode.Single)]
    public class ClientService : IClientService
    {
        public int GetResults(List<int> inputs)
        {
            // Option 1:
            var tasks = inputs.Select(input => Execute((ITargetService service) => service.GetResult(input)));
            var results1 = Task.WhenAll(tasks).Result.Sum();

            // Option 2:
            var bag = new ConcurrentBag<int>();
            Parallel.ForEach(inputs, input => bag.Add(Execute((ITargetService service) => service.GetResult(input)).Result));
            var results2 = bag.Sum();

            return results1 + results2;
        }

        private TResult Execute<TService, TResult>(Func<TService, TResult> operation)
        {
            var address = new EndpointAddress("http://localhost:34801/TargetService.svc");
            var binding = new BasicHttpBinding();
            var factory = new ChannelFactory<TService>(binding, address);
            var channel = factory.CreateChannel();
            var result = operation(channel);
            ((IClientChannel)channel).Close();
            return result;
        }
    }

    [ServiceContract]
    public interface IClientService
    {
        [OperationContract]
        int GetResults(List<int> inputs);
    }
}

Answer 1

如果您的服务的 ConcurrencyMode 是 Single(默认 - 可以在 ServiceBehavior 属性中覆盖)，这意味着 服务按顺序处理调用。所以这两个选项实际上是按顺序执行的，只是第二个选项得到了无序的结果。您可以切换到 ConcurrencyMode.Multiple，这更危险，因为这意味着必须仔细编写服务才能保证线程安全。

1. Parallel 针对 CPU 密集型操作进行了优化，并将根据系统中的内核数量对您的调用进行并行处理。实际上，您可以更多地并行化 IO 绑定(bind)操作，因此整个操作会执行得更慢。此外，您在每个线程上都使用了 .Result，在每个 Parallel 产生的任务中浪费了等待时间。我不会使用这种方法。最后，ConcurrentBag 是无序的，这对您来说可能重要也可能不重要。

2. 在第一个选项中，您将从 UI 线程按顺序启动每个 WCF 调用。这很可能会导致 ConcurrencyMode.Single 服务按照列表的相同顺序处理调用。

您可能应该使用 Task.WaitAll() 而不是 Task.WhenAll().Result。不过，我强烈反对您不要在 UI 线程上这样做。这是许多讨厌的 UI 挂起的根本原因。您可以简单地从一个同步方法启动一个异步方法(无需 Wait() ) - 只是触发并忘记。等待任务后，只需在异步方法中根据需要更新 UI。

最后一个建议 - 在使用同一 channel 进行多个并发调用之前，您应该Open() 它以获得更好的性能。尽管 channel 会自动执行此操作，但由于某些 channel 会锁定 channel ，所以这里有一个好处。

编辑 -

看到更新后的代码后，问题是您正在启动一个任务，然后等待 channel 同步关闭(在调用完成之前阻塞)。这是一个更好的实现:

private async Task<TResult> Execute<TService, TResult>(Func<TService, Task<TResult>> operation)
{
    var address = new EndpointAddress("http://localhost:34801/A");
    var binding = new BasicHttpBinding();
    var channel = ChannelFactory<TService>.CreateChannel(binding, address);
    var clientChannel = (IClientChannel)channel;
    try
    {
        var result = await operation(channel).ConfigureAwait(false);
        return result;
    }
    finally
    {
        if (clientChannel.State != CommunicationState.Faulted)
        {
            await Task.Factory.FromAsync(clientChannel.BeginClose, clientChannel.EndClose, null).ConfigureAwait(false);
        }
        else if (clientChannel.State != CommunicationState.Closed)
        {
            clientChannel.Abort();
        }
    }
}

我还修改了它以使用缓存的 ChannelFactory，并正确关闭和中止 channel 。