标签: task-parallel-library

在使用Parallel库时,我遇到了Task Factory的意外结果.

假设:任务工厂将随机处理呼叫

结果:每次都进行顺序处理(无论是相同的功能还是不同的功能)

    static void Do(string test)
    {
        Console.WriteLine("testttttt " + test);
    }

    static void Main(string[] args)
    {
        Task.Factory.StartNew(() =>
        {
            Do("1"); Do("2"); Do("3"); Do("4"); Do("5");
            Do("1"); Do("2"); Do("3"); Do("4"); Do("5");
            Do("1"); Do("2"); Do("3"); Do("4"); Do("5");
        });
        Console.ReadKey();
    }

以下代码生成随机结果:

        Task.Factory.StartNew(() =>
        {
            Do("1"); Do("2"); Do("3"); Do("4"); Do("5");
        });
        Task.Factory.StartNew(() =>
        {
            Do("1"); Do("2"); Do("3"); Do("4"); Do("5");
        });
        Task.Factory.StartNew(() =>
        {
            Do("1"); Do("2"); Do("3"); Do("4"); Do("5");
        });

为什么第一个例子总是产生连续的结果,这应该被认为是理所当然的所有时间？

当我从类I下面运行方法DoStuff()时,我有时会得到一个异常:

The process cannot access the file because it is being used by another process.

为什么会发生？我认为ReaderWriterLockSlim应该解决这个潜在的问题？

public class Test
{
    private ReaderWriterLockSlim lock_ = new ReaderWriterLockSlim();
    public Test()
    {

    }

    public void DoStuff()
    {
        while (true)
        {
            Task[] tasks = new Task[5];
            for (int i = 0; i < 5; i++)
            {
                tasks[i] = Task.Factory.StartNew(() =>
                {
                        lock_.EnterReadLock();
                        try
                        {
                            File.AppendAllText("test.txt", "test");
                        }
                        finally
                        {
                            lock_.ExitReadLock();
                        }
                    }
                });
            }
            Task.WaitAll(tasks);
            Thread.Sleep(5000);
        }
    }
}

在继续执行之前如何等待上一个方法完成？我觉得这很容易,但事实并非如此.即使我已经阅读了很多例子,我也必须做一些非常愚蠢的事情.在下面的代码中,我不能让GetDocVM()方法执行,直到AddUserDocuments()方法完成.为什么？因为GetDocVM() 不会引入刚刚添加的记录.我继承了这段代码,并试图改进它.

ut.ModelJSON = await Task.Run(() => _userTransactionService.ConvertToModelJson(typeof(UserDocument).Name, "", transactionDocs)).ConfigureAwait(false);
var taskReturnsVoid = Task.Run(() => _genericUploadService.AddUserDocuments(ut, docs));
List<GenericUploadDocumentViewModel> viewModel = new List<GenericUploadDocumentViewModel>();
await taskReturnsVoid.ContinueWith((t) =>
           {
                 viewModel = GetDocVM();//I EXPECTED THIS TO WAIT TO BE EXECUTED
           });
return Json(viewModel, JsonRequestBehavior.AllowGet);  //GETTING HERE TOO SOON

在下面的测试代码中，我期待这个结果：

1, 2000
2, 4000
3, 6000

然而实际的结果是：

3, 6000    
2, 4000
1, 2000

此外，我只能在 6 秒后在屏幕上看到结果。这意味着任何被竞争的输入都在等待并处理到下一个阶段。

