עדיף ביחד: Java וכיתה Thread. חלק ו' - אש!

מבוא

חוטים הם דבר מעניין. בביקורות קודמות, בדקנו כמה מהכלים הזמינים להטמעת ריבוי השרשורים. בואו נראה אילו עוד דברים מעניינים אנחנו יכולים לעשות. בשלב זה, אנחנו יודעים הרבה. לדוגמה, מתוך " Ber together: Java and the Thread class. Part I — Threads of execution ", אנו יודעים שהמחלקה Thread מייצגת חוט של ביצוע. אנחנו יודעים ששרשור מבצע משימה כלשהי. אם אנחנו רוצים שהמשימות שלנו יהיו מסוגלות run, אז אנחנו חייבים לסמן את החוט עם Runnable. כדי לזכור, אנו יכולים להשתמש במהדר Java Online של Tutorialspoint :

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 and the Thread class. Part II — Synchronization . אם שרשור אחד רוכש נעילה, אז שרשור אחר שינסה לרכוש את המנעול ייאלץ לחכות לשחרור המנעול:

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();
	}
}

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

סמפור

הדרך הפשוטה ביותר לשלוט בכמה שרשורים יכולים לרוץ בו זמנית היא סמפור. זה כמו אות רכבת. ירוק פירושו להמשיך. אדום פירושו לחכות. חכה למה מהסמפור? גִישָׁה. כדי לקבל גישה, עלינו לרכוש אותה. וכאשר אין עוד צורך בגישה, עלינו למסור אותה או לשחרר אותה. בוא נראה איך זה עובד. אנחנו צריכים לייבא את 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);
}

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

CountDownLatch

המנגנון הבא הוא CountDownLatch. באופן לא מפתיע, מדובר בתפס עם ספירה לאחור. כאן אנו זקוקים להצהרת הייבוא ​​המתאימה לכיתה 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(), כלומר לחכות עד שהמונה יהפוך לאפס. מעניין שזה מונה חד פעמי. התיעוד של Java אומר, "כאשר שרשורים חייבים לספור שוב ושוב לאחור בדרך זו, השתמשו במקום זאת ב-CyclicBarrier". במילים אחרות, אם אתה צריך דלפק לשימוש חוזר, אתה צריך אפשרות אחרת: CyclicBarrier.

CyclicBarrier

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

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

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

חַלְפָן

המנגנון הבא הוא Exchanger. בהקשר זה, Exchange היא נקודת סנכרון שבה דברים משתנים מוחלפים או מחליפים. כפי שהיית מצפה, an 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() + " exchanged with " + 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");
}

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

Phaser

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

public static void main(String[] args) throws InterruptedException {
        Phaser phaser = new Phaser();
        // By calling the register method, we register the current (main) thread as a party
        phaser.register();
        System.out.println("Phasecount is " + phaser.getPhase());
        testPhaser(phaser);
        testPhaser(phaser);
        testPhaser(phaser);
        // After 3 seconds, we arrive at the barrier and deregister. Number of arrivals = number of registrations = start
        Thread.sleep(3000);
        phaser.arriveAndDeregister();
        System.out.println("Phasecount is " + phaser.getPhase());
    }

    private static void testPhaser(final Phaser phaser) {
        // We indicate that there will be a +1 party on the Phaser
        phaser.register();
        // Start a new thread
        new Thread(() -> {
            String name = Thread.currentThread().getName();
            System.out.println(name + " arrived");
            phaser.arriveAndAwaitAdvance(); // The threads register arrival at the phaser.
            System.out.println(name + " after passing barrier");
        }).start();
    }

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

סיכום

כפי שניתן לראות מדוגמאות אלו, ישנן דרכים שונות לסנכרן שרשורים. קודם לכן, ניסיתי להיזכר בהיבטים של ריבוי השרשורים. אני מקווה שהפרקים הקודמים בסדרה זו היו שימושיים. יש אנשים שאומרים שהדרך לריבוי השרשורים מתחילה בספר "מקבילות ג'אווה בתרגול". למרות שהוא יצא לאור בשנת 2006, אנשים אומרים שהספר די בסיסי ועדיין רלוונטי היום. לדוגמה, אתה יכול לקרוא את הדיון כאן: האם "Java Concurrency In Practice" עדיין תקף? . כדאי גם לקרוא את הקישורים בדיון. לדוגמה, יש קישור לספר The Well-Grounded Java Developer , ואנחנו נזכיר במיוחד את פרק 4. Modern concurrency . יש גם סקירה שלמה על הנושא הזה: האם "מקבילות ג'אווה בפועל" עדיין תקף בעידן Java 8? מאמר זה מציע גם הצעות לגבי מה עוד לקרוא כדי להבין באמת את הנושא הזה. לאחר מכן, תוכל להציץ בספר נהדר כמו OCA/OCP Java SE 8 מתכנת מבחנים . אנו מעוניינים בראשי התיבות השניים: OCP (Oracle Certified Professional). תמצא בדיקות ב"פרק 20: במקביל ג'אווה". בספר זה יש גם שאלות וגם תשובות עם הסברים. לדוגמה: אנשים רבים עשויים להתחיל לומר שהשאלה הזו היא עוד דוגמה לשינון שיטות. מצד אחד, כן. מצד שני, אתה יכול לענות על שאלה זו על ידי היזכרות שזה ExecutorServiceסוג של "שדרוג" של Executor. והוא Executorנועד פשוט להסתיר את הדרך בה נוצרים שרשורים, אבל זו לא הדרך העיקרית לבצע אותם, כלומר להתחיל אובייקט Runnableעל שרשור חדש. זו הסיבה שאין execute(Callable)- כי ב- ExecutorService, הפשוט Executorמוסיף submit()שיטות שיכולות להחזיר Futureאובייקט. כמובן, אנחנו יכולים לשנן רשימה של שיטות, אבל הרבה יותר קל לעשות את התשובה שלנו על סמך הידע שלנו על אופי השיעורים עצמם. והנה כמה חומרים נוספים בנושא:

• החבילה java.util.concurrent
• סנכרוני Java
• Concurrency Concepts in Java מאת דאגלס הוקינס

עדיף ביחד: Java וכיתה Thread. חלק א' - חוטי ביצוע טוב יותר ביחד: ג'אווה ומחלקת ה-Thread. חלק ב' - סנכרון טוב יותר ביחד: Java ומחלקת Thread. חלק שלישי - אינטראקציה טובה יותר ביחד: Java ומחלקת Thread. חלק IV — ניתן להתקשרות, עתיד וחברים טובים יותר ביחד: Java וכיתה Thread. חלק V - מבצע, ThreadPool, Fork/Join