streamalism.org

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

במאמר זה נבחן את ה- DelayQueue לבנות מתוך java.util.concurrent חֲבִילָה. זהו תור חוסם שיכול לשמש בתוכניות יצרן-צרכנים.

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

2. יישום מוּשׁהֶה עבור אלמנטים ב DelayQueue

כל אלמנט שאנחנו רוצים להכניס ל DelayQueue צריך ליישם את מוּשׁהֶה מִמְשָׁק. בואו נגיד שאנחנו רוצים ליצור a DelayObject מעמד. מקרים מאותה כיתה יוכנסו ל DelayQueue.

נעביר את חוּט נתונים ו delayInMilliseconds כמו וטיעונים לבנאי שלו:

מחלקה ציבורית DelayObject מיישמת דחוי {פרטי מחרוזת פרטיים; התחלה ארוכה פרטית; DelayObject ציבורי (נתוני מחרוזת, עיכוב ארוךInMilliseconds) {this.data = נתונים; this.startTime = System.currentTimeMillis () + delayInMilliseconds; }

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

לכן, עלינו להשתמש ב- TimeUnit.convert () שיטה להחזרת העיכוב שנותר בתקן זמן יחידה:

@Override ציבורי ארוך getDelay (יחידת TimeUnit) {long diff = startTime - System.currentTimeMillis (); return unit.convert (diff, TimeUnit.MILLISECONDS); }

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

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

@Override int intlientTo (עיכוב o) {להחזיר Ints.saturatedCast (this.startTime - ((DelayObject) o) .startTime); }

3. DelayQueue Cתושב ומפיק

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

ואז כאשר לָרוּץ() השיטה מופעלת, היא מכניסה אלמנטים לתור, וישנה במשך 500 אלפיות השנייה לאחר כל הצעה:

מחלקה ציבורית DelayQueueProducer מיישמת Runnable {private BlockingQueue queue; מספר שלם פרטיOfElementsToProduc; עיכוב שלם שלם פרטיOfEachProducedMessageMilliseconds; // קונסטרוקטור סטנדרטי @ הריצה בטלנית ציבורית () {עבור (int i = 0; i <numberOfElementsToProduc; i ++) {אובייקט DelayObject = DelayObject חדש (UUID.randomUUID (). toString (), delayOfEachProducedMessageMilliseconds); System.out.println ("שים אובייקט:" + אובייקט); נסה את {queue.put (object); Thread.sleep (500); } לתפוס (InterruptedException כלומר) {ie.printStackTrace (); }}}}

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

מחלקה ציבורית DelayQueueConsumer מיישמת Runnable {private BlockingQueue queue; מספר שלם פרטיOfElementsToTake; מספר AtomicInteger numberOfConsumedElements = AtomicInteger חדש (); // קונסטרוקטורים סטנדרטיים @Override הפעלה בטלנית ציבורית () {for (int i = 0; i <numberOfElementsToTake; i ++) {נסה {DelayObject object = queue.take (); numberOfConsumedElements.incrementAndGet (); System.out.println ("לצרכן לקחת:" + אובייקט); } לתפוס (InterruptedException e) {e.printStackTrace (); }}}}

4. DelayQueue מבחן שימוש

כדי לבדוק את ההתנהגות של DelayQueue, ניצור חוט מפיק אחד וחוט צרכני אחד.

המפיק יעשה זאת לָשִׂים() שני עצמים בתור עם עיכוב של 500 אלפיות השנייה. הבדיקה קובעת כי הצרכן צרך שתי הודעות:

@Test ציבורי בטל givenDelayQueue_whenProducElement _thenShouldConsumeAfterGivenDelay () זורק InterruptedException {// נתון ExecutorService מבצע = Executors.newFixedThreadPool (2); תור BlockingQue = DelayQueue חדש (); int numberOfElementsToProduc = 2; int delayOfEachProducedMessageMilliseconds = 500; DelayQueueConsumer הצרכן = DelayQueueConsumer חדש (תור, numberOfElementsToProduc); מפיק DelayQueueProducer = DelayQueueProducer חדש (תור, numberOfElementsToProduc, delayOfEachProducedMessageMilliseconds); // כאשר executor.submit (מפיק); executor.submit (צרכן); // אז executor.awaitTermination (5, TimeUnit.SECONDS); executor.shutdown (); assertEquals (consumer.numberOfConsumedElements.get (), numberOfElementsToProduce); }

אנו יכולים לראות כי הפעלת תוכנית זו תניב את הפלט הבא:

שים אובייקט: {data = '86046157-e8a0-49b2-9cbb-8326124bcab8', startTime = 1494069868007} לקיחת הצרכן: {data = '86046157-e8a0-49b2-9cbb-8326124bcab8', startTime = 14940988 'd47927ef-18c7-449b-b491-5ff30e6795ed', startTime = 1494069868512} הצרכן לוקח: {data = 'd47927ef-18c7-449b-b491-5ff30e6795ed', startTime = 1494069868512}

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

אותו מצב התרחש לגבי האלמנט השני.

5. הצרכן אינו מסוגל לצרוך בזמן הנתון

בואו נגיד שיש לנו מפיק שמייצר אלמנט שיהיה יפוג תוך 10 שניות:

int numberOfElementsToProduc = 1; int delayOfEachProducedMessageMilliseconds = 10_000; DelayQueueConsumer הצרכן = DelayQueueConsumer חדש (תור, numberOfElementsToProduc); מפיק DelayQueueProducer = DelayQueueProducer חדש (תור, numberOfElementsToProduc, delayOfEachProducedMessageMilliseconds);

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

executor.send (מפיק); executor.submit (צרכן); executor.awaitTermination (5, TimeUnit.SECONDS); executor.shutdown (); assertEquals (consumer.numberOfConsumedElements.get (), 0);

שים לב, כי של הצרכן numberOfConsumedElements יש ערך השווה לאפס.

6. הפקת אלמנט עם תפוגה מיידית

כאשר היישומים של מוּשׁהֶה הוֹדָעָה getDelay () השיטה מחזירה מספר שלילי, כלומר האלמנט הנתון כבר פג. במצב זה, המפיק יצרוך את האלמנט הזה באופן מיידי.

אנו יכולים לבדוק את מצב ייצור האלמנט בעיכוב שלילי:

int numberOfElementsToProduc = 1; int delayOfEachProducedMessageMilliseconds = -10_000; DelayQueueConsumer הצרכן = DelayQueueConsumer חדש (תור, numberOfElementsToProduc); מפיק DelayQueueProducer = DelayQueueProducer חדש (תור, numberOfElementsToProduc, delayOfEachProducedMessageMilliseconds);

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

executor.send (מפיק); executor.submit (צרכן); executor.awaitTermination (1, TimeUnit.SECONDS); executor.shutdown (); assertEquals (consumer.numberOfConsumedElements.get (), 1);

7. מסקנה

במאמר זה, הסתכלנו על DelayQueue לבנות מתוך java.util.concurrent חֲבִילָה.

יישמנו א מוּשׁהֶה אלמנט שיוצר ונצרך מהתור.

מינונו את היישום שלנו של DelayQueue לצרוך אלמנטים שפג תוקפם.

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