הפעלת משימות אסינכרוניות במקביל

הושלם

תכנות מקביל הוא טכניקה רבת עוצמה המאפשרת לך לבצע משימות מרובות בו זמנית, ולשפר את הביצועים וההיענות של היישומים שלך.

ב-C#, אתה יכול להשתמש בספריית המשימות המקבילה (TPL) כדי לפשט את תהליך כתיבת הקוד המקביל. ה-TPL הוא קבוצה של סוגים ציבוריים וממשקי API במרחבי השמות System.Threading ו System.Threading.Tasks -. מטרת ה-TPL היא להפוך את המפתחים לפרודוקטיביים יותר על ידי פישוט התהליך של הוספת מקביליות ומקביליות ליישומים. ה-TPL משנה באופן דינמי את מידת המקביליות כדי להשתמש בכל המעבדים הזמינים בצורה היעילה ביותר. בנוסף, ה-TPL מטפל בחלוקת העבודה, תזמון הליכי משנה ב-ThreadPool, תמיכה בביטולים, ניהול מצבים ופרטים אחרים ברמה נמוכה. על ידי שימוש ב-TPL, אתה יכול למקסם את ביצועי הקוד שלך תוך התמקדות בעבודה שהתוכנית שלך נועדה לבצע.

ה-TPL מספק תמיכה בתחומים הבאים:

• מקביליות נתונים: ה-TPL מספק שיטות לביצוע מקביליות נתונים, ומאפשר לך לבצע את אותה פעולה על מספר רכיבי נתונים בו זמנית. אפשרות זו שימושית במיוחד כשיש לך ערכות נתונים גדולות וברצונך לבצע חישובים או טרנספורמציות על כל רכיב בנפרד.
• תכנות אסינכרוני מבוסס משימות: ה-TPL מספק את המחלקה Task , המייצגת פעולה אסינכרונית. באפשרותך להשתמש במילות המפתח async ו await - כדי לפשט את תהליך כתיבת הקוד האסינכרוני. זה מאפשר לך לכתוב קוד שקל יותר לקרוא ולתחזק תוך ניצול המקביליות.
• תזרים נתונים: ה- TPL מספק מודל תכנות תזרים נתונים המאפשר לך ליצור צינורות עיבוד נתונים מורכבים. מודל זה מבוסס על הרעיון של "בלוקים" שיכולים לעבד נתונים באופן אסינכרוני ולתקשר זה עם זה באמצעות הודעות.

חשוב

תכנות מקבילי וריבוי הליכי משנה הם נושאים מתקדמים הדורשים הבנה טובה של בו-זמנית וסינכרון. למרות שה- TPL מפשט תרחישים מרובי הליכי משנה, מומלץ שתהיה לך הבנה בסיסית של מושגי הליכי השחלה, לדוגמה, נעילה, מבוי סתום ותנאי מירוץ, כדי שתוכל להשתמש ב- TPL ביעילות. הכשרה זו מספקת מבוא מוגבל לתכנות מקבילי באמצעות TPL.

תקבולת נתונים

מקביליות נתונים היא סוג של תכנות מקבילי המתמקד בביצוע אותה פעולה על מספר רכיבי נתונים בו זמנית. אפשרות זו שימושית במיוחד כשיש לך ערכות נתונים גדולות וברצונך לבצע חישובים או טרנספורמציות על כל רכיב בנפרד. ב-C#, אתה יכול להשתמש בשיטות Parallel.For ו Parallel.ForEach כדי להשיג הקבלה נתונים בקלות. שיטות אלה מאפשרות לך לחזור על אוספים או טווחי נתונים במקביל, ולהפיץ את עומס העבודה על פני מספר הליכי משנה.

הספרייה המקבילה למשימות תומכת במקביליות נתונים דרך הכיתה System.Threading.Tasks.Parallel . מחלקה זו מספקת מימושים מקבילים מבוססי שיטה של for ולולאות foreach . אתה כותב את לוגיקת הלולאה עבור לולאה או Parallel.ForParallel.ForEach בדומה לכתיבה של לולאה רציפה. ה-TPL מטפל בכל העבודה ברמה נמוכה עבורך.

דוגמת הקוד הבאה מציגה לולאת foreach פשוטה והמקבילה המקבילה שלה.

// Sequential version
foreach (var item in sourceCollection)
{
    Process(item);
}

// Parallel equivalent
Parallel.ForEach(sourceCollection, item => Process(item));

ה- TPL מספק גם קבוצה של מבני נתונים הממוטבים לגישה בו-זמנית, כגון ConcurrentBag, ConcurrentQueueו ConcurrentDictionary- . מבני נתונים אלה מאפשרים לך להוסיף, להסיר ולגשת בבטחה לרכיבים ממספר שרשורים ללא צורך בנעילה מפורשת.

דוגמת הקוד הבאה מדגימה כיצד להשתמש ב- כדי ConcurrentBag לאחסן תוצאות ממשימות מרובות הפועלות במקביל:

