المهام Task في C#: Task.Run وWhenAll وWhenAny والإلغاء والخيوط

Task كائن يمثّل عملًا قيد التنفيذ. قد يعمل على خيط من مجمّع الخيوط، أو ينتظر مؤقّتًا، أو ينتظر ردًّا من الشبكة؛ وفي كل الأحوال يمكنك أن تسأل هل انتهى، وتحصل على نتيجته، وتنتظره، أو تلحق به عملًا آخر. async وawait صياغة اللغة فوق المهام. وتغطي هذه الصفحة واجهة Task نفسها، والخيوط.

Task.Run: العمل على مجمّع الخيوط

تأخذ Task.Run تعبير lambda وتضعه في طابور مجمّع الخيوط، وهو مجموعة خيوط تحتفظ بها .NET كي لا تنشئ خيطًا جديدًا لكل مهمة. وتعيد Task، أو Task<T> إن كان تعبير lambda يعيد قيمة.

using System;
using System.Threading.Tasks;

class Program
{
    static int CountPrimes(int max)
    {
        int count = 0;
        for (int n = 2; n <= max; n++)
        {
            bool prime = true;
            for (int d = 2; d * d <= n; d++)
            {
                if (n % d == 0) { prime = false; break; }
            }
            if (prime) count++;
        }
        return count;
    }

    static void Main()
    {
        Task<int> task = Task.Run(() => CountPrimes(200000));
        Console.WriteLine("Started, completed yet: " + task.IsCompleted);

        int primes = task.Result;   // blocks until the task finishes
        Console.WriteLine("Primes up to 200,000: " + primes);
        Console.WriteLine("Status: " + task.Status);
    }
}

المخرجات:

Started, completed yet: False
Primes up to 200,000: 17984
Status: RanToCompletion

تحجز Result الخيط المستدعي حتى تنتهي المهمة. هذا مقبول في Main برنامج كونسول، لكن داخل الشيفرة غير المتزامنة ستكتب int primes = await task;، التي تنتظر دون حجز. وينطبق الأمر نفسه على task.Wait() لـ Task بلا نتيجة.

Task.Run للعمل كثيف المعالجة: حسابات تبقي نواة مشغولة. أما للإدخال والإخراج، مثل قراءة ملف أو استدعاء واجهة API على الويب، فاستخدم الدالة غير المتزامنة التي توفرها المكتبة أصلًا (File.ReadAllTextAsync وHttpClient.GetStringAsync) وانتظرها مباشرة بـ await. فأثناء انتظار الإدخال والإخراج لا يحتجز الاستدعاء غير المتزامن المنتظَر أي خيط، لذا لا يضيف تغليفه في Task.Run شيئًا سوى مرور إضافي عبر مجمّع الخيوط. أما ما يشغل خيطًا من المجمّع فعلًا فهو Task.Run حول استدعاء متزامن حاجب مثل File.ReadAllText: يبقى ذلك الخيط خاملًا حتى يستجيب القرص أو الشبكة.

عدة مهام معًا: WhenAll

قسّم العمل المستقل إلى مهام، وابدأها كلها، وادمج النتائج بـ Task.WhenAll. على جهاز بعدة أنوية تعمل الأجزاء بالتوازي.

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

class Program
{
    static long SumRange(int from, int to)
    {
        long sum = 0;
        for (int i = from; i < to; i++) sum += i % 7;
        return sum;
    }

    static void Main()
    {
        int chunks = 4, size = 5000000;

        Task<long>[] parts = Enumerable.Range(0, chunks)
            .Select(c => Task.Run(() => SumRange(c * size, (c + 1) * size)))
            .ToArray();

        long[] results = Task.WhenAll(parts).Result;
        Console.WriteLine("Parts: " + string.Join(", ", results));
        Console.WriteLine("Total: " + results.Sum());
    }
}

المخرجات:

Parts: 14999995, 14999999, 15000003, 15000000
Total: 59999997

كل جزء يعمل على نطاقه الخاص ويعيد مجموعه الخاص، فلا تتشارك المهام شيئًا ولا تحتاج إلى أقفال. وتعيد WhenAll النتائج بترتيب المهام، أيًّا كان ترتيب انتهاء الخيوط. وحين تحتاج المهام إلى تحديث حالة مشتركة راجع lock.

