עדיף ביחד: Java וכיתה Thread. חלק IV - ניתן להתקשרות, עתיד וחברים

מבוא

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

public class HelloWorld {

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

האם זו האפשרות היחידה להתחיל משימה בשרשור?

java.util.concurrent.ניתן להתקשר

מסתבר של- java.lang.Runnable יש אח בשם java.util.concurrent.Callable שהגיע לעולם ב-Java 1.5. מהם ההבדלים? אם אתה מסתכל מקרוב על Javadoc עבור ממשק זה, אנו רואים שבניגוד ל- Runnable, הממשק החדש מכריז על call()שיטה שמחזירה תוצאה. כמו כן, הוא זורק את Exception כברירת מחדל. כלומר, זה חוסך מאיתנו את הצורך try-catchלחסום עבור חריגים מסומנים. לא נורא, נכון? עכשיו יש לנו משימה חדשה במקום Runnable:

Callable task = () -> {
	return "Hello, World!";
};

אבל מה עושים עם זה? למה אנחנו צריכים משימה שפועלת על שרשור שמחזיר תוצאה? ברור שלכל פעולה שתבוצע בעתיד, אנו מצפים לקבל את התוצאה של פעולות אלו בעתיד. ויש לנו ממשק עם שם מתאים:java.util.concurrent.Future

java.util.concurrent.Future

ממשק java.util.concurrent.Future מגדיר API לעבודה עם משימות שאת תוצאותיהן אנחנו מתכננים לקבל בעתיד: שיטות לקבלת תוצאה, ושיטות לבדיקת סטטוס. בקשר ל- Future, אנו מעוניינים ביישום שלו במחלקה java.util.concurrent.FutureTask . זוהי ה"משימה" שתבוצע ב- Future. מה שהופך את היישום הזה למעניין עוד יותר הוא שהוא מיישם גם Runnable. אתה יכול לראות בזה סוג של מתאם בין המודל הישן של עבודה עם משימות על שרשורים לבין המודל החדש (חדש במובן שהוא הופיע ב-Java 1.5). הנה דוגמא:

import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;

public class HelloWorld {

