Jobb együtt: Java és a Thread osztály. IV. rész – Hívható, jövő és barátok

Bevezetés

Az I. részben áttekintettük , hogyan jönnek létre a szálak. Emlékezzünk vissza még egyszer. A szálat a Thread osztály képviseli, amelynek run()metódusa meghívásra kerül. Tehát használjuk a Tutorialspoint online Java fordítót , és futtassuk a következő kódot:

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

Ez az egyetlen lehetőség egy feladat elindítására egy szálon?

java.util.concurrent.Callable

Kiderült, hogy a java.lang.Runnable-nek van egy java.util.concurrent.Callable nevű testvére , aki a Java 1.5-ben jött a világra. Mik a különbségek? Ha alaposan megnézi ennek a felületnek a Javadoc-ot, azt látjuk, hogy az új felülettel ellentétben Runnableaz új felület olyan metódust deklarál call(), amely eredményt ad vissza. Ezenkívül alapértelmezés szerint kivételt dob. Vagyis megkímél minket attól, hogy le kell try-catchtiltanunk az ellenőrzött kivételeket. Nem rossz, igaz? Most új feladatunk van ahelyett Runnable:

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

De mit kezdjünk vele? Miért van szükségünk egy olyan szálon futó feladatra, amely eredményt ad vissza? Nyilvánvaló, hogy a jövőben végrehajtott bármely tevékenység esetén elvárjuk, hogy a jövőben megkapjuk az adott cselekvés eredményét. És van egy interfészünk a megfelelő névvel:java.util.concurrent.Future

java.util.concurrent.Future

A java.util.concurrent.Future interfész API-t definiál az olyan feladatokkal való munkavégzéshez, amelyek eredményeit a jövőben tervezzük megkapni: az eredmény elérésének és az állapot ellenőrzésének módszerei. Ami a -t illeti Future, érdekel minket annak megvalósítása a java.util.concurrent.FutureTask osztályban. Ez az a "Feladat", amely a következőben kerül végrehajtásra Future. Ezt az implementációt még érdekesebbé teszi, hogy a Runnable-t is megvalósítja. Ezt egyfajta adapternek tekintheti a szálakon végzett feladatokkal való munkavégzés régi modellje és az új modell között (új modell abban az értelemben, hogy a Java 1.5-ben jelent meg). Íme egy példa:

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

Ahogy a példából is látható, a getmódszert arra használjuk, hogy megkapjuk a feladat eredményét. Jegyzet:amikor a metódussal megkapod az eredményt get(), a végrehajtás szinkron lesz! Ön szerint milyen mechanizmust fognak itt használni? Igaz, nincs szinkronizálási blokk. Ez az oka annak, hogy a JVisualVM-ben a WAITING-t nem a monitorvagy -ként fogjuk látni wait, hanem a megszokott park()metódusként (mivel a LockSupportmechanizmust használják).

Funkcionális interfészek

Ezután a Java 1.8 osztályairól fogunk beszélni, ezért jól tesszük, ha röviden bemutatjuk. Nézd meg a következő kódot:

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

Sok-sok extra kód, nem mondanád? A deklarált osztályok mindegyike egy funkciót hajt végre, de ennek meghatározásához egy csomó extra támogató kódot használunk. És így gondolták a Java fejlesztők. Ennek megfelelően bevezettek egy sor "funkcionális interfészt" ( @FunctionalInterface), és úgy döntöttek, hogy most már maga a Java végzi a "gondolkodást", csak a fontos dolgokat hagyva nekünk aggódni:

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

Egy Supplierkellékek. Paraméterei nincsenek, de visszaad valamit. Így látja el a dolgokat. A Consumerfogyaszt. Valamit inputnak (érvnek) vesz, és csinál vele valamit. Az érv az, hogy mit fogyaszt. Akkor nekünk is van Function. Bemeneteket (érveket) vesz, csinál valamit, és visszaad valamit. Látható, hogy aktívan használunk generikus gyógyszereket. Ha nem biztos benne, felfrissülhet, ha elolvassa a " Generics in Java: a szögletes zárójelek használata a gyakorlatban " című részt.

CompletableFuture

Telt-múlt az idő, és egy új osztály jelent CompletableFuturemeg a Java 1.8-ban. Ez valósítja meg a Futurefelületet, azaz a feladataink a jövőben elkészülnek, és get()az eredményért telefonálhatunk. De az interfészt is megvalósítja CompletionStage. A név mindent elárul: ez egy bizonyos szakasza egy bizonyos számítási sorozatnak. A téma rövid bevezetője az itt található áttekintésben található: Introduction to CompletionStage és CompletableFuture. Térjünk a lényegre. Nézzük meg a rendelkezésre álló statikus módszerek listáját, amelyek segítenek az indulásban: Íme a használatukra vonatkozó lehetőségek:

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

Ha végrehajtjuk ezt a kódot, látni fogjuk, hogy a létrehozása CompletableFutureegy teljes folyamat elindítását is magában foglalja. Ezért a Java8 SteamAPI-jához való bizonyos hasonlóság mellett itt találjuk meg a különbséget e megközelítések között. Például:

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