using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        var results = new ConcurrentBag<int>();
        Parallel.For(0, 100, i =>
        {
            // Simulate some work
            Task.Delay(100).Wait();
            results.Add(i);
        });

        Console.WriteLine($"Processed {results.Count} items in parallel.");
    }
}

בדוגמה זו, ה ConcurrentBag - משמש לאחסון תוצאות העיבוד המקביל. כל משימה מוסיפה את התוצאה שלה לתיק ללא צורך במנעולים מפורשים, מה שמבטיח את בטיחות החוט.

שימוש Task.WhenAll וביצוע Task.WhenAny משימות במקביל

המתודות Task.WhenAll ו Task.WhenAny הן חלק מהספרייה המקבילה למשימות ב-C#. שיטות אלה מאפשרות לך להפעיל משימות מרובות במקביל ולהמתין להשלמתן.

Task.WhenAll משמש כאשר ברצונך להמתין להשלמת כל המשימות לפני שתמשיך. הוא לוקח מערך של משימות כקלט ומחזיר משימה בודדת המייצגת את השלמת כל משימות הקלט. זה שימושי כשיש לכם כמה משימות עצמאיות שאפשר לבצע בו-זמנית, כמו ביצוע קריאות API מרובות או עיבוד של כמה קבצים בו-זמנית.

Task.WhenAny משמש כאשר ברצונך להמתין להשלמת כל אחת מהמשימות. הוא לוקח מערך של משימות כקלט ומחזיר משימה המייצגת את המשימה הראשונה שהושלמה. אפשרות זו שימושית כאשר ברצונך לבצע פעולה כלשהי מיד עם סיום כל אחת מהפעילויות, מבלי להמתין להשלמת כולן.

דוגמת הקוד הבאה מדגימה כיצד להשתמש כדי Task.WhenAll להריץ משימות מרובות במקביל ולהמתין להשלמתן:

using System;
using System.Net.Http;
using System.Threading.Tasks;
using System.Collections.Generic;

class Program
{
    static async Task Main(string[] args)
    {
        var urls = new List<string>
        {
            "https://example.com",
            "https://example.org",
            "https://example.net"
        };

        var tasks = new List<Task<string>>();

        foreach (var url in urls)
        {
            tasks.Add(FetchDataAsync(url));
        }

        // Wait for all tasks to complete
        var results = await Task.WhenAll(tasks);

        foreach (var result in results)
        {
            Console.WriteLine(result);
        }
    }

    static async Task<string> FetchDataAsync(string url)
    {
        using (var client = new HttpClient())
        {
            return await client.GetStringAsync(url);
        }
    }
}

בדוגמה זו, פעולת השירות FetchDataAsync מביאה נתונים ממספר כתובות URL במקביל באמצעות Task.WhenAll. התוצאות מודפסות על הקונסולה לאחר השלמת כל המשימות.

ביצוע פעולות קלט/פלט מרובות של קבצים בו-זמנית

במקרים רבים, איטרציה של קבצים היא פעולה שניתן להקביל בקלות.