WhenAny والمهل الزمنية

تكتمل Task.WhenAny حين تكتمل أولى مهامها وتعيد تلك المهمة. مسابقة العمل الحقيقي ضد Task.Delay تعطي مهلة زمنية:

using System;
using System.Threading.Tasks;

class Program
{
    static async Task<string> QueryServerAsync(string name, int ms)
    {
        await Task.Delay(ms);
        return name + " answered after " + ms + " ms";
    }

    static async Task RunAsync()
    {
        Task<string> fast = QueryServerAsync("fast mirror", 100);
        Task<string> slow = QueryServerAsync("main server", 800);
        Task<string> first = await Task.WhenAny(fast, slow);
        Console.WriteLine(await first);

        Task<string> work = QueryServerAsync("report service", 2000);
        Task timeout = Task.Delay(300);
        if (await Task.WhenAny(work, timeout) == timeout)
        {
            Console.WriteLine("Timed out after 300 ms");
        }
    }

    static void Main()
    {
        RunAsync().GetAwaiter().GetResult();
    }
}

المخرجات:

fast mirror answered after 100 ms
Timed out after 300 ms

تعيد WhenAny المهمة الفائزة نفسها، لا نتيجتها، فتنتظرها بـ await مجددًا لتحصل على القيمة (أو الاستثناء إن فشلت). لاحظ أن المهام الخاسرة تستمر في العمل؛ لا توقفها WhenAny. ولإيقاف العمل البطيء فعلًا ألغه برمز كما يظهر أدناه. في .NET 6 وما بعده تعبّر await work.WaitAsync(TimeSpan.FromMilliseconds(300)) عن المهلة نفسها في استدعاء واحد.

الاستكمالات بـ ContinueWith

قبل await كانت طريقة تشغيل شيفرة بعد انتهاء مهمة هي ContinueWith. ما زلت ستراها في الشيفرات الأقدم:

using System;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        Task<int> load = Task.Run(() => 21);

        Task<string> report = load
            .ContinueWith(t => t.Result * 2)
            .ContinueWith(t => "Report total: " + t.Result);

        Console.WriteLine(report.Result);
    }
}

المخرجات:

Report total: 42

يتلقى كل استكمال المهمة السابقة ويُنفَّذ حين تكتمل. تفعل await الشيء نفسه بتكلّف أقل بكثير، وتعالج الاستثناءات بشكل طبيعي، وتعود إلى السياق الصحيح، ففضّلها في الشيفرة الجديدة. وتُنفَّذ ContinueWith حتى حين تكون المهمة السابقة قد فشلت أو أُلغيت، فعلى الشيفرة الأقدم فحص t.IsFaulted أو تمرير TaskContinuationOptions لتجنّب استدعاء .Result على مهمة فاشلة.

الإلغاء بـ CancellationToken

تُلغى المهام بشكل تعاوني. يصدر CancellationTokenSource رمزًا؛ والشيفرة التي تؤدي العمل تفحص الرمز وتوقف نفسها. لا شيء يُجهض قسرًا.

using System;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static async Task<int> ProcessOrdersAsync(int count, CancellationToken token)
    {
        int done = 0;
        for (int i = 0; i < count; i++)
        {
            token.ThrowIfCancellationRequested();   // stop between items
            await Task.Delay(100, token);           // pass the token on
            done++;
        }
        return done;
    }

    static async Task RunAsync()
    {
        using (var cts = new CancellationTokenSource())
        {
            cts.CancelAfter(350);                   // cancel in 350 ms
            try
            {
                int n = await ProcessOrdersAsync(10, cts.Token);
                Console.WriteLine("Processed " + n);
            }
            catch (OperationCanceledException)
            {
                Console.WriteLine("Cancelled before finishing all 10 orders");
            }
        }

        using (var cts = new CancellationTokenSource())
        {
            int n = await ProcessOrdersAsync(3, cts.Token); // never cancelled
            Console.WriteLine("Processed " + n);
        }
    }

    static void Main()
    {
        RunAsync().GetAwaiter().GetResult();
    }
}

المخرجات:

Cancelled before finishing all 10 orders
Processed 3

