具有 WCF 服务参考异步功能的任务计划程序

Question

我正在尝试使用服务引用，使用任务调度程序同时发出多个请求。该服务包括一个返回结果集的同步和异步函数。我有点困惑，我有几个初步的问题，然后我将分享我在每个问题上的进展情况。我正在使用一些日志记录、并发可视化工具和提琴手进行调查。最终，我想使用反应式调度程序来发出尽可能多的请求。

1) 我应该使用异步函数来发出所有请求吗？

2) 如果我要在多个任务中使用同步功能，那么可能会导致我的线程数不足的有限资源是什么？

这是我目前所拥有的：

var myScheduler = new myScheduler();
var myFactory = new Factory(myScheduler); 
var myClientProxy = new ClientProxy(); 
var tasks = new List<Task<Response>>();
foreach( var request in Requests )
{
    var localrequest = request;
    tasks.Add( myFactory.StartNew( () =>
        {
            // log stuff
            return  client.GetResponsesAsync( localTransaction.Request );
            // log some more stuff
        }).Unwrap() );
}

Task.WaitAll( tasks.ToArray() );
// process all the requests after they are done

这会运行，但根据提琴手的说法，它只是尝试一次完成所有请求。它可能是调度程序，但我相信比我做上述更多。

我也尝试在没有 unwrap 命令的情况下实现它，而是使用异步等待委托，它做同样的事情。我也尝试过引用 .result ，这似乎是按顺序进行的。使用调度程序/工厂的非同步服务函数调用，每个客户端最多只能同时获得大约 20 个同时请求。

Answer 1

1. 是的。它可以让您的应用程序通过使用更少的线程来完成更多任务来更好地扩展。
2. 线程。当您启动一个本质上是异步的同步操作时（例如I/O），您有一个阻塞线程等待操作完成。但是，您可以同时使用此线程来执行 CPU 密集型操作。

限制并发请求数量的最简单方法是使用SemaphoreSlim，它允许asynchronously wait 输入：

async Task ConsumeService()
{
    var client = new ClientProxy();
    var semaphore = new SemaphoreSlim(100);
    var tasks = Requests.Select(async request =>
    {
        await semaphore.WaitAsync();
        try
        {
            return await client.GetResponsesAsync(request);
        }
        finally
        {
            semaphore.Release();
        }
    }).ToList();
    await Task.WhenAll(tasks);

    // TODO: Process responses...
}

Answer 2

无论您如何调用 WCF 服务，无论是异步调用还是同步调用，您都将受到 WCF serviceThrottling 限制的约束。您应该查看这些设置并可能将它们调整得更高（如果由于某种原因将它们设置为较低的值），在 .NET4 中默认值非常好，但是在旧版本的 .NET 框架中，这些默认值要保守得多比 .NET4。

.NET 4.0

• MaxConcurrentSessions：默认为 100 * ProcessorCount
• MaxConcurrentCalls：默认为 16 * ProcessorCount
• MaxConcurrentInstances：默认为 MaxConcurrentCalls+MaxConcurrentSessions

Answer 3

1.) 是的。 2.)是的。

如果您想控制同时请求的数量，可以尝试使用 Stephen Toub 的 ForEachAsyncmethod。它允许您控制同时处理多少个任务。

public static class Extensions
{
    public static Task ForEachAsync<T>(this IEnumerable<T> source, int dop, Func<T, Task> body)
    { 
        return Task.WhenAll( 
            from partition in Partitioner.Create(source).GetPartitions(dop) 
            select Task.Run(async delegate { 
                using (partition) 
                    while (partition.MoveNext()) 
                        await body(partition.Current); 
            })); 
    }
}

void Main()
{
    var myClientProxy = new ClientProxy(); 
    var responses = new List<Response>();

    // Max 10 concurrent requests
    Requests.ForEachAsync<Request>(10, async (r) =>
    {
        var response = await client.GetResponsesAsync( localTransaction.Request );
        responses.Add(response);
    }).Wait();      
}