Mai bine împreună: Java și clasa Thread. Partea a IV-a — Apelabil, viitor și prieteni

Introducere

În partea I , am analizat modul în care sunt create firele. Să ne amintim încă o dată. Un thread este reprezentat de clasa Thread, a cărei run()metodă este apelată. Deci, să folosim compilatorul Java online Tutorialspoint și să executăm următorul cod:

public class HelloWorld {

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

Este aceasta singura opțiune pentru a începe o sarcină pe un fir?

java.util.concurrent.Callable

Se pare că java.lang.Runnable are un frate numit java.util.concurrent.Callable care a venit pe lume în Java 1.5. Care sunt diferențele? Dacă vă uitați îndeaproape la Javadoc pentru această interfață, vedem că, spre deosebire de Runnable, noua interfață declară o call()metodă care returnează un rezultat. De asemenea, aruncă excepție în mod implicit. Adică, ne scutește de a fi try-catchblocați pentru excepțiile verificate. Nu-i rău, nu? Acum avem o nouă sarcină în loc de Runnable:

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

Dar ce facem cu ea? De ce avem nevoie de o sarcină care rulează pe un fir care returnează un rezultat? Evident, pentru orice acțiuni efectuate în viitor, ne așteptăm să primim rezultatul acelor acțiuni în viitor. Și avem o interfață cu un nume corespunzător:java.util.concurrent.Future

java.util.concurrent.Future

Interfața java.util.concurrent.Future definește un API pentru lucrul cu sarcini ale căror rezultate intenționăm să le primim în viitor: metode pentru a obține un rezultat și metode pentru a verifica starea. În ceea ce privește Future, suntem interesați de implementarea sa în clasa java.util.concurrent.FutureTask . Aceasta este „Sarcina” care va fi executată în Future. Ceea ce face această implementare și mai interesantă este că implementează și Runnable. Puteți considera acesta un fel de adaptor între vechiul model de lucru cu sarcini pe fire și noul model (nou în sensul că a apărut în Java 1.5). Iată un exemplu:

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

După cum puteți vedea din exemplu, folosim getmetoda pentru a obține rezultatul sarcinii. Notă:când obțineți rezultatul folosind get()metoda, execuția devine sincronă! Ce mecanism credeți că va fi folosit aici? Adevărat, nu există un bloc de sincronizare. De aceea, nu vom vedea WAITING în JVisualVM ca a monitorsau wait, ci ca metoda familiară park()(pentru că LockSupportmecanismul este folosit).

Interfețe funcționale

În continuare, vom vorbi despre clasele din Java 1.8, așa că am face bine să oferim o scurtă introducere. Uită-te la următorul cod:

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

O mulțime de coduri suplimentare, nu ați spune? Fiecare dintre clasele declarate îndeplinește o funcție, dar folosim o grămadă de cod suplimentar pentru a o defini. Și așa gândeau dezvoltatorii Java. În consecință, au introdus un set de „interfețe funcționale” ( @FunctionalInterface) și au decis că acum Java însuși va face „gândirea”, lăsând doar lucrurile importante pentru care să ne îngrijorăm:

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

A Supplierprovizii. Nu are parametri, dar returnează ceva. Așa furnizează lucrurile. A Consumerconsumă. Ia ceva ca intrare (un argument) și face ceva cu el. Argumentul este ceea ce consumă. Atunci avem și Function. Preia intrări (argumente), face ceva și returnează ceva. Puteți vedea că folosim în mod activ medicamente generice. Dacă nu sunteți sigur, puteți obține o reîmprospătare citind „ Generice în Java: cum să utilizați parantezele unghiulare în practică ”.

CompletabilViitorul

Timpul a trecut și o nouă clasă numită CompletableFuturea apărut în Java 1.8. Implementează Futureinterfața, adică sarcinile noastre vor fi finalizate în viitor și putem apela get()pentru a obține rezultatul. Dar implementează și CompletionStageinterfața. Numele spune totul: aceasta este o anumită etapă a unui set de calcule. O scurtă introducere a subiectului poate fi găsită în recenzia aici: Introduction to CompletionStage și CompletableFuture. Să trecem direct la obiect. Să ne uităm la lista metodelor statice disponibile care ne vor ajuta să începem: Iată opțiunile de utilizare a acestora:

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

Dacă executăm acest cod, vom vedea că crearea unui CompletableFutureimplică și lansarea unui întreg pipeline. Prin urmare, cu o anumită similitudine cu SteamAPI de la Java8, aici găsim diferența dintre aceste abordări. De exemplu:

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

Acesta este un exemplu de API Stream din Java 8. Dacă rulați acest cod, veți vedea că „Executat” nu va fi afișat. Cu alte cuvinte, atunci când un flux este creat în Java, fluxul nu începe imediat. În schimb, așteaptă ca cineva să-și dorească o valoare de la ea. Dar CompletableFutureîncepe să execute conducta imediat, fără a aștepta ca cineva să-i ceară o valoare. Cred că acest lucru este important de înțeles. Deci avem un CompletableFuture. Cum putem face o conductă (sau lanț) și ce mecanisme avem? Amintiți-vă acele interfețe funcționale despre care am scris mai devreme.

• Avem un Functioncare ia un A și returnează un B. Are o singură metodă: apply().
• Avem un Consumercare ia un A și nu returnează nimic (Void). Are o singură metodă: accept().
• Avem Runnable, care rulează pe fir și nu ia nimic și nu returnează nimic. Are o singură metodă: run().

Următorul lucru de reținut este că CompletableFuturefolosește Runnable, Consumers, și Functionsîn activitatea sa. În consecință, puteți ști întotdeauna că puteți face următoarele cu 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);
}