    public static void main(String[] args) throws Exception {
        Callable task = () -> {
            return "Hello, World!";
        };
        FutureTask<String> future = new FutureTask<>(task);
        new Thread(future).start();
        System.out.println(future.get());
    }
}

כפי שניתן לראות מהדוגמה, אנו משתמשים בשיטה getכדי לקבל את התוצאה מהמשימה. הערה:כאשר אתה מקבל את התוצאה באמצעות get()השיטה, הביצוע הופך לסינכרוני! באיזה מנגנון לדעתך ייעשה שימוש כאן? נכון, אין בלוק סנכרון. לכן לא נראה את WAITING ב-JVisualVM בתור monitorאו wait, אלא כשיטה המוכרת park()(מכיוון שהמנגנון LockSupportנמצא בשימוש).

ממשקים פונקציונליים

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

Supplier<String> supplier = new Supplier<String>() {
	@Override
	public String get() {
		return "String";
	}
};
Consumer<String> consumer = new Consumer<String>() {
	@Override
	public void accept(String s) {
		System.out.println(s);
	}
};
Function<String, Integer> converter = new Function<String, Integer>() {
	@Override
	public Integer apply(String s) {
		return Integer.valueOf(s);
	}
};

המון המון קוד נוסף, לא היית אומר? כל אחת מהמחלקות המוצהרות מבצעת פונקציה אחת, אך אנו משתמשים בחבורה של קוד תומך נוסף כדי להגדיר אותה. וכך חשבו מפתחי ג'אווה. בהתאם לכך, הם הציגו קבוצה של "ממשקים פונקציונליים" ( @FunctionalInterface) והחליטו שעכשיו ג'אווה עצמה תעשה את ה"חשיבה", ומשאירה לנו רק את הדברים החשובים לדאוג לגביהם:

Supplier<String> supplier = () -> "String";
Consumer<String> consumer = s -> System.out.println(s);
Function<String, Integer> converter = s -> Integer.valueOf(s);

אספקה Supplier. אין לו פרמטרים, אבל הוא מחזיר משהו. ככה זה מספק דברים. א Consumerצורכת. זה לוקח משהו כקלט (טיעון) ועושה איתו משהו. הטיעון הוא מה הוא צורך. אז יש לנו גם Function. זה לוקח תשומות (טיעונים), עושה משהו ומחזיר משהו. אתה יכול לראות שאנחנו משתמשים באופן פעיל בגנריות. אם אינך בטוח, תוכל לקבל רענון על ידי קריאת " גנריות ב-Java: כיצד להשתמש בסוגריים זוויתיים בפועל ".

CompletableFuture

הזמן חלף ומחלקה חדשה בשם CompletableFutureהופיעה ב-Java 1.8. הוא מיישם את Futureהממשק, כלומר המשימות שלנו יושלמו בעתיד, ונוכל להתקשר get()כדי לקבל את התוצאה. אבל זה גם מיישם את CompletionStageהממשק. השם אומר הכל: זהו שלב מסוים של מערכת חישובים כלשהי. מבוא קצר לנושא ניתן למצוא בסקירה כאן: Introduction to CompletionStage ו- CompletableFuture. בואו נגיע ישר לנקודה. בואו נסתכל על רשימת השיטות הסטטיות הזמינות שיעזרו לנו להתחיל: להלן אפשרויות השימוש בהן:

import java.util.concurrent.CompletableFuture;
public class App {
    public static void main(String[] args) throws Exception {
        // A CompletableFuture that already contains a Result
        CompletableFuture<String> completed;
        completed = CompletableFuture.completedFuture("Just a value");
        // A CompletableFuture that runs a new thread from Runnable. That's why it's Void
        CompletableFuture<Void> voidCompletableFuture;
        voidCompletableFuture = CompletableFuture.runAsync(() -> {
            System.out.println("run " + Thread.currentThread().getName());
        });
        // A CompletableFuture that starts a new thread whose result we'll get from a Supplier
        CompletableFuture<String> supplier;
        supplier = CompletableFuture.supplyAsync(() -> {
            System.out.println("supply " + Thread.currentThread().getName());
            return "Value";
        });
    }
}

אם נבצע את הקוד הזה, נראה שיצירת a CompletableFutureכרוכה גם בהשקת צינור שלם. לכן, עם דמיון מסוים ל-SteamAPI מ-Java8, זה המקום שבו אנו מוצאים את ההבדל בין הגישות הללו. לדוגמה:

List<String> array = Arrays.asList("one", "two");
Stream<String> stringStream = array.stream().map(value -> {
	System.out.println("Executed");
	return value.toUpperCase();
});

זוהי דוגמה ל-Stream API של Java 8. אם תפעיל את הקוד הזה, תראה ש"בוצע" לא יוצג. במילים אחרות, כאשר נוצר זרם ב-Java, הזרם אינו מתחיל מיד. במקום זאת, הוא מחכה שמישהו ירצה ערך ממנו. אבל CompletableFutureמתחיל לבצע את הצינור מיד, בלי לחכות שמישהו יבקש ממנו ערך. אני חושב שזה חשוב להבין. אז, יש לנו CompletableFuture. איך אנחנו יכולים לעשות צינור (או שרשרת) ואילו מנגנונים יש לנו? זכור אותם ממשקים פונקציונליים שכתבנו עליהם קודם לכן.

• יש לנו a Functionשלוקח A ומחזיר B. יש לו שיטה אחת: apply().
• יש לנו a Consumerשלוקח A ולא מחזיר כלום (Void). יש לו שיטה אחת: accept().
• יש לנו Runnable, שרץ על החוט, ולא לוקח כלום ולא מחזיר כלום. יש לו שיטה אחת: run().

הדבר הבא שצריך לזכור הוא CompletableFutureשמשתמש Runnable, Consumers, Functionsובעבודתו. בהתאם, אתה תמיד יכול לדעת שאתה יכול לעשות את הפעולות הבאות עם CompletableFuture:

public static void main(String[] args) throws Exception {
        AtomicLong longValue = new AtomicLong(0);
        Runnable task = () -> longValue.set(new Date().getTime());
        Function<Long, Date> dateConverter = (longvalue) -> new Date(longvalue);
        Consumer<Date> printer = date -> {
            System.out.println(date);
            System.out.flush();
        };
        // CompletableFuture computation
        CompletableFuture.runAsync(task)
                         .thenApply((v) -> longValue.get())
                         .thenApply(dateConverter)
                         .thenAccept(printer);
}

לשיטות thenRun(), thenApply(), ו- thenAccept()יש גרסאות "אסינכרון". המשמעות היא שהשלבים האלה יושלמו בשרשור אחר. השרשור הזה יילקח ממאגר מיוחד - כך שלא נדע מראש אם זה יהיה שרשור חדש או ישן. הכל תלוי עד כמה המשימות אינטנסיביות מבחינה חישובית. בנוסף לשיטות הללו, יש עוד שלוש אפשרויות מעניינות. למען הבהירות, בואו נדמיין שיש לנו שירות מסוים שמקבל איזשהו מסר מאיפשהו - וזה לוקח זמן:

public static class NewsService {
	public static String getMessage() {
		try {
			Thread.currentThread().sleep(3000);
			return "Message";
		} catch (InterruptedException e) {
			throw new IllegalStateException(e);
		}
	}
}

כעת, בואו נסתכל על יכולות אחרות CompletableFutureשמספקות. נוכל לשלב את התוצאה של CompletableFutureתוצאה של תוצאה אחרת CompletableFuture:

Supplier newsSupplier = () -> NewsService.getMessage();

CompletableFuture<String> reader = CompletableFuture.supplyAsync(newsSupplier);
CompletableFuture.completedFuture("!!")
				 .thenCombine(reader, (a, b) -> b + a)
				 .thenAccept(result -> System.out.println(result))
				 .get();

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

CompletableFuture.completedFuture(2L)
				.thenCompose((val) -> CompletableFuture.completedFuture(val + 2))
                               .thenAccept(result -> System.out.println(result));

כאן אני רוצה לציין שהשיטה CompletableFuture.completedFuture()שימשה לקיצור. שיטה זו אינה יוצרת שרשור חדש, ולכן שאר הצינור יבוצע באותו שרשור שבו completedFutureנקרא. יש גם thenAcceptBoth()שיטה. זה מאוד דומה ל- accept(), אבל אם thenAccept()מקבל a Consumer, thenAcceptBoth()מקבל אחר CompletableStage+ BiConsumerכקלט, כלומר a consumerשלוקח 2 מקורות במקום אחד. ישנה עוד יכולת מעניינת שמציעות שיטות ששמותיהן כוללות את המילה "Either": שיטות אלו מקבלות אלטרנטיבה CompletableStageומבוצעות על זה CompletableStageשמתבצע קודם. לבסוף, אני רוצה לסיים את הסקירה הזו עם תכונה מעניינת נוספת של CompletableFuture: טיפול בשגיאות.

CompletableFuture.completedFuture(2L)
				 .thenApply((a) -> {
					throw new IllegalStateException("error");
				 }).thenApply((a) -> 3L)
				 //.exceptionally(ex -> 0L)
				 .thenAccept(val -> System.out.println(val));

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

• היכרות עם CompletableFuture

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

סיכום

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

• ג'אווה במקביל (Multithreading) - CompletableFuture Explained
• היכרות עם CompletionStage ו-ComletableFuture
• במקביל ושיפורים ב-Java 8: חלק 2
• מדריך ל- CompletableFuture

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