הדוגמה הבאה חוזרת על הספריות ברצף, אך מעבדת את הקבצים במקביל. זו כנראה הגישה הטובה ביותר כאשר יש לך יחס קובץ לתיקייה גדול. אפשר גם להקביל את איטרציית הספריות, ולגשת לכל קובץ ברצף. זה כנראה לא יעיל להקביל את שתי הלולאות אלא אם כן אתה מכוון ספציפית למכונה עם מספר רב של מעבדים. עם זאת, כמו בכל המקרים, עליך לבדוק את הבקשה שלך ביסודיות כדי לקבוע את הגישה הטובה ביותר.

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Security;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        try
        {
            TraverseTreeParallelForEach(@"C:\Program Files", (f) =>
            {
                // Exceptions are no-ops.
                try
                {
                    // Do nothing with the data except read it.
                    byte[] data = File.ReadAllBytes(f);
                }
                catch (FileNotFoundException) { }
                catch (IOException) { }
                catch (UnauthorizedAccessException) { }
                catch (SecurityException) { }
                // Display the filename.
                Console.WriteLine(f);
            });
        }
        catch (ArgumentException)
        {
            Console.WriteLine(@"The directory 'C:\Program Files' does not exist.");
        }

        // Keep the console window open.
        Console.ReadKey();
    }

    public static void TraverseTreeParallelForEach(string root, Action<string> action)
    {
        //Count of files traversed and timer for diagnostic output
        int fileCount = 0;
        var sw = Stopwatch.StartNew();

        // Determine whether to parallelize file processing on each folder based on processor count.
        int procCount = Environment.ProcessorCount;

        // Data structure to hold names of subfolders to be examined for files.
        Stack<string> dirs = new Stack<string>();

        if (!Directory.Exists(root))
        {
            throw new ArgumentException(
                "The given root directory doesn't exist.", nameof(root));
        }
        dirs.Push(root);

        while (dirs.Count > 0)
        {
            string currentDir = dirs.Pop();
            string[] subDirs = { };
            string[] files = { };

            try
            {
                subDirs = Directory.GetDirectories(currentDir);
            }
            // Thrown if we do not have discovery permission on the directory.
            catch (UnauthorizedAccessException e)
            {
                Console.WriteLine(e.Message);
                continue;
            }
            // Thrown if another process has deleted the directory after we retrieved its name.
            catch (DirectoryNotFoundException e)
            {
                Console.WriteLine(e.Message);
                continue;
            }

            try
            {
                files = Directory.GetFiles(currentDir);
            }
            catch (UnauthorizedAccessException e)
            {
                Console.WriteLine(e.Message);
                continue;
            }
            catch (DirectoryNotFoundException e)
            {
                Console.WriteLine(e.Message);
                continue;
            }
            catch (IOException e)
            {
                Console.WriteLine(e.Message);
                continue;
            }

            // Execute in parallel if there are enough files in the directory.
            // Otherwise, execute sequentially.Files are opened and processed
            // synchronously but this could be modified to perform async I/O.
            try
            {
                if (files.Length < procCount)
                {
                    foreach (var file in files)
                    {
                        action(file);
                        fileCount++;
                    }
                }
                else
                {
                    Parallel.ForEach(files, () => 0,
                        (file, loopState, localCount) =>
                        {
                            action(file);
                            return (int)++localCount;
                        },
                        (c) =>
                        {
                            Interlocked.Add(ref fileCount, c);
                        });
                }
            }
            catch (AggregateException ae)
            {
                ae.Handle((ex) =>
                {
                    if (ex is UnauthorizedAccessException)
                    {
                        // Here we just output a message and go on.
                        Console.WriteLine(ex.Message);
                        return true;
                    }
                    // Handle other exceptions here if necessary...

                    return false;
                });
            }

            // Push the subdirectories onto the stack for traversal.
            // This could also be done before handing the files.
            foreach (string str in subDirs)
                dirs.Push(str);
        }

        // For diagnostic purposes.
        Console.WriteLine($"Processed {fileCount} files in {sw.ElapsedMilliseconds} milliseconds");
    }
}

בדוגמה זו, קלט/פלט של הקובץ מתבצע באופן סינכרוני. כאשר הקוד שלך פועל עם קבצים גדולים או חיבורי רשת איטיים, ייתכן שעדיף לגשת לקבצים באופן אסינכרוני. ניתן לשלב טכניקות קלט/פלט אסינכרוניות עם איטרציה מקבילה.

הדוגמה משתמשת במשתנה fileCount המקומי כדי לשמור ספירה של המספר הכולל של הקבצים המעובדים. מכיוון שניתן לגשת למשתנה בו-זמנית על-ידי משימות מרובות, הגישה מסונכרנת על-ידי קריאה לשיטה Interlocked.Add .

שים לב שאם מופיעה חריגה בהליך המשנה הראשי, הליכי המשנה המופעלים על-ידי פעולת השירות ForEach עשויים להמשיך לפעול. כדי לעצור את הליכי המשנה האלה, אתה יכול להגדיר משתנה בוליאני במטפלי החריגים שלך, ולבדוק את הערך שלו בכל איטרציה של הלולאה המקבילית. אם הערך מציין שבוצעה חריגה, השתמש במשתנה ParallelLoopState כדי לעצור או להתנתק מהלולאה.

סיכום

יחידה זו התמקדה בתקבולת ובספריית המשימות המקבילות (TPL). הוא מכסה כיצד להריץ משימות אסינכרוניות במקביל באמצעות השיטה Task.WhenAll , מקביליות נתונים עם Parallel.For ושיטות Parallel.ForEach , ושימוש במבני נתונים בו-זמניים כמו ConcurrentBag, ConcurrentQueueו ConcurrentDictionary- . התוכן גם מדגים כיצד לבצע פעולות קלט/פלט מרובות של קבצים בו-זמנית.

נקודות עיקריות

• תכנות מקביל ב-C# מאפשר ביצוע משימות מרובות בו זמנית.
• הכיתה Task ומילות async המפתח await משמשות ליישום תכנות מקביל.
• Task.WhenAll השיטה משמשת להמתנה להשלמת משימות מרובות לפני שתמשיך.
• מקביליות נתונים מושגת באמצעות Parallel.For ושיטות Parallel.ForEach .
• מבני נתונים בו-זמניים כגון ConcurrentBag, ConcurrentQueue, וממוטבים ConcurrentDictionary לגישה בו-זמנית.
• Task.WhenAll ושיטות Task.WhenAny מאפשרות להריץ מספר משימות במקביל ולהמתין להשלמתן.
• ניתן לבצע פעולות קלט/פלט מרובות של קבצים בו-זמנית.