Metodele thenRun(), thenApply(), și thenAccept()au versiuni „Async”. Aceasta înseamnă că aceste etape vor fi finalizate pe un fir diferit. Acest thread va fi preluat dintr-un pool special - așa că nu vom ști dinainte dacă va fi un thread nou sau vechi. Totul depinde de cât de intensive din punct de vedere computațional sunt sarcinile. Pe lângă aceste metode, există încă trei posibilități interesante. Pentru claritate, să ne imaginăm că avem un anumit serviciu care primește un fel de mesaj de undeva - și asta necesită timp:

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

Acum, să aruncăm o privire la alte abilități care CompletableFuturele oferă. Putem combina rezultatul lui a CompletableFuturecu rezultatul altuia 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();

Rețineți că firele sunt fire daemon în mod implicit, așa că pentru claritate, obișnuim get()să așteptăm rezultatul. Nu numai că putem combina CompletableFutures, dar putem și returna CompletableFuture:

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

Aici vreau să remarc că CompletableFuture.completedFuture()metoda a fost folosită pentru concizie. Această metodă nu creează un fir nou, astfel încât restul conductei va fi executat pe același fir în care completedFuturea fost apelat. Există și o thenAcceptBoth()metodă. Este foarte asemănător cu accept(), dar dacă thenAccept()acceptă a Consumer, thenAcceptBoth()acceptă un alt CompletableStage+ BiConsumerca intrare, adică a consumercare ia 2 surse în loc de una. Există o altă abilitate interesantă oferită de metodele al căror nume include cuvântul „Fire”: Aceste metode acceptă o alternativă CompletableStageși sunt executate pe CompletableStagecare se execută mai întâi. În cele din urmă, vreau să închei această recenzie cu o altă caracteristică interesantă a CompletableFuture: gestionarea erorilor.

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

Acest cod nu va face nimic, pentru că va exista o excepție și nu se va întâmpla nimic altceva. Dar prin decomentarea afirmației „excepțional”, definim comportamentul așteptat. Apropo de CompletableFuture, vă recomand să urmăriți și următorul videoclip:

• Introducere în CompletableFuture

După umila mea părere, acestea sunt printre cele mai explicative videoclipuri de pe internet. Ei ar trebui să explice clar cum funcționează toate acestea, ce trusă de instrumente avem la dispoziție și de ce sunt necesare toate acestea.

Concluzie

Sperăm că acum este clar cum puteți utiliza firele pentru a obține calcule după ce sunt finalizate. Material suplimentar:

• Concurență Java (multithreading) — CompletableFuture Explained
• Introducere în CompletionStage și CompletableFuture
• Concurență și îmbunătățiri în Java 8: Partea 2
• Ghid pentru CompletableFuture

Mai bine împreună: Java și clasa Thread. Partea I — Fire de execuție Mai bine împreună: Java și clasa Thread. Partea a II-a — Sincronizare Mai bine împreună: Java și clasa Thread. Partea a III-a — Interacțiunea Mai bine împreună: Java și clasa Thread. Partea V — Executor, ThreadPool, Fork/Join Better împreună: Java și clasa Thread. Partea a VI-a — Foc departe!