标签: task-parallel-library

我一直在尝试对我的 CPU 绑定函数使用异步方法来计算一些聚合函数。问题是存在一些死锁（我想），因为计算时间太不同了。我是这个任务并行世界的新手，我也阅读了 Stephem Cleary 的文章，但我仍然不确定这种异步方法的各个方面。我的代码：

private static void Main(string[] args)
{
    PIServer server = ConnectToDefaultPIServer();
    AFTimeRange timeRange = new AFTimeRange("1/1/2012", "6/30/2012");
    Program p = new Program();
    for (int i = 0; i < 10; i++)
    {
        p.TestAsynchronousCall(server, timeRange);
        //p.TestAsynchronousCall(server, timeRange).Wait();-same results
    }
    Console.WriteLine("Main check-disconnected done");
    Console.ReadKey();
}

private async Task TestAsynchronousCall(PIServer server, AFTimeRange timeRange)
{
    AsyncClass asyn;
    for (int i = 0; i < 1; i++)
    {
        asyn = new AsyncClass();
        await asyn.DoAsyncTask(server, timeRange);
        //asyn.DoAsyncTask(server, timeRange);-same results
    }
} …

我正在编写一个 UI 应用程序。如果发生某些异常，我需要在应用程序退出之前做一些工作。所以我订阅了AppDomain.CurrentDomain.UnhandledException事件。

AppDomain.CurrentDomain.UnhandledException += HandleException;

HandleException 方法执行异步保存到远程（因为没有同步 api）。

private void HandleException(object sender, UnhandledExceptionEventArgs args)
{
    foreach (var user in UsersCollection.ToArray())
    {
        try
        {
            foreach (var session in user.Sessions.ToArray())
            {
                try
                {
                    SaveSessionAsync(session).Wait();
                }
                catch (Exception e)
                {
                    Logger.Error("Can't save session: " + session, e);
                }
            }
        }
        catch (Exception e)
        {
            Logger.Error("Can't save sessions of user " + user, e);
        }
    }
}

在应用程序退出之前，我需要确保我保存了所有会话（至少尝试过）。但是如果我把Wait()它放在那里，我就会陷入僵局，应用程序永远不会停止。据我所知，await在正常情况下可能会帮助我（当我处于 UI 线程但不在应用程序终止状态时），但await不会等待应用程序退出。所以我的保存任务可能会中止。但我需要他们完成。

有没有一种方法可以在SaveSessionAsync不造成死锁的情况下等待有保证的任务完成？

顺便说一句：SaveSessionAsync 里面有 …

我正在尝试使用生产者/消费者模式来实现，BlockingCollection<T>所以我编写了一个简单的控制台应用程序来测试它。

public class Program
{
    public static void Main(string[] args)
    {
        var workQueue = new WorkQueue();
        workQueue.StartProducingItems();
        workQueue.StartProcessingItems();

        while (true)
        {

        }
    }
}

public class WorkQueue
{
    private BlockingCollection<int> _queue;
    private static Random _random = new Random();

    public WorkQueue()
    {
        _queue = new BlockingCollection<int>();

        // Prefill some items.
        for (int i = 0; i < 100; i++)
        {
            //_queue.Add(_random.Next());
        }
    }

