אתה לא יכול לקלקל את ג'אווה עם שרשור: חלק VI - אל המחסום!

מבוא

זרמים הם דבר מעניין. בסקירות קודמות, בדקנו כמה מהכלים הזמינים להטמעת ריבוי הליכי שרשור. בואו נראה אילו דברים מעניינים נוספים אנחנו יכולים לעשות. בשלב זה אנחנו יודעים הרבה. לדוגמה, מתוך " אי אפשר לקלקל את ג'אווה עם שרשור: חלק א' - חוטים ", אנחנו יודעים ששרשור הוא שרשור. אנחנו יודעים ששרשור מבצע משימה כלשהי. אם אנחנו רוצים שהמשימה שלנו תהיה מסוגלת לרוץ ( run), אז עלינו לציין שהשרשור יהיה מסויים Runnable. כדי לזכור, אנו יכולים להשתמש ב- Tutorialspoint Java Online Compiler :

public static void main(String []args){
	Runnable task = () -> {
 		Thread thread = Thread.currentThread();
		System.out.println("Hello from " + thread.getName());
	};
	Thread thread = new Thread(task);
	thread.start();
}

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

import java.util.concurrent.locks.*;

public class HelloWorld{
	public static void main(String []args){
		Lock lock = new ReentrantLock();
		Runnable task = () -> {
			lock.lock();
			Thread thread = Thread.currentThread();
			System.out.println("Hello from " + thread.getName());
			lock.unlock();
		};
		Thread thread = new Thread(task);
		thread.start();
	}
}

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

סמפור

האמצעי הפשוט ביותר לשלוט בכמה חוטים יכולים לעבוד בו זמנית הוא סמפור. כמו על מסילת הברזל. האור הירוק דולק - אתה יכול. הנורה האדומה דולקת - אנחנו מחכים. מה אנחנו מצפים מסמאפור? הרשאות. אישור באנגלית - אישור. כדי לקבל אישור, אתה צריך להשיג אותו, אשר באנגלית יהיה לרכוש. וכשאין צורך יותר ברשות, עלינו למסור אותה, כלומר לשחרר אותה או להיפטר ממנה, מה שבאנגלית יהיה release. בואו נראה איך זה עובד. נצטרך לייבא את המחלקה java.util.concurrent.Semaphore. דוגמא:

public static void main(String[] args) throws InterruptedException {
	Semaphore semaphore = new Semaphore(0);
	Runnable task = () -> {
		try {
			semaphore.acquire();
			System.out.println("Finished");
			semaphore.release();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	};
	new Thread(task).start();
	Thread.sleep(5000);
	semaphore.release(1);
}

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

CountDownLatch

המנגנון הבא הוא CountDownLatch. CountDown באנגלית הוא ספירה לאחור, ו-Latch הוא בורג או תפס. כלומר, אם נתרגם את זה, אז זה תפס עם ספירה לאחור. כאן אנחנו צריכים את היבוא המתאים של המחלקה java.util.concurrent.CountDownLatch. זה כמו מרוץ או מרוץ שבו כולם מתאספים על קו הזינוק וכשכולם מוכנים, ניתן אישור וכולם מתחילים באותו זמן. דוגמא:

public static void main(String[] args) {
	CountDownLatch countDownLatch = new CountDownLatch(3);
	Runnable task = () -> {
		try {
			countDownLatch.countDown();
			System.out.println("Countdown: " + countDownLatch.getCount());
			countDownLatch.await();
			System.out.println("Finished");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	};
	for (int i = 0; i < 3; i++) {
		new Thread(task).start();
 	}
}

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

CyclicBarrier

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

public static void main(String[] args) throws InterruptedException {
	Runnable action = () -> System.out.println("На старт!");
	CyclicBarrier berrier = new CyclicBarrier(3, action);
	Runnable task = () -> {
		try {
			berrier.await();
			System.out.println("Finished");
		} catch (BrokenBarrierException | InterruptedException e) {
			e.printStackTrace();
		}
	};
	System.out.println("Limit: " + berrier.getParties());
	for (int i = 0; i < 3; i++) {
		new Thread(task).start();
	}
}

כפי שאתה יכול לראות, השרשור מבוצע await, כלומר ממתין. במקרה זה, ערך המחסום יורד. המחסום נחשב שבור ( berrier.isBroken()) כאשר הספירה לאחור מגיעה לאפס. כדי לאפס את המחסום, עליך להתקשר berrier.reset(), שהיה חסר ב CountDownLatch.

חַלְפָן

התרופה הבאה היא Exchanger. החלפה מאנגלית מתורגמת כ- Exchange או Exchange. A Exchangerהוא מחליף, כלומר משהו שדרכו הם מחליפים. בואו נסתכל על דוגמה פשוטה:

public static void main(String[] args) {
	Exchanger<String> exchanger = new Exchanger<>();
	Runnable task = () -> {
		try {
			Thread thread = Thread.currentThread();
			String withThreadName = exchanger.exchange(thread.getName());
			System.out.println(thread.getName() + " обменялся с " + withThreadName);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	};
	new Thread(task).start();
	new Thread(task).start();
}

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

public static void main(String[] args) throws InterruptedException {
	SynchronousQueue<String> queue = new SynchronousQueue<>();
	Runnable task = () -> {
		try {
			System.out.println(queue.take());
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	};
	new Thread(task).start();
	queue.put("Message");
}

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

Phaser

ולבסוף, הדבר הכי מתוק - Phaser. נצטרך לייבא את המחלקה java.util.concurrent.Phaser. בואו נסתכל על דוגמה פשוטה:

public static void main(String[] args) throws InterruptedException {
        Phaser phaser = new Phaser();
        // Вызывая метод register, мы регистрируем текущий поток (main) How участника
        phaser.register();
        System.out.println("Phasecount is " + phaser.getPhase());
        testPhaser(phaser);
        testPhaser(phaser);
        testPhaser(phaser);
        // Через 3 секунды прибываем к барьеру и снимаемся регистрацию. Кол-во прибывших = кол-во регистраций = пуск
        Thread.sleep(3000);
        phaser.arriveAndDeregister();
        System.out.println("Phasecount is " + phaser.getPhase());
    }

    private static void testPhaser(final Phaser phaser) {
        // Говорим, что будет +1 участник на Phaser
        phaser.register();
        // Запускаем новый поток
        new Thread(() -> {
            String name = Thread.currentThread().getName();
            System.out.println(name + " arrived");
            phaser.arriveAndAwaitAdvance(); //threads register arrival to the phaser.
            System.out.println(name + " after passing barrier");
        }).start();
    }

הדוגמה מלמדת כי המחסום, בעת שימוש Phaserב-a, נשבר כאשר מספר הרישומים עולה בקנה אחד עם מספר הכניסות למחסום. אתה יכול לגלות עוד Phaserבמאמר מהרכזת " Synchronizer New Phaser ".

תוצאות

כפי שניתן לראות מהדוגמאות, ישנן דרכים שונות לסנכרן שרשורים. מוקדם יותר ניסיתי להיזכר במשהו על ריבוי הליכי שרשור, אני מקווה שהחלקים הקודמים היו שימושיים. הם אומרים שהדרך לריבוי השרשורים מתחילה בספר "מקבילות ג'אווה בתרגול". למרות שהוא יצא ב-2006, אנשים מגיבים שהספר הוא די בסיסי ועדיין פוגע. לדוגמה, אתה יכול לקרוא את הדיונים כאן: " האם Java Concurrency In Practice עדיין בתוקף? ". זה גם מועיל לקרוא את הקישורים מהדיון. לדוגמה, יש קישור לספר " מפתח ג'אווה מבוסס היטב ", בו כדאי לשים לב ל"פרק 4. במקביליות מודרנית ". יש עוד סקירה שלמה על אותו נושא: " האם מטבעות משותפים של Java בפועל עדיין רלוונטיים בעידן של ג'אווה 8 ". יש בו גם טיפים על מה עוד כדאי לקרוא כדי להבין באמת את הנושא. לאחר מכן, אתה יכול להסתכל מקרוב על ספר נפלא כמו " OCA OCP JavaSE 8 Programmer Practice Tests ". אנחנו מתעניינים בחלק השני, כלומר, OCP. ויש מבחנים ב"∫". ספר זה מכיל שאלות ותשובות עם הסברים. לדוגמה: רבים עשויים להתחיל לומר שזה רק עוד שינון של שיטות. מצד אחד, כן. מצד שני, ניתן לענות על שאלה זו על ידי לזכור שמדובר ExecutorServiceבסוג של "שדרוג" Executor. והוא Executorנועד פשוט להסתיר את שיטת יצירת השרשורים, אך לא את הדרך העיקרית לביצועם, כלומר, ריצה בשרשור חדש Runnable. לכן, execute(Callable)לא, כי הם פשוט הוסיפו שיטות ExecutorServiceשיכולות לחזור . כפי שאתה יכול לראות, אנחנו יכולים לשנן רשימה של שיטות, אבל זה הרבה יותר קל לנחש אם אנחנו יודעים את אופי השיעורים עצמם. ובכן, כמה חומרים נוספים בנושא: ExecutorsubmitFuture

• " Reference Synchronizer java.util.concurrent.* "
• " מסנכרני חבילות במקביל "
• יורי טקך " סינכרונים - במקביל מס' 4 - ג'אווה מתקדמת "

#ויאצ'סלב