Ez egy példa a Java 8 Stream API-jára. Ha futtatja ezt a kódot, látni fogja, hogy a „Végrehajtva” nem jelenik meg. Más szóval, amikor egy adatfolyamot Java nyelven hoz létre, az adatfolyam nem indul el azonnal. Ehelyett arra vár, hogy valaki értéket akarjon tőle. De CompletableFutureazonnal megkezdi a csővezeték végrehajtását, anélkül, hogy megvárná, hogy valaki értéket kérjen tőle. Szerintem ezt fontos megérteni. Szóval van egy CompletableFuture. Hogyan készítsünk csővezetéket (vagy láncot), és milyen mechanizmusokkal rendelkezünk? Emlékezzünk vissza azokra a funkcionális interfészekre, amelyekről korábban írtunk.

• Van a Function, amely egy A-t vesz fel, és egy B-t ad vissza. Egyetlen módszere van: apply().
• Van olyan Consumer, amelyik egy A-t vesz fel, és nem ad vissza semmit (Vid). Egyetlen módszere van: accept().
• Van Runnable, amely a szálon fut, és nem vesz el semmit, és nem ad vissza semmit. Egyetlen módszere van: run().

A következő dolog, amit meg kell jegyeznünk, az, hogy a , és -t CompletableFuturehasználja a munkájában. Ennek megfelelően mindig tudhatja, hogy a következőket teheti : RunnableConsumersFunctionsCompletableFuture

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

A thenRun(), thenApply(), és thenAccept()metódusoknak "Async" verziója van. Ez azt jelenti, hogy ezek a szakaszok egy másik szálon fognak befejeződni. Ez a szál egy speciális gyűjteményből lesz átvéve – így nem tudjuk előre, hogy új vagy régi szál lesz-e. Minden attól függ, hogy a feladatok mennyire számításigényesek. Ezeken a módszereken kívül még három érdekes lehetőség adódik. Az egyértelműség kedvéért képzeljük el, hogy van egy bizonyos szolgáltatásunk, amely valahonnan valamilyen üzenetet kap – és ez időbe telik:

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

Most pedig vessünk egy pillantást a további képességekre, amelyeket CompletableFuturebiztosít. Összevonhatjuk a eredményét egy CompletableFuturemásik eredményével 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();

Vegye figyelembe, hogy a szálak alapértelmezés szerint démonszálak, ezért az egyértelműség kedvéért várunk get()az eredményre. Nem csak kombinálhatjuk CompletableFutures, hanem vissza is küldhetjük CompletableFuture:

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

Itt szeretném megjegyezni, hogy a CompletableFuture.completedFuture()módszert a tömörség kedvéért használták. Ez a metódus nem hoz létre új szálat, így a folyamat többi része ugyanazon a szálon lesz végrehajtva, ahol completedFuturemeghívásra került. Van egy thenAcceptBoth()módszer is. Nagyon hasonló a -hoz accept(), de ha thenAccept()elfogad egy -t Consumer, akkor egy másik +-t thenAcceptBoth()fogad be bemenetként, azaz a-t , amely 2 forrást vesz fel egy helyett. Egy másik érdekes képességet kínálnak a metódusok, amelyek nevében benne van az "vagy" szó: ezek a metódusok elfogadnak egy alternatívát , és az elsőként végrehajtotton futnak le . Végül szeretném befejezni ezt az áttekintést egy másik érdekes tulajdonsággal : a hibakezeléssel. CompletableStageBiConsumerconsumerCompletableStageCompletableStageCompletableFuture

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

Ez a kód nem fog semmit, mert lesz kivétel, és semmi más nem fog történni. De a "kivételesen" állítás megjegyzésének megszüntetésével meghatározzuk az elvárt viselkedést. Apropó CompletableFuture, javaslom a következő videó megtekintését is:

• Bevezetés a CompletableFuture-be

Szerény véleményem szerint ezek a legmagyarázatosabb videók közé tartoznak az interneten. Világossá kell tenniük, hogyan működik mindez, milyen eszköztár áll rendelkezésünkre, és miért van szükség erre.

Következtetés

Remélhetőleg most már világos, hogyan használhatja a szálakat a számítások elvégzésére azok befejezése után. Kiegészítő anyag:

• Java párhuzamosság (Multithreading) – CompletableFuture Explained
• A CompletionStage és a CompletableFuture bemutatása
• Egyidejűség és fejlesztések a Java 8-ban: 2. rész
• Útmutató a CompletableFuture-hoz

Jobb együtt: Java és a Thread osztály. I. rész – A végrehajtás szálai Jobb együtt: Java és a Thread osztály. II. rész – Szinkronizálás Jobb együtt: Java és a Thread osztály. III. rész – Interakció Jobb együtt: Java és a szál osztály. V. rész – Végrehajtó, ThreadPool, Fork/Join Better together: Java és a Thread osztály. VI. rész – Tüzet el!