    public void StartProducingItems()
    {
        Task.Run(() =>
        {
            _queue.Add(_random.Next()); // Should be adding items to the queue …

如果我有一个异步方法:

async Task Process()
{
    while (condition)
    {
        await ...;
    }
}

我需要安排执行此方法.所以我使用:

Task.Run(Process);

我是否需要保留方法Task返回的对象的引用,Task.Run以确保Process执行完成？

这里有更多的上下文:我需要在我的应用程序中创建大量(大约50K)任务队列.所以我想创建一个无线程队列处理设计,当队列为空时(因为大多数队列都是这样),它没有成本(除了内存).

样本类的要点:https://gist.github.com/hemant-jangid/e990b27507596c086e5651f504d0521f

我需要使用wait运算符在c#中多次执行一个方法async.以下是我的代码

Task.Factory.StartNew(() => SentSMS(null, creation.SenderMobile, messageToSender));
Task.Factory.StartNew(() => SentSMS(null, creation.ReceiverMobile, messageToReciver));
Task.Factory.StartNew(() => SentSMS(null, empMobile, messageToEmp));

private async Task SentSMS(SMSReceived smsrecived, string mobilenumber ,string message)
{
    try
    {
        sent = new SMSSent();
        sent.MobileNumber = Util.ParseMobileNo(mobilenumber);
        sent.MobileOperator = Util.GetMobileOperator(mobilenumber);
        sent.QueryCodeId = 1;
        sent.ReplyByTelcoID = 1;
        sent.ReplyText = message;
        sent.ReplyByTelcoID = 1;
        sent.SMSReceivedId = (smsrecived == null ? 0 : smsrecived.ID);
        sent.SMSSentDate = DateTime.Now;
        sent.Status = 1;
        sentBL.Save(sent);
    }
    catch (Exception)
    {

    }
}

问题是,当最后一个请求开始时,它会覆盖其余2个记录.执行此代码后,我已在数据库中检查它保存一次/相同的记录3次.

不要告诉我,我传递相同的数据三次.所有的价值都是不同的.我需要帮助.如果我缺少某些地方,请告诉我.我也尝试过Task.Run().我不想使用等待.等待,结果也是一样的

屏幕截图附有数据库

我试图理解 System.Task，但我不确定我所做的是否正确。我的目标是编写一种方法来批量并行处理图像。

我的理解是否正确，如果batchSize = 3该方法将排队 3 个任务并尽快并行运行所有 3 个任务Task.WaitAll()在调用时？

如果这样做有没有更优雅的方法？

private static void ProcessImages(int batchSize)
{
      List<Task> tasks = new List<Task>();
      foreach (var image in ImageSource.ReadImages())
      {
            if(tasks.Count < batchSize)
            { 
                 tasks.Add(Task.Run(() => ImageProcessor.ProcessImage(image)));
            }
            else
            {
                 Task.WaitAll(tasks.ToArray());
                 tasks.Clear();
            }
      }
}

我使用HttpClient,我想写这样的东西:

HttpClient client = ...;
Task<string> getContent()
{
    return client.GetAsync(...)
   .ContinueWith( t => t.Result.Content.ReadAsStringAsync() );
}

我知道我可以写

.ContinueWith( t => t.Result.Content.ReadAsStringAsync().Result);

但池中的一个线程将被阻止.我想在Google Task库中使用 continueWithTask .

我怎么能完成它？

UPDATE

是的,我确实需要使用Tasks而不是async/await,我真的知道我想要什么.

UPDATE2

我修改了我的观点,现在我认为选择技术时我错了.如果有人怀疑,这里有一个很好的代码的好例子.

这是我的C#功能

public async Task GetAttendance(IEnumerable<Zone> zones)
{
    try
    {
        foreach (var zone in zones)
        {
            var req = new AttendeeRequestTO(zone.StartTime, zone.EndTime, zone.ZoneId.ToString(), accessToken);

            //THROWING COMPILE TIME ERROR
            zone.AttendanceCount = Task.Factory.StartNew(() => _vClient.GetAttendeesCount(req));
        }
    }
    catch (Exception ex)
    {
    }
}

错误

错误 CS0029 无法将类型“System.Threading.Tasks.Task>”隐式转换为“int?”

我不想为每个任务应用等待，因为每个任务都是独立的，我希望每个任务在自己的上下文中运行，而无需等待任何其他任务。

我的意思是
任务T1-转到API，获取值并设置计数
任务T2-转到API，获取值并设置计数
任务T3-转到API，获取值并设置计数

T2 不应等待 T1 完成，T3 不应等待 T2 完成等等。

如何将每个任务的输出值分配给zone.AttendanceCount？

以下代码块 await loader.Completion;

我只是不知道为什么？

using System;
using System.Linq;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace tests {
    class Program {
        static async Task Main(string[] args) {
            Planner pl = new Planner();
            Console.WriteLine(await pl.Count());
        }
    }

    public class Planner {
        private TransformBlock<int, string[]> loader;
        private int _im = 0;

        public Planner(int im = 5) {
            _im = im;
            loader =
                new TransformBlock<int, string[]>(
                    async i => {
                        Console.WriteLine(i);
                        await Task.Delay(1000);
                        return new string[] { i.ToString() };
                    }
               );
        }

        public async Task<long> Count() …

我需要并行运行三个异步I / O操作，尤其是数据库调用。因此，我编写了以下代码：

// I need to know whether these tasks started running here
var task1 = _repo.GetThingOneAsync();
var task2 = _repo.GetThingTwoAsync();
var task3 = _repo.GetThingThreeAsync();

// await the results
var task1Result = await task1;
var task2Result = await task2;
var task3Result = await task3;

这些GetThingOneAsync(), GetThingTwoAsync(), GetThingThreeAsync()方法彼此非常相似，只不过它们具有不同的返回类型（Task<string>, Task<int>, Task<IEnumerable<int>>）。下面是数据库调用之一的示例：

public async Task<IEnumerable<int>> GetThingOneAsync(string code)
{
    return await db.DrType.Where(t => t.Code == code).Select(t => t.IdType).ToListAsync();
}

在调试模式下，我可以看到var task1 = _repo.GetThingOneAsync();开始运行GetThingOneAsync()异步方法（与其他两个任务相同）。

我的同事说，_repo.GetThingOneAsync() …