streamalism.org

1. הקדמה

מאמר זה הוא מבט על מאגרי חוטים בג'אווה - החל מהיישומים השונים בספריית הג'אווה הרגילה ואז התבוננות בספריית גויאבה של גוגל.

2. בריכת החוטים

ב- Java, אשכולות ממופים לשרשורים ברמת המערכת המהווים את המשאבים של מערכת ההפעלה. אם אתה יוצר שרשורים ללא שליטה, אתה עלול להיגמר מהמשאבים הללו במהירות.

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

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

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

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

3. בריכות הברגה בג'אווה

3.1. מוציאים לפועל, מוציא להורג ו שירות ExecutorService

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

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

ה מוציא להורג יש ממשק יחיד לבצע שיטה להגיש ניתן לרוץ מקרים לביצוע.

הנה דוגמה מהירה כיצד ניתן להשתמש ב- מוציאים לפועל ממשק API לרכישת מוציא להורג מופע מגובה על ידי מאגר שרשורים יחיד ותור בלתי מוגבל לביצוע משימות ברצף. כאן אנו מבצעים משימה אחת שמדפיסה בפשטות "שלום עולם" על המסך. המשימה מוגשת כלמבה (תכונה של Java 8) שמוסכם שהיא ניתן לרוץ.

מוציא לפועל = Executors.newSingleThreadExecutor (); executor.execute (() -> System.out.println ("שלום עולם"));

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

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

ExecutorService executorService = Executors.newFixedThreadPool (10); עתיד עתידי = executorService.submit (() -> "שלום עולם"); // כמה פעולות מחרוזת תוצאה = future.get ();

כמובן שבתרחיש אמיתי בדרך כלל אינך רוצה להתקשר future.get () מיד אבל דחו את הקריאה לזה עד שתזדקקו בפועל לערך החישוב.

ה שלח השיטה עמוסה מדי כדי לקחת את אחת מהן ניתן לרוץ אוֹ ניתן להתקשר שניהם ממשקים פונקציונליים וניתן להעבירם כמבדות (החל מ- Java 8).

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

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

לקבלת דוגמאות נוספות לשימוש ב- שירות ExecutorService ממשק וחוזים עתידיים, עיין ב"מדריך לשירות Java ExecutorService ".

3.2. ThreadPoolExecutor

ה ThreadPoolExecutor הוא יישום מאגר חוטים הניתן להרחבה עם הרבה פרמטרים ווים לכוונון עדין.

פרמטרי התצורה העיקריים שנדון כאן הם: corePoolSize, maximumPoolSize, ו keepAliveTime.

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