如何使管道在完成后立即吐出每个输入的结果？

    public static void Run()
    {
        Func<int, string> fn = n =>
        {
            var sleep = n * 2000;
            Thread.Sleep(sleep);
            return n + ", " + sleep;
        };

        var opts = new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 4
        };

        var transformBlock = new TransformBlock<int, string>(fn, opts);
        var bufferBlock = new BufferBlock<string>(opts);

        transformBlock.LinkTo(bufferBlock, new DataflowLinkOptions { PropagateCompletion = true });

        for (var i = 3; …

我有以下方法(简称为简称):

public Task CreateAsync(TUser user)
{
    using (var connection = new SqlConnection(_connection))
    {
        return Task.FromResult(connection.Execute("CreateUser", param, commandType: CommandType.StoredProcedure));
    }
 }

我想合并一个try-catch block,所以我可以记录任何潜在的Sql错误.

public Task CreateAsync(TUser user)
{
     var result = ???; // what is the return type here?
     try
     {
         result = FromResult(connection.Execute("CreateUser", param, commandType: CommandType.StoredProcedure));
      }
      catch(SqlException sqlEx)
      {
          // log error here
       }

      return result;
}

我想我不确定返回类型Task是什么？

我有很多行（300k+）要在尽可能短的时间内更新插入到 SQL Server 数据库中，所以想法是使用并行化和分区数据，并使用异步将数据泵入 SQL，当时的 X 线程，每个上下文 100 行，上下文被回收以最大限度地减少跟踪开销。然而，这意味着要并行使用多个连接，因此CommittableTransaction将TransactionScope使用分布式事务，这将导致并行事务征用操作返回臭名昭著的"This platform does not support distributed transactions."异常。

我确实需要能够提交/回滚整组更新插入。它是批量上传过程的一部分，任何错误都应将更改回滚到以前的工作/稳定条件（应用程序方面）。

我有什么选择？缺少使用一个连接且没有并行化？

注意：问题并不像一批插入命令那么简单，如果是这样的话，我只会生成插入并在服务器上作为查询运行它们，或者确实使用SqlBulkCopy. 其中大约一半是更新，一半是插入，其中 SQL Server 生成新键，需要获取这些键并在接下来要插入的子对象上重新设置键，行分布在 3 级层次结构中的大约 12 个表中。

我有2个等待我运行从外部服务获取数据:

aaa= await Gateway.GetMyAAA();
bbb= await Gateway.GetBBBB();

我希望两者同时运行,然后当两者都完成时,继续.

我如何等待他们？

假设我有两个任务,具有以下要求:

1. 两者都是异步的.
2. 两者并行运行
3. 在其中一个完成的那一刻,我需要知道哪一个做了.

我提出了以下代码,但它只是在两个任务开始后挂起(WaitAny函数永远不会返回).我在Run函数下也得到了一条波浪线,告诉我在其中添加await,但当我尝试在Task.WaitAny前面添加它时,VS会抱怨.我应该在另一个任务中包装WaitAny吗？我究竟做错了什么？

async void Run()
{
    Task task1 = Task1();
    Task task2 = Task2();

    int completedTaskIdx = Task.WaitAny(task1, task2);

    Debug.WriteLine("completedTaskIdx = {0}", completedTaskIdx.ToString());
}

async Task Task1()
{
    Debug.WriteLine("Task 1 Start");
    await Task.Delay(5000);
    Debug.WriteLine("Task 1 Stop");
}

async Task Task2()
{
    Debug.WriteLine("Task 2 Start");
    await Task.Delay(10000);
    Debug.WriteLine("Task 2 Stop");
}

我有一组需要一次又一次并行处理的项目.

我可以做这个:

while (true)
{
   Parallel.ForEach(sources, source =>
   {
      // do lots of work with source
   })
}

但问题是,如果一个源比其他源需要更长的时间,它将有效地挂起while循环,我想重新处理列表中的每个项目,而无需等待其他项目完成.

我怎么能做到这一点？为每个拥有循环的源代码启动任务,还是有更优雅的方式来执行此操作？

我在ASP.Net Web表单页面中具有以下代码，该页面基本上检查高速缓存中是否有某些值，如果没有，它将调用获取数据并将其存储在高速缓存中的方法。获取数据的方法如下

ChartRenderingHelper.GenerateBidsStatusCreated(currYear.ToString(), currQuarter.ToString(), currYearType.ToString())

使用EF调用存储的proc，这3个调用全部调用单独的SP。现在，我按顺序执行操作，因此，如果每个SP花费5秒，则总操作需要15。确定如何更改当前代码以执行此操作。

  bidsCreated.Value = DashboardCacheHelper.IsIncache(bidsCreatedKey, useCaching) ? DashboardCacheHelper.GetFromCache(bidsCreatedKey) : (string)DashboardCacheHelper.SaveCache(bidsCreatedKey, JsonConvert.SerializeObject(ChartRenderingHelper.GenerateBidsStatusCreated(currYear.ToString(), currQuarter.ToString(), currYearType.ToString())), DateTime.Now.AddDays(cacheDays));
            bidsSubmitted.Value = DashboardCacheHelper.IsIncache(bidsSubmittedKey, useCaching) ? DashboardCacheHelper.GetFromCache(bidsSubmittedKey) : (string)DashboardCacheHelper.SaveCache(bidsSubmittedKey, JsonConvert.SerializeObject(ChartRenderingHelper.GenerateBidsStatusSubmitted(currYear.ToString(), currQuarter.ToString(), currYearType.ToString())), DateTime.Now.AddDays(cacheDays));
            bidsClosed.Value = DashboardCacheHelper.IsIncache(bidsClosedKey, useCaching) ? DashboardCacheHelper.GetFromCache(bidsClosedKey) : (string)DashboardCacheHelper.SaveCache(bidsClosedKey, JsonConvert.SerializeObject(ChartRenderingHelper.GenerateBidsStatusClosed(currYear.ToString(), currQuarter.ToString(), currYearType.ToString())), DateTime.Now.AddDays(cacheDays));

如何并行执行这3个作业？使用TPL，我知道我们可以并行运行方法

Parallel.Invoke(() => DoSomeWork(), () => DoSomeOtherWork());

这是推荐的方法吗？如果我说需要并行运行的12个操作，那么使用EF的所有调用SQL存储的Procs都会导致性能问题。