streamalism.org

1. סקירה כללית

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

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

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

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

2. ForkJoinPool

ה ForkJoinPool הוא לב המסגרת. זהו יישום של שירות ExecutorService שמנהל אשכולות עובדים ומספק לנו כלים לקבל מידע על מצב הבריכה והביצועים.

חוטי עובדים יכולים לבצע רק משימה אחת בכל פעם, אך את ForkJoinPool לא יוצר שרשור נפרד לכל תת-משימת יחיד. במקום זאת, לכל חוט בבריכה יש תור כפול משלו (או דק, מבוטא) סִיפּוּן) המאחסן משימות.

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

2.1. עבודה בגניבת אלגוריתם

במילים פשוטות - חוטים חופשיים מנסים "לגנוב" עבודה מדקים של חוטים תפוסים.

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

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

2.2. ForkJoinPool מיידיות

בג'אווה 8, הדרך הנוחה ביותר לקבל גישה למופע של ForkJoinPool היא להשתמש בשיטה הסטטית שלה בריכה נפוצה (). כשמו כן הוא, זה יספק התייחסות לבריכה המשותפת, המהווה מאגר חוטים המוגדר כברירת מחדל לכל אחד ForkJoinTask.

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

ForkJoinPool commonPool = ForkJoinPool.commonPool ();

את אותה התנהגות ניתן להשיג ב- Java 7 על ידי יצירת a ForkJoinPool והקצאתו לא סטטי ציבורי שדה של מחלקת שירות:

ForkJoinPool fork static forkJoinPool = ForkJoinPool חדש (2);

עכשיו ניתן לגשת אליו בקלות:

ForkJoinPool forkJoinPool = PoolUtil.forkJoinPool;

עם ForkJoinPool's בונים, ניתן ליצור מאגר חוטים מותאם אישית עם רמה מסוימת של מקבילות, מפעל חוטים ומטפל חריג. בדוגמה לעיל, לבריכה יש רמת מקבילות של 2. המשמעות היא שבריכה תשתמש בשתי ליבות מעבד.

3. ForkJoinTask

ForkJoinTask הוא סוג הבסיס למשימות שמבוצעות בפנים ForkJoinPool. בפועל, יש להרחיב את אחת משתי מחלקות המשנה שלה: ה- רקורסיבית אקשן ל בָּטֵל משימות ו משימת רקורסיביות למשימות שמחזירות ערך.לשניהם יש שיטה מופשטת לְחַשֵׁב() בו מוגדרת ההיגיון של המשימה.

3.1. RecursiveAction - דוגמא

בדוגמה שלהלן, יחידת העבודה לעיבוד מיוצגת על ידי חוּט שקוראים לו עומס עבודה. למטרות הדגמה, המשימה היא שטותית: היא פשוט מקדשת את הקלט ורושמת אותה.

כדי להדגים את התנהגות המזלגות של המסגרת, הדוגמה מחלקת את המשימה אם עומס עבודה.אורך() גדול מסף מוגדרמשתמש ב createSubtask () שיטה.

המחרוזת מחולקת רקורסיבית למתחים, יוצרת CustomRecursiveTask מקרים שמבוססים על משטחים אלה.

כתוצאה מכך, השיטה מחזירה א רשימה.

הרשימה מוגשת ל ForkJoinPool משתמש ב להפעיל את כל () שיטה:

המחלקה הציבורית CustomRecursiveAction מרחיבה את RecursiveAction {private String workload = ""; סופי סטטי פרטי int THRESHOLD = 4; לוגר לוגר סטטי פרטי = Logger.getAnonymousLogger (); Public CustomRecursiveAction (עומס עבודה מחרוזת) {this.workload = עומס עבודה; } @ מחשב ריק מוגן @Override () {if (workload.length ()> THRESHOLD) {ForkJoinTask.invokeAll (createSubtasks ()); } אחר {עיבוד (עומס עבודה); }} רשימה פרטית createSubtasks () {List subtasks = ArrayList new (); מחרוזת partOne = workload.substring (0, workload.length () / 2); מחרוזת partTwo = workload.substring (workload.length () / 2, workload.length ()); subtasks.add (חדש CustomRecursiveAction (partOne)); subtasks.add (חדש CustomRecursiveAction (partTwo)); להחזיר משימות משנה; } עיבוד חלל פרטי (מחרוזת עבודה) {מחרוזת תוצאה = work.toUpperCase (); logger.info ("תוצאה זו - (" + תוצאה + ") - עובדה על ידי" + Thread.currentThread (). getName ()); }}

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

3.2. משימת רקורסיביות

עבור משימות שמחזירות ערך, ההיגיון כאן דומה, אלא שהתוצאה עבור כל משימה משנה מאוחדת בתוצאה אחת:

המחלקה הציבורית CustomRecursiveTask מרחיב את RecursiveTask {private int [] arr; סופי סטטי פרטי int THRESHOLD = 20; CustomRecursiveTask ציבורי (int [] arr) {this.arr = arr; } @Override מוגן מחשב שלם () {אם (arrllength> THRESHOLD) {return ForkJoinTask.invokeAll (createSubtasks ()) .stream () .mapToInt (ForkJoinTask :: join) .sum (); } אחר {עיבוד חוזר (arr); }} אוסף פרטי createSubtasks () {List sharedTasks = ArrayList new (); splitTasks.add (משימה מותאמת אישית חדשה (Arrays.copyOfRange (arr, 0, arrlength / 2))); splitTasks.add (משימה CustomRecursiveTime חדשה (Arrays.copyOfRange (arr, arrlength / 2, arrllength))); משימות מחולקות } עיבוד פרטי שלם שלם (int [] arr) {return Arrays.stream (arr) .filter (a -> a> 10 && a a * 10) .sum (); }}

בדוגמה זו, העבודה מיוצגת על ידי מערך המאוחסן ב- arr שדה של CustomRecursiveTask מעמד. ה createSubtasks () השיטה מחלקת את המשימה רקורסיבית לפיסות עבודה קטנות יותר עד שכל יצירה קטנה מהסף. אז ה להפעיל את כל () השיטה מגישה את משימות המשנה לבריכה המשותפת ומחזירה רשימה של עתיד.

כדי להפעיל ביצוע, לְהִצְטַרֵף() שיטה נקראת לכל משימה משנה.

בדוגמה זו, הדבר נעשה באמצעות Java 8 ממשק API של זרם; ה סְכוּם() השיטה משמשת כייצוג של שילוב תוצאות משנה לתוצאה הסופית.

4. הגשת משימות ל ForkJoinPool

כדי להגיש משימות למאגר האשכולות, ניתן להשתמש בגישות מעטות.

ה שלח() אוֹ לבצע()שיטה (מקרי השימוש שלהם זהים):

forkJoinPool.execute (customRecursiveTask); int result = customRecursiveTask.join ();

ה לעורר()השיטה מזייפת את המשימה ומחכה לתוצאה, ואינה זקוקה להצטרפות ידנית:

int result = forkJoinPool.invoke (customRecursiveTask);

ה להפעיל את כל () השיטה היא הדרך הנוחה ביותר להגיש רצף של ForkJoinTasks אל ה ForkJoinPool. זה לוקח משימות כפרמטרים (שתי משימות, var args או אוסף), מזלגות ואז מחזיר אוסף של עתיד חפצים לפי סדר הפקתם.

לחלופין, אתה יכול להשתמש בנפרד מזלג() ו לְהִצְטַרֵף() שיטות. ה מזלג() השיטה מגישה משימה לבריכה, אך היא אינה מפעילה את ביצועה. ה לְהִצְטַרֵף() יש להשתמש בשיטה למטרה זו. במקרה של רקורסיבית אקשן, ה לְהִצְטַרֵף() לא מחזיר דבר מלבד ריק; ל משימת רקורסיביות, הוא מחזיר את תוצאת ביצוע המשימה:

customRecursiveTaskFirst.fork (); תוצאה = customRecursiveTaskLast.join ();

בשלנו משימת רקורסיביות דוגמא השתמשנו ב- להפעיל את כל () שיטה להגיש רצף של משימות משנה לבריכה. ניתן לעשות את אותה עבודה מזלג() ו לְהִצְטַרֵף()אף על פי שיש לכך השלכות על סדר התוצאות.

כדי למנוע בלבול, זה בדרך כלל רעיון טוב להשתמש להפעיל את כל () שיטה להגיש יותר ממשימה אחת ל ForkJoinPool.

5. מסקנות

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

• השתמש בכמה שפחות בריכות שרשור - ברוב המקרים, ההחלטה הטובה ביותר היא להשתמש במאגר שרשורים אחד לכל יישום או מערכת
• השתמש במאגר האשכולות המשותף המוגדר כברירת מחדל, אם אין צורך בכוונון ספציפי
• השתמש בסף סביר לפיצול ForkJoinTask למשימות משנה
• הימנע מכל חסימה שלךForkJoinTasks

הדוגמאות המשמשות במאמר זה זמינות במאגר GitHub המקושר.