streamalism.org

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

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

2. סֵמָפוֹר

נתחיל עם java.util.concurrent.Semaphore. אנו יכולים להשתמש בסמפורות כדי להגביל את מספר השרשורים המקבילים הניגשים למשאב ספציפי.

בדוגמה הבאה ניישם תור כניסה פשוט להגבלת מספר המשתמשים במערכת:

כניסה בכיתהQueueUsingSemaphore {Semaphore private semaphore; כניסה ציבוריתQueueUsingSemaphore (int slotLimit) {semaphore = new Semaphore (slotLimit); } tryLogin בוליאני () {return semaphore.tryAcquire (); } התנתקות בטלה () {semaphore.release (); } int availableSlots () {return semaphore.availablePermits (); }}

שימו לב כיצד השתמשנו בשיטות הבאות:

• tryAcquire () - להחזיר אמת אם היתר זמין באופן מיידי ולרכוש אותו אחרת להחזיר כוזב, אך לִרְכּוֹשׁ() רוכש היתר וחסימה עד שיהיה זמין
• שחרר () - שחרר היתר
• availablePermits () - מספר ההחזר של ההיתרים הנוכחיים הזמינים

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

@Test הציבור בטל givenLoginQueue_whenReachLimit_thenBlocked () {int slots = 10; ExecutorService executorService = Executors.newFixedThreadPool (חריצים); LoginQueueUsingSemaphore loginQueue = LoginQueueUsingSemaphore חדש (משבצות); IntStream.range (0, slots) .forEach (user -> executorService.execute (loginQueue :: tryLogin)); executorService.shutdown (); assertEquals (0, loginQueue.availableSlots ()); assertFalse (loginQueue.tryLogin ()); }

לאחר מכן, נראה אם ​​ישנם חריצים זמינים לאחר התנתקות:

@Test הציבור בטל givenLoginQueue_whenLogout_thenSlotsAvailable () {int slots = 10; ExecutorService executorService = Executors.newFixedThreadPool (משבצות); LoginQueueUsingSemaphore loginQueue = LoginQueueUsingSemaphore חדש (משבצות); IntStream.range (0, slots) .forEach (user -> executorService.execute (loginQueue :: tryLogin)); executorService.shutdown (); assertEquals (0, loginQueue.availableSlots ()); loginQueue.logout (); assertTrue (loginQueue.availableSlots ()> 0); assertTrue (loginQueue.tryLogin ()); }

3. מתוזמן סֵמָפוֹר

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

אנחנו יכולים להשתמש TimedSemaphore לבנות תור עיכוב פשוט כדלקמן:

class DelayQueueUsingTimedSemaphore {private TimedSemaphore semaphore; DelayQueueUsingTimedSemaphore (תקופה ארוכה, int slotLimit) {semaphore = new TimedSemaphore (period, TimeUnit.SECONDS, slotLimit); } בוליאני tryAdd () {return semaphore.tryAcquire (); } int availableSlots () {return semaphore.getAvailablePermits (); }}

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

חלל ציבורי givenDelayQueue_whenReachLimit_thenBlocked () {int slots = 50; ExecutorService executorService = Executors.newFixedThreadPool (משבצות); DelayQueueUsingTimedSemaphore delayQueue = DelayQueueUsingTimedSemaphore חדש (1, משבצות); IntStream.range (0, slots) .forEach (user -> executorService.execute (delayQueue :: tryAdd)); executorService.shutdown (); assertEquals (0, delayQueue.availableSlots ()); assertFalse (delayQueue.tryAdd ()); }

אבל אחרי שישן זמן מה, על הסמפור לאפס ולשחרר את ההיתרים:

@Test ציבור בטל givenDelayQueue_whenTimePass_thenSlotsAvailable () זורק InterruptedException {int slots = 50; ExecutorService executorService = Executors.newFixedThreadPool (חריצים); DelayQueueUsingTimedSemaphore delayQueue = DelayQueueUsingTimedSemaphore חדש (1, משבצות); IntStream.range (0, slots) .forEach (user -> executorService.execute (delayQueue :: tryAdd)); executorService.shutdown (); assertEquals (0, delayQueue.availableSlots ()); Thread.sleep (1000); assertTrue (delayQueue.availableSlots ()> 0); assertTrue (delayQueue.tryAdd ()); }

4. סמפור מול מוטקס

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

בדוגמה הבאה נשתמש בסמפור בינארי פשוט לבניית מונה:

כיתה CounterUsingMutex {Semaphore mutex פרטית; ספירת אינטראקציות פרטיות; CounterUsingMutex () {mutex = סמפור חדש (1); ספירה = 0; } עלייה בטלה () זורקת את InterruptedException {mutex.acquire (); this.count = this.count + 1; Thread.sleep (1000); mutex.release (); } int getCount () {להחזיר this.count; } hasQueuedThreads בוליאני () {return mutex.hasQueuedThreads (); }}

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

@Test ציבורי בטל כאשרMutexAndMultipleThreads_thenBlocked () זורק InterruptedException {int count = 5; ExecutorService executorService = Executors.newFixedThreadPool (ספירה); מונה CounterUsingMutex = CounterUsingMutex חדש (); IntStream.range (0, count) .forEach (user -> executorService.execute (() -> {try {counter.increase ();} catch (InterruptedException e) {e.printStackTrace ();}})); executorService.shutdown (); assertTrue (counter.hasQueuedThreads ()); }

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

@Test הציבור בטל givenMutexAndMultipleThreads_ThenDelay_thenCorrectCount () זורק ExceptionedException {int count = 5; ExecutorService executorService = Executors.newFixedThreadPool (ספירה); מונה CounterUsingMutex = CounterUsingMutex חדש (); IntStream.range (0, count) .forEach (user -> executorService.execute (() -> {try {counter.increase ();} catch (InterruptedException e) {e.printStackTrace ();}})); executorService.shutdown (); assertTrue (counter.hasQueuedThreads ()); Thread.sleep (5000); assertFalse (counter.hasQueuedThreads ()); assertEquals (count, counter.getCount ()); }

5. מסקנה

במאמר זה בחנו את יסודות הסמפורות בג'אווה.

כמו תמיד, קוד המקור המלא זמין ב- GitHub.