streamalism.org

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

במאמר זה נבחן את ה- SynchronousQueue מ ה java.util.concurrent חֲבִילָה.

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

2. סקירת API

ה SynchronousQueue יש רק שתי פעולות נתמכות: לקחת() ו לָשִׂים(), ושניהם חוסמים.

לדוגמא, כאשר אנו רוצים להוסיף אלמנט לתור, עלינו להתקשר ל- לָשִׂים() שיטה. שיטה זו תיחסם עד ששרשור אחר יקרא ל- לקחת() שיטה, מאותתת שהיא מוכנה לקחת אלמנט.

למרות ש SynchronousQueue יש ממשק של תור, עלינו לחשוב על זה כנקודת חילוף לאלמנט יחיד בין שני שרשורים, שבו חוט אחד מחלק אלמנט, וחוט אחר לוקח את האלמנט הזה.

3. יישום מסירות ידיים באמצעות משתנה משותף

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

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

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

נגדיר א sharedState משתנה ו CountDownLatch שישמש לתיאום עיבוד:

ExecutorService executor = Executors.newFixedThreadPool (2); AtomicInteger sharedState = AtomicInteger חדש (); CountDownLatch countDownLatch = CountDownLatch חדש (1);

המפיק ישמור מספר שלם אקראי ל- sharedState משתנה, ולבצע את ספירה לאחור() שיטה על countDownLatch, איתות לצרכן שהוא יכול להביא ערך מה- sharedState:

מפיק ניתן להריצה = () -> {Integer producedElement = ThreadLocalRandom .current () .nextInt (); sharedState.set (producedElement); countDownLatch.countDown (); };

הצרכן ימתין על countDownLatch משתמש ב לְהַמתִין() שיטה. כאשר היצרן מסמן שהמשתנה הוגדר, הצרכן יביא אותו מה- sharedState:

צרכן ניתן להריצה = () -> {נסה {countDownLatch.await (); מספר שלם consumedElement = sharedState.get (); } לתפוס (InterruptedException ex) {ex.printStackTrace (); }};

אחרון חביב, בואו נתחיל בתוכנית שלנו:

executor.execute (מפיק); executor.execute (צרכן); executor.awaitTermination (500, TimeUnit.MILLISECONDS); executor.shutdown (); assertEquals (countDownLatch.getCount (), 0);

זה יפיק את התפוקה הבאה:

שמירת אלמנט: -1507375353 לנקודת החילוף צרכה אלמנט: -1507375353 מנקודת ההחלפה

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

4. יישום מסירות ידיים באמצעות SynchronousQueue

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

ראשית, נגדיר תור:

ExecutorService executor = Executors.newFixedThreadPool (2); תור SynchronousQue = SynchronousQueue חדש ();

המפיק יקרא a לָשִׂים() שיטה שתיחסם עד שאיזה שרשור אחר מוציא אלמנט מהתור:

מפיק ניתן להריצה = () -> {Integer producedElement = ThreadLocalRandom .current () .nextInt (); נסה את {queue.put (producedElement); } לתפוס (InterruptedException ex) {ex.printStackTrace (); }};

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

צרכן ניתן להפעיל = () -> {נסה {Integer consumedElement = queue.take (); } לתפוס (InterruptedException ex) {ex.printStackTrace (); }};

לאחר מכן נתחיל בתכניתנו:

executor.execute (מפיק); executor.execute (צרכן); executor.awaitTermination (500, TimeUnit.MILLISECONDS); executor.shutdown (); assertEquals (queue.size (), 0);

זה יפיק את התפוקה הבאה:

שמירת אלמנט: 339626897 לנקודת החילוף צרכה אלמנט: 339626897 מנקודת ההחלפה

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

5. מסקנה

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

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