يُمرَّر الرمز عبر كل طبقة، وتقبله أيضًا واجهات API غير المتزامنة مثل Task.Delay وHttpClient واستدعاءات قواعد البيانات، فيوقفها الإلغاء في منتصف الانتظار. التقط OperationCanceledException؛ فـ TaskCanceledException التي ترميها بعض الواجهات مشتقة منها. وCancellationTokenSource من نوع IDisposable، ومن هنا using.

Thread مقابل Task

Thread خيط من نظام التشغيل تنشئه بنفسك. هو أثقل: لكل خيط مكدّسه الخاص (1 MB افتراضيًا على Windows)، ويستغرق بدؤه وقتًا ملموسًا، ولا يمنحك نتيجة ولا معالجة استثناءات.

using System;
using System.Threading;

class Program
{
    static void Main()
    {
        int result = 0;
        var worker = new Thread(() =>
        {
            Thread.Sleep(100);
            result = 6 * 7;
        });
        worker.Start();
        Console.WriteLine("Thread started");
        worker.Join();          // wait for it to finish
        Console.WriteLine("Result: " + result);
    }
}

المخرجات:

Thread started
Result: 42

لا تلجأ إلى Thread إلا حين تحتاج إلى شيء لا يمنحه المجمّع: حلقة طويلة العمر يجب ألا تستعير خيطًا من المجمّع أبدًا، أو أولوية معينة، أو شقة أحادية الخيط لـ COM. وللمهام طويلة العمر تحصل Task.Factory.StartNew(work, TaskCreationOptions.LongRunning) على خيط مخصص مع الإبقاء على واجهة المهام.

Parallel.For وParallel.ForEach

لـ "افعل هذا بكل عنصر، مستخدمًا كل الأنوية" تكون System.Threading.Tasks.Parallel أبسط من إنشاء المهام يدويًا. تقسّم النطاق على الخيوط وتعود حين تنتهي كل التكرارات:

using System;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        int[] sizes = { 120, 340, 90, 410, 275, 60 };
        long totalPixels = 0;

        Parallel.For(0, sizes.Length, i =>
        {
            long pixels = (long)sizes[i] * sizes[i];      // "process" image i
            Interlocked.Add(ref totalPixels, pixels);      // thread-safe add
            Console.WriteLine("image " + i + " on thread " + Thread.CurrentThread.ManagedThreadId);
        });

        Console.WriteLine("Total pixels: " + totalPixels);
    }
}

مخرجات على سبيل المثال:

image 4 on thread 7
image 3 on thread 6
image 5 on thread 8
image 0 on thread 1
image 1 on thread 4
image 2 on thread 5
Total pixels: 385425

تخرج أسطر كل صورة بأي ترتيب نفّذتها به الخيوط، وتختلف معرّفات الخيوط من تشغيل لآخر؛ المجموع وحده ثابت. هذه طبيعة الحلقات المتوازية: يجب ألا تعتمد التكرارات على ترتيب بعضها، وأي حالة مشتركة تحتاج إلى تزامن (Interlocked.Add هنا). وتفعل Parallel.ForEach الشيء نفسه على أي مجموعة، وتضيف .NET 6 الدالة Parallel.ForEachAsync للأجسام غير المتزامنة.

أخطاء شائعة

• .Result أو .Wait() داخل شيفرة غير متزامنة. تحجز خيطًا وقد تسبّب الجمود في تطبيقات الواجهات؛ استخدم await بدلًا منهما.
• التقاط نوع الاستثناء الأصلي حول .Result. تصل الإخفاقات مغلّفة في AggregateException؛ وawait تفكّ تغليفها.
• أطلق وانسَ. Task.Run(...) دون انتظار المهمة أو تخزينها يضيّع استثناءاتها.
• Task.Run حول الإدخال والإخراج. حول دالة غير متزامنة لا يضيف سوى مرور إضافي عبر مجمّع الخيوط، وحول دالة حاجبة مثل File.ReadAllText يحتجز خيطًا من المجمّع طوال مدة الانتظار. انتظر دالة الإدخال والإخراج غير المتزامنة مباشرة.
• تجاهل الرمز. تمرير CancellationToken لا يفعل شيئًا ما لم يفحصه العمل.
• حالة مشتركة غير متزامنة في الشيفرة المتوازية. استخدم Interlocked أو lock، أو امنح كل مهمة بياناتها الخاصة.