标签: task-parallel-library

ParallelEnumerable有一个静态成员AsParallel.如果我有一个IEnumerable<T>并且想要使用Parallel.ForEach那意味着我应该一直使用AsParallel吗？

例如,这些都是正确的(其他一切都是平等的)吗？

没有AsParallel:

List<string> list = new List<string>();
Parallel.ForEach<string>(GetFileList().Where(file => reader.Match(file)), f => list.Add(f));

还是用AsParallel？

List<string> list = new List<string>();
Parallel.ForEach<string>(GetFileList().Where(file => reader.Match(file)).AsParallel(), f => list.Add(f));

如果已经被调用,ContinueWith(...)在a 上使用该方法是否安全？ TaskCompletionSource.TaskTaskCompletionSource.SetResult(...)

这个基本代码有望帮助构建问题:

// this was written inside the question box, please excuse any silly errors and lack of error checking (I'm not near VS right now)...

private WebClient _webClient = new WebClient();

public Task<string> GetExamplePage() {

    var tcs = new TaskCompletionSource<string>();

    web.DownloadStringCompleted += (s, ea) => tcs.SetResult(ea.Result);

    web.DownloadStringAsync(new URI(@"http://www.example.com/example.html"));

    return tcs.task;
}

public void ProcessExamplePage() {

    var task = GetExamplePage();

    Thread.Sleep(1000);

    task.ContinueWith(t => Console.WriteLine(t.Result)); // *line in question*
}

Console.WriteLine(...)如果WebClient.DownloadStringCompleted事件在task.ContinueWith设置之前已经被触发,那么执行吗？ …

我在ViewModel中有代码通过任务调用服务.任务完成后,它将填充ObservableCollection.问题是,它等待的任务,通过使用ContinueWith方法,并提供TaskScheduler.FromCurrentSynchronizationContext的任务调度,使OC获取UI线程上更新完成.

到目前为止一切都那么好,但是当谈到单元测试时,它会抛出一个异常,说"当前的SynchronizationContext可能不会被用作TaskScheduler".如果我在单元测试中使用模拟SynchronizationContext,那么ObservableCollection会抛出一个错误,因为它是根据调度程序线程进行更新的.

有什么方法可以解决这个问题吗？

谢谢.

我试图通过使用来加快我的计算时间Parallel.For.我有一个带有8个内核的英特尔酷睿i7 Q840 CPU,但与顺序for环路相比,我只能获得4的性能比.这是否可以获得Parallel.For,或者可以调整方法调用以提高性能？

这是我的测试代码,顺序:

var loops = 200;
var perloop = 10000000;

var sum = 0.0;
for (var k = 0; k < loops; ++k)
{
    var sumk = 0.0;
    for (var i = 0; i < perloop; ++i) sumk += (1.0 / i) * i;
    sum += sumk;
}

并行:

sum = 0.0;
Parallel.For(0, loops,
                k =>
                    {
                        var sumk = 0.0;
                        for (var i = 0; i < perloop; ++i) sumk += …

在Windows Phone 8中我有方法public async Task<bool> authentication().函数的返回类型是bool但当我尝试在if条件错误中使用其返回值时表示无法转换Task<bool>为bool.

public async Task<bool> authentication()
{
    var pairs = new List<KeyValuePair<string, string>>
    {
        new KeyValuePair<string, string> ("user", _username),
        new KeyValuePair<string, string> ("password", _password)
    };

    var serverData = serverConnection.connect("login.php", pairs);

    RootObject json = JsonConvert.DeserializeObject<RootObject>(await serverData);

    if (json.logined != "false")
    {
        _firsname = json.data.firsname;
        _lastname = json.data.lastname;
        _id = json.data.id;
        _phone = json.data.phone;
        _ProfilePic = json.data.profilePic;
        _thumbnail = json.data.thumbnail;
        _email = json.data.email;
        return true;
    }
    else
        return …

我必须编写一个名为Microsoft Dynamics CRM Web服务的控制台应用程序,以对超过八千个CRM对象执行操作.Web服务调用的细节是无关紧要的,这里没有显示,但我需要一个多线程客户端,以便我可以并行调用.我希望能够控制配置设置中使用的线程数,并且如果服务错误的数量达到配置定义的阈值,则应用程序也可以取消整个操作.

我使用任务并行库Task.Run和ContinueWith编写它,跟踪正在进行的调用(线程)数,我们收到了多少错误,以及用户是否已从键盘取消.一切都运行良好,我有大量的日志记录,以确保自己的线程干净整洁,并且在运行结束时一切都很整洁.我可以看到程序正在并行使用最大线程数,如果达到了我们的最大限制,则等到运行任务完成后再启动另一个.

在我的代码审查期间,我的同事建议最好使用async/await而不是任务和continuation,所以我创建了一个分支并以这种方式重写它.结果很有趣 - 异步/等待版本几乎慢两倍,并且它从未达到允许的并行操作/线程的最大数量.TPL总是并行获得10个线程,而async/await版本从未超过5个.

我的问题是:我在编写async/await代码(或甚至是TPL代码)方面犯了错误吗？如果我没有编写错误,你能解释为什么async/await效率较低,这是否意味着最好继续使用TPL进行多线程代码.

请注意,我测试过的代码实际上并没有调用CRM - CrmClient类只是在配置中指定的持续时间内线程休眠(5秒)然后抛出异常.这意味着没有可能影响性能的外部变量.

出于这个问题的目的,我创建了一个结合两个版本的精简程序; 调用哪个是由配置设置决定的.它们中的每一个都以一个引导运行器开始,它设置环境,创建队列类,然后使用TaskCompletionSource等待完成.CancellationTokenSource用于发信号通知用户取消.要处理的ID列表从嵌入文件中读取并推送到ConcurrentQueue.他们都开始调用StartCrmRequest和max-threads一样多次; 随后,每次处理结果时,ProcessResult方法再次调用StartCrmRequest,继续执行直到处理完所有ID.

您可以从这里克隆/下载完整的程序:https://bitbucket.org/kentrob/pmgfixso/

以下是相关配置:

<appSettings>
    <add key="TellUserAfterNCalls" value="5"/>
    <add key="CrmErrorsBeforeQuitting" value="20"/>
    <add key="MaxThreads" value="10"/>
    <add key="CallIntervalMsecs" value="5000"/>
    <add key="UseAsyncAwait" value="True" />
</appSettings>

从TPL版本开始,这里是启动队列管理器的bootstrap运行器:

public static class TplRunner
{
    private static readonly CancellationTokenSource CancellationTokenSource = new CancellationTokenSource();

    public static void StartQueue(RuntimeParameters parameters, IEnumerable<string> idList)
    {
        Console.CancelKeyPress += (s, args) =>
        {
            CancelCrmClient();
            args.Cancel = true;
        };

        var start = DateTime.Now;
        Program.TellUser("Start: " + start);

        var …

我的问题是基于一个真正的项目问题,但我从未使用过该System.Threading.Tasks库或执行任何涉及线程的严肃编程,因此我的问题可能是缺乏对特定库的了解以及对异步实际意味着什么的更普遍的误解.节目.

所以我的真实情况就是这个 - 我需要获取有关用户的数据.在我目前的情况下,它是财务数据,所以假设我需要所有人Accounts,所有人Deposits和所有人Consignations.在我的情况下,这意味着查询每个属性的数百万条记录,并且每个查询本身相对较慢,但是获取它Accounts比获取数据慢几倍Deposits.所以我已经为我要使用的三种银行产品定义了三个类,当我想获取某些用户的所有银行产品的数据时,我做了类似这样的事情:

List<Account> accounts = GetAccountsForClient(int clientId);
List<Deposit> deposits = GetDepositsForClient(int clientId);
List<Consignation> consignations = GetConsignationsForClient(int clientId);

所以问题从这里开始我需要同时获取所有这三个列表,因为我要将它们传递给我显示所有用户数据的视图.但正如现在一样,执行是同步的(如果我在这里正确使用这个术语),那么收集所有三种产品的数据的总时间是:

Total_Time = Time_To_Get_Accounts + Time_To_Get_Deposits + Time_To_Get_Consignations

这不好,因为每个查询都比较慢,所以总时间很长,但是,accounts查询比其他两个查询花费的时间要多得多,所以今天进入我脑海的想法是 - "如果我能执行怎么办？这同时查询".也许这是我对这个主题的最大误解,但对我而言,最接近这个想法是让它们异步,所以也许Total_Time不会是最慢查询的时间,但是会比所有三个查询的总和快得多.

由于我的代码很复杂,我创建了一个简单的用例,我认为,反映了我想要做的很好.我有两种方法:

public static async Task<int> GetAccounts()
{
    int total1 = 0;
    using (SqlConnection connection = new SqlConnection(connString))
    {
        string query1 = "SELECT COUNT(*) FROM [MyDb].[dbo].[Accounts]";
        SqlCommand command = new SqlCommand(query1, connection);
        connection.Open();
        for (int …

Currenly使用以下代码等待完成任务的集合.但是,我现在有一种情况,我希望能够通过取消令牌取消/中止WhenAll调用.我该怎么办呢？

  Dim TaskCollection As New List(Of Tasks.Task)
  For x As Integer = 1 To Threads
    Dim NewTask As Tasks.Task = TaskHandler.Delegates(DelegateKey).Invoke(Me, Proxies, TotalParams).ContinueWith(Sub() ThreadFinished())
    TaskCollection.Add(NewTask)
  Next

  Await Tasks.Task.WhenAll(TaskCollection)

我假设它会接下来的代码,但是我不确定'XXX'中会出现什么.

Await Tasks.Task.WhenAny(Tasks.Task.WhenAll(TaskCollection), XXX)

在我阅读.Net 4.5中的异步编程async和await关键字时,我在这里阅读以下段落

处理异步请求

在启动时看到大量并发请求或具有突发性负载(并发性突然增加)的Web应用程序中,使这些Web服务调用异步将提高应用程序的响应能力.异步请求与同步请求相同的处理时间.例如,如果请求进行需要两秒钟才能完成的Web服务调用,则无论是同步执行还是异步执行,请求都需要两秒钟.但是,在异步调用期间,在等待第一个请求完成时,不阻止线程响应其他请求.因此,当有许多并发请求调用长时间运行的操作时,异步请求会阻止请求排队和线程池增长.

对于粗体字,我无法理解异步请求如何像同步请求一样处理相同的时间？

例如:

public async Task MyMethod()
{
    Task<int> longRunningTask = LongRunningOperation();
    //indeed you can do independent to the int result work here 

    //and now we call await on the task 
    int result = await longRunningTask;
    //use the result 
    Console.WriteLine(result);
}

public async Task<int> LongRunningOperation() // assume we return an int from this long running operation 
{
    await Task.Delay(1000); //1 seconds delay
    return 1;
}

我所理解的是LongRunningOperation()从第一行调用开始执行并在调用后Task<int> …

我最近遇到过很多async方法,其中方法同步执行,但无论如何都返回一个Task,因此可以等待它们,例如

public virtual Task CreateAsync(TUser user)
{
    ThrowIfDisposed();
    if (user == null) throw new ArgumentNullException("user");
    Context.Save(user);
    Context.Flush();
    return Task.FromResult(0);
}

当然最好将可能长时间运行的操作分派给一个线程并返回仍处于活动状态的任务,以便真正等待:

public virtual Task CreateAsync(TUser user)
{
    ThrowIfDisposed();
    if (user == null) throw new ArgumentNullException("user");
    return Task.Run(() =>
    {
        Context.Save(user);
        Context.Flush();
    });
}

不过,我有点怀疑,只是脱掉TPL线程并不是最安全的做法.有关这两种不同模式的评论吗？