ה keepAliveTime פרמטר הוא מרווח הזמן שלפיו השרשורים המוגזמים (מיישרים מעבר ל corePoolSizeמותר להתקיים במצב סרק. כברירת מחדל, ה- ThreadPoolExecutor רואה רק בחוטים שאינם ליבה להסרה. על מנת להחיל את אותה מדיניות ההסרה על נושאי הליבה, אנו יכולים להשתמש ב- allowCoreThreadTimeOut (נכון) שיטה.

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

לדוגמה, newFixedThreadPool שיטה יוצרת א ThreadPoolExecutor עם שווה corePoolSize ו maximumPoolSize ערכי פרמטר ואפס keepAliveTime. פירוש הדבר שמספר האשכולות במאגר האשכולות הזה תמיד זהה:

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool (2); executor.submit (() -> {Thread.sleep (1000); return null;}); executor.submit (() -> {Thread.sleep (1000); return null;}); executor.submit (() -> {Thread.sleep (1000); return null;}); assertEquals (2, executor.getPoolSize ()); assertEquals (1, executor.getQueue (). size ());

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

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

עוד מוגדר מראש ThreadPoolExecutor ניתן ליצור עם Executors.newCachedThreadPool () שיטה. שיטה זו אינה מקבלת מספר שרשורים כלל. ה corePoolSize מוגדר למעשה ל- 0, וה- maximumPoolSize נקבע ל מספר שלם.MAX_VALUE למשל. ה keepAliveTime הוא 60 שניות עבור זה.

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

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newCachedThreadPool (); executor.submit (() -> {Thread.sleep (1000); return null;}); executor.submit (() -> {Thread.sleep (1000); return null;}); executor.submit (() -> {Thread.sleep (1000); return null;}); assertEquals (3, executor.getPoolSize ()); assertEquals (0, executor.getQueue (). size ());

גודל התור בדוגמה לעיל תמיד יהיה אפס מכיוון שבפנים א SynchronousQueue נעשה שימוש במופע. ב SynchronousQueue, זוג של לְהַכנִיס ו לְהַסִיר פעולות תמיד מתרחשות בו זמנית, כך שהתור אף פעם לא מכיל דבר.

ה Executors.newSingleThreadExecutor () API יוצר צורה אופיינית אחרת של ThreadPoolExecutor המכיל חוט יחיד. מבצע הברגה היחידה אידיאלי ליצירת לולאת אירועים. ה corePoolSize ו maximumPoolSize הפרמטרים שווים ל- 1, ו- keepAliveTime הוא אפס.

משימות בדוגמה שלעיל יבוצעו ברצף, כך שערך הדגל יהיה 2 לאחר השלמת המשימה:

מונה AtomicInteger = חדש AtomicInteger (); ExecutorService executor = Executors.newSingleThreadExecutor (); executor.submit (() -> {counter.set (1);}); executor.submit (() -> {counter.compareAndSet (1, 2);});

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

3.3. ScheduledThreadPoolExecutor

ה ScheduledThreadPoolExecutor מרחיב את ThreadPoolExecutor בכיתה וגם מיישם את ScheduledExecutorService ממשק למספר שיטות נוספות:

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

ה Executors.newScheduledThreadPool () בדרך כלל משתמשים בשיטה ליצירת ScheduledThreadPoolExecutor עם נתון corePoolSize, ללא גבולות maximumPoolSize ואפס keepAliveTime. כך תזמן משימה לביצוע תוך 500 אלפיות השנייה:

ScheduledExecutorService executor = Executors.newScheduledThreadPool (5); executor.schedule (() -> {System.out.println ("שלום עולם");}, 500, TimeUnit.MILLISECONDS);

הקוד הבא מראה כיצד לבצע משימה לאחר עיכוב של 500 אלפיות השנייה ואז לחזור עליה כל 100 אלפיות השנייה. לאחר תזמון המשימה, אנו ממתינים עד שהיא יורה שלוש פעמים באמצעות ה- CountDownLatch לנעול, ואז בטל אותו באמצעות Future.cancel () שיטה.

מנעול CountDownLatch = CountDownLatch חדש (3); ScheduledExecutorService executor = Executors.newScheduledThreadPool (5); ScheduledFuture future = executor.scheduleAtFixedRate (() -> {System.out.println ("שלום עולם"); lock.countDown ();}, 500, 100, TimeUnit.MILLISECONDS); lock.await (1000, TimeUnit.MILLISECONDS); future.cancel (נכון);

3.4. ForkJoinPool

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

ב מזלג / הצטרף מסגרת, כל משימה יכולה להשריץ (מזלג) מספר משימות משנה והמתינו להשלמתם באמצעות ה- לְהִצְטַרֵף שיטה. היתרון של מזלג / הצטרף המסגרת היא שזה אינו יוצר שרשור חדש לכל משימה או משימה משנה, מיישם במקום זאת את האלגוריתם Work Stealing. מסגרת זו מתוארת ביסודיות במאמר "מדריך למזלג / הצטרף למסגרת בג'אווה"

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

מחלקה סטטית TreeNode {int ערך; ילדים מוגדרים; TreeNode (ערך int, TreeNode ... ילדים) {this.value = value; this.children = Sets.newHashSet (ילדים); }}

כעת אם אנו רוצים לסכם את כל הערכים בעץ במקביל, עלינו ליישם a משימת רקורסיביות מִמְשָׁק. כל משימה מקבלת את הצומת שלה ומוסיפה את ערכה לסכום הערכים שלה יְלָדִים. לחישוב הסכום של יְלָדִים ערכים, ביצוע המשימות עושה את הפעולות הבאות:

• זורם את יְלָדִים מַעֲרֶכֶת,
• ממפה מעל זרם זה, ויוצר חדש ספירת משימות לכל אלמנט,
• מבצעת כל משימת משנה על ידי זיוף אותה,
• אוסף את התוצאות על ידי התקשרות ל- לְהִצְטַרֵף שיטה על כל משימה מסודרת,
• מסכם את התוצאות באמצעות Collectors.summingInt אַסְפָן.

מחלקה סטטית ציבורית CountingTask מרחיבה את RecursiveTask {הצומת TreeNode הסופי הפרטי; ספירת משימה ציבורית (צומת TreeNode) {this.node = צומת; } מחשב שלם מוגן @ @ Override () {return node.value + node.children.stream () .map (childNode -> CountingTask new (childNode) .fork ()) .collect (Collectors.summingInt (ForkJoinTask :: להצטרף)) ; }}

הקוד להפעלת החישוב על עץ בפועל הוא פשוט מאוד:

עץ TreeNode = TreeNode חדש (5, TreeNode חדש (3), TreeNode חדש (2, TreeNode חדש (2), TreeNode חדש (8))); ForkJoinPool forkJoinPool = ForkJoinPool.commonPool (); סכום int = forkJoinPool.invoke (CountingTask חדש (עץ));

4. יישום בריכת החוטים בגויאבה

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

4.1. הוספת גויאבה כתלות של Maven

הוסף את התלות הבאה לקובץ Maven pom שלך כדי לכלול את ספריית גויאבה לפרויקט שלך. תוכל למצוא את הגרסה האחרונה של ספריית גויאבה במאגר Maven Central:

 com.google.guava גויאבה 19.0 

4.2. מפקח ישיר ושירות להוצאה ישירה

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

בשמחה, גויאבה מספקת מקרים מוגדרים מראש עבורנו.

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

מוציא לפועל = MoreExecutors.directExecutor (); AtomicBoolean מבוצע = AtomicBoolean חדש (); executor.execute (() -> {נסה {Thread.sleep (500);} לתפוס (InterruptedException e) {e.printStackTrace ();} executed.set (true);}); assertTrue (executed.get ());

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

כדאי להעדיף שיטה זו על פני MoreExecutors.newDirectExecutorService () מכיוון ש- API זה יוצר יישום שירות ביצועים מן המניין בכל שיחה.

4.3. יציאה משירותי ביצוע

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

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

שירותים אלה מוסיפים גם וו כיבוי עם ה- Runtime.getRuntime (). AddShutdownHook () שיטה ולמנוע מ- VM להסתיים למשך זמן מוגדר לפני שהוא מוותר על משימות תלויות.

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

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool (5); ExecutorService executorService = MoreExecutors.getExitingExecutorService (מבצע, 100, TimeUnit.MILLISECONDS); executorService.submit (() -> {while (true) {}});

4.4. מעצבי האזנה

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

לעתים רחוקות תרצה להשתמש ListenableFuture.addListener () השיטה באופן ישיר, אך היא כן חיוני לרוב שיטות העוזר ב עתיד מחלקת שירות. למשל, עם Futures.allAsList () שיטה אתה יכול לשלב כמה האזנה לעתיד מקרים בסינגל האזנה לעתיד המסתיים עם השלמתם המוצלחת של כל העתיד ביחד:

ExecutorService executorService = Executors.newCachedThreadPool (); ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator (executorService); ListenableFuture future1 = listeningExecutorService.submit (() -> "שלום"); ListenableFuture future2 = listeningExecutorService.submit (() -> "עולם"); ברכת מחרוזת = Futures.allAsList (future1, future2) .get () .stream () .collect (Collectors.joining ("")); assertEquals ("שלום עולם", ברכה);

5. מסקנה

במאמר זה דנו בתבנית בריכת האשכולות וביישומיה בספריית Java הרגילה ובספריית גויאבה של גוגל.

קוד המקור של המאמר זמין באתר GitHub.