Mas mahusay na magkasama: Java at ang klase ng Thread. Bahagi V — Tagapagpatupad, ThreadPool, Fork/Join

Panimula

Kaya, alam namin na ang Java ay may mga thread. Mababasa mo iyon sa pagsusuri na pinamagatang Better together: Java and the Thread class. Bahagi I — Mga Thread ng pagpapatupad . Tingnan natin muli ang karaniwang code:

public static void main(String[] args) throws Exception {
	Runnable task = () -> {
		System.out.println("Task executed");
	};
	Thread thread = new Thread(task);
	thread.start();
}

Tulad ng nakikita mo, ang code upang magsimula ng isang gawain ay medyo tipikal, ngunit kailangan nating ulitin ito para sa bagong gawain. Ang isang solusyon ay ilagay ito sa isang hiwalay na paraan, hal execute(Runnable runnable). Ngunit isinaalang-alang ng mga tagalikha ng Java ang aming kalagayan at nakabuo sila ng Executorinterface:

public static void main(String[] args) throws Exception {
	Runnable task = () -> System.out.println("Task executed");
	Executor executor = (runnable) -> {
		new Thread(runnable).start();
	};
	executor.execute(task);
}

Ang code na ito ay malinaw na mas maigsi: ngayon ay nagsusulat lang kami ng code upang simulan ang Runnablesa thread. Iyan ay mahusay, hindi ba? Ngunit ito ay simula lamang:

https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executor.html

Tulad ng nakikita mo, ang Executorinterface ay may ExecutorServicesubinterface. Ang Javadoc para sa interface na ito ay nagsasabi na ang isang ExecutorServiceay naglalarawan ng isang partikular Executorna nagbibigay ng mga pamamaraan upang isara ang Executor. Ginagawa rin nitong posible na makakuha ng isang java.util.concurrent.Futureupang masubaybayan ang proseso ng pagpapatupad. Dati, sa Better together: Java at ang Thread class. Part IV — Callable, Future, at mga kaibigan , saglit naming sinuri ang mga kakayahan ng Future. Kung nakalimutan mo o hindi mo nabasa ito, iminumungkahi kong i-refresh mo ang iyong memorya ;) Ano pa ang sinasabi ng Javadoc? Sinasabi nito sa amin na mayroon kaming espesyal java.util.concurrent.Executorsna pabrika na hinahayaan kaming lumikha ng mga default na pagpapatupad ng ExecutorService.

ExecutorService

Magreview tayo. Kailangan nating Executormagsagawa (ibig sabihin, tumawag execute()) ng isang tiyak na gawain sa isang thread, at ang code na lumilikha ng thread ay nakatago sa amin. Mayroon kaming ExecutorService— isang partikular Executorna may ilang mga opsyon para sa pagkontrol sa pag-unlad. At mayroon kaming Executorspabrika na nagbibigay-daan sa amin na lumikha ng isang ExecutorService. Ngayon gawin natin ito sa ating sarili:

public static void main(String[] args) throws ExecutionException, InterruptedException {
	Callable<String> task = () -> Thread.currentThread().getName();
	ExecutorService service = Executors.newFixedThreadPool(2);
	for (int i = 0; i < 5; i++) {
		Future result = service.submit(task);
		System.out.println(result.get());
	}
	service.shutdown();
}

Makikita mo na tinukoy namin ang isang nakapirming thread pool na ang laki ay 2. Pagkatapos ay isa-isang isinusumite namin ang mga gawain sa pool. Ang bawat gawain ay nagbabalik ng isang Stringnaglalaman ng pangalan ng thread ( currentThread().GetName()). Mahalagang isara ang ExecutorServicesa pinakadulo, dahil kung hindi ay hindi matatapos ang ating programa. Ang Executorspabrika ay may mga karagdagang pamamaraan ng pabrika. Halimbawa, maaari tayong lumikha ng pool na binubuo lamang ng isang thread ( newSingleThreadExecutor) o isang pool na may kasamang cache ( newCachedThreadPool) kung saan aalisin ang mga thread pagkatapos na idle ang mga ito sa loob ng 1 minuto. Sa katotohanan, ang mga ito ExecutorServiceay sinusuportahan ng isang blocking queue , kung saan inilalagay ang mga gawain at kung saan ang mga gawain ay isinasagawa. Higit pang impormasyon tungkol sa pagharang sa mga pila ay matatagpuan sa video na ito . Mababasa mo rin itopagsusuri tungkol sa BlockingQueue . At tingnan ang sagot sa tanong na "Kailan mas gusto ang LinkedBlockingQueue kaysa ArrayBlockingQueue?" Sa pinakasimpleng termino, BlockingQueuehinaharangan ng isang thread ang isang thread sa dalawang kaso:

• sinusubukan ng thread na kumuha ng mga item mula sa isang walang laman na pila
• sinusubukan ng thread na ilagay ang mga item sa isang buong pila

Kung titingnan natin ang pagpapatupad ng mga pamamaraan ng pabrika, makikita natin kung paano gumagana ang mga ito. Halimbawa:

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
}

o

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
}

Tulad ng nakikita natin, ang mga pagpapatupad ng ExecutorServiceay nilikha sa loob ng mga pamamaraan ng pabrika. At sa karamihan, pinag-uusapan natin ang ThreadPoolExecutor. Tanging ang mga parameter na nakakaapekto sa trabaho ay binago.

https://en.wikipedia.org/wiki/Thread_pool#/media/File:Thread_pool.svg

ThreadPoolExecutor

Tulad ng nakita natin kanina, ThreadPoolExecutorito ang karaniwang nalilikha sa loob ng mga pamamaraan ng pabrika. Naaapektuhan ang functionality ng mga argumentong ipinapasa namin bilang maximum at minimum na bilang ng mga thread, pati na rin kung anong uri ng pila ang ginagamit. Ngunit anumang pagpapatupad ng java.util.concurrent.BlockingQueueinterface ay maaaring gamitin. Sa pagsasalita ng ThreadPoolExecutor, dapat nating banggitin ang ilang mga kagiliw-giliw na tampok. Halimbawa, hindi ka maaaring magsumite ng mga gawain sa isang ThreadPoolExecutorkung walang available na espasyo:

public static void main(String[] args) throws ExecutionException, InterruptedException {
	int threadBound = 2;
	ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(0, threadBound,
            0L, TimeUnit.SECONDS, new SynchronousQueue<>());
	Callable<String> task = () -> {
		Thread.sleep(1000);
		return Thread.currentThread().getName();
	};
	for (int i = 0; i < threadBound + 1; i++) {
		threadPoolExecutor.submit(task);
	}
	threadPoolExecutor.shutdown();
}

Ang code na ito ay mag-crash na may error na tulad nito:

Task java.util.concurrent.FutureTask@7cca494b rejected from java.util.concurrent.ThreadPoolExecutor@7ba4f24f[Running, pool size = 2, active threads = 2, queued tasks = 0, completed tasks = 0]

Sa madaling salita, taskhindi maaaring isumite, dahil SynchronousQueueidinisenyo upang ito ay aktwal na binubuo ng isang elemento at hindi pinapayagan kaming maglagay ng higit pa dito. Makikita natin na mayroon tayong zero queued tasks("mga nakapila na gawain = 0") dito. Ngunit walang kakaiba tungkol dito, dahil ito ay isang espesyal na tampok ng SynchronousQueue, na sa katunayan ay isang 1-element na pila na laging walang laman! Kapag naglagay ang isang thread ng elemento sa queue, maghihintay ito hanggang sa kunin ng isa pang thread ang elemento mula sa queue. Alinsunod dito, maaari naming palitan ito ng new LinkedBlockingQueue<>(1)at ang error ay mababago sa ngayon ay ipakita queued tasks = 1. Dahil 1 element lang ang queue, hindi kami makakapagdagdag ng pangalawang elemento. At iyon ang dahilan kung bakit mabigo ang programa. Sa pagpapatuloy ng aming pagtalakay sa pila, ito ay nagkakahalaga ng pagpuna na angThreadPoolExecutorAng klase ay may mga karagdagang pamamaraan para sa pagseserbisyo sa pila. Halimbawa, threadPoolExecutor.purge()aalisin ng pamamaraan ang lahat ng nakanselang gawain mula sa pila upang makapagbakante ng espasyo sa pila. Ang isa pang kawili-wiling function na nauugnay sa queue ay ang handler para sa mga tinanggihang gawain:

public static void main(String[] args) {
	ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 1,
            0L, TimeUnit.SECONDS, new SynchronousQueue());
	Callable<String> task = () -> Thread.currentThread().getName();
	threadPoolExecutor.setRejectedExecutionHandler((runnable, executor) -> System.out.println("Rejected"));
	for (int i = 0; i < 5; i++) {
		threadPoolExecutor.submit(task);
	}
	threadPoolExecutor.shutdown();
}

Sa halimbawang ito, ipinapakita lang ng aming handler Rejectedsa tuwing tatanggihan ang isang gawain sa pila. Maginhawa, hindi ba? Bilang karagdagan, ThreadPoolExecutormayroong isang kawili-wiling subclass: ScheduledThreadPoolExecutor, na isang ScheduledExecutorService. Nagbibigay ito ng kakayahang magsagawa ng isang gawain batay sa isang timer.

ScheduledExecutorService

ScheduledExecutorService(na isang uri ng ExecutorService) ay nagbibigay-daan sa amin na magpatakbo ng mga gawain sa isang iskedyul. Tingnan natin ang isang halimbawa:

public static void main(String[] args) {
	ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(4);
	Callable<String> task = () -> {
		System.out.println(Thread.currentThread().getName());
		return Thread.currentThread().getName();
	};
	scheduledExecutorService.schedule(task, 1, TimeUnit.MINUTES);
	scheduledExecutorService.shutdown();
}

Simple lang ang lahat dito. Ang mga gawain ay isinumite at pagkatapos ay makakakuha tayo ng java.util.concurrent.ScheduledFuture. Ang isang iskedyul ay maaari ding makatulong sa sumusunod na sitwasyon:

ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(4);
Runnable task = () -> {
	System.out.println(Thread.currentThread().getName());
};
scheduledExecutorService.scheduleAtFixedRate(task, 1, 2, TimeUnit.SECONDS);

Dito nagsusumite kami ng Runnablegawain para sa pagpapatupad sa isang nakapirming dalas ("FixedRate") na may tiyak na paunang pagkaantala. Sa kasong ito, pagkatapos ng 1 segundo, magsisimulang isagawa ang gawain tuwing 2 segundo. Mayroong katulad na opsyon:

scheduledExecutorService.scheduleWithFixedDelay(task, 1, 2, TimeUnit.SECONDS);

Ngunit sa kasong ito, ang mga gawain ay ginagampanan na may isang tiyak na pagitan sa PAGITAN ng bawat pagpapatupad. Iyon ay, ang taskay isasagawa pagkatapos ng 1 segundo. Pagkatapos, sa sandaling makumpleto ito, lilipas ang 2 segundo, at magsisimula ang isang bagong gawain. Narito ang ilang karagdagang mapagkukunan sa paksang ito:

• Isang panimula sa mga thread pool sa Java
• Panimula sa Mga Thread Pool sa Java
• Java Multithreading Steeplechase: Pagkansela ng Mga Gawain Sa Mga Tagapagpatupad
• Paggamit ng Java Executors para sa Mga Gawain sa Background

https://dzone.com/articles/diving-into-java-8s-newworkstealingpools

WorkStealingPool

Bilang karagdagan sa mga thread pool sa itaas, mayroong isa pa. Masasabi nating ito ay medyo espesyal. Ito ay tinatawag na work-stealing pool. Sa madaling salita, ang pagnanakaw sa trabaho ay isang algorithm kung saan ang mga idle na thread ay nagsisimulang kumuha ng mga gawain mula sa iba pang mga thread o mga gawain mula sa isang shared queue. Tingnan natin ang isang halimbawa:

public static void main(String[] args) {
	Object lock = new Object();
	ExecutorService executorService = Executors.newCachedThreadPool();
	Callable<String> task = () -> {
		System.out.println(Thread.currentThread().getName());
		lock.wait(2000);
		System.out.println("Finished");
		return "result";
	};
	for (int i = 0; i < 5; i++) {
		executorService.submit(task);
	}
	executorService.shutdown();
}

Kung patakbuhin natin ang code na ito, ExecutorServicelilikha ang the 5 thread para sa atin, dahil ang bawat thread ay ilalagay sa wait queue para sa lock object. Naisip na namin ang mga monitor at lock sa Better together: Java at ang Thread class. Bahagi II — Pag-synchronize . Ngayon ay palitan natin Executors.newCachedThreadPool()ng Executors.newWorkStealingPool(). Ano ang magbabago? Makikita namin na ang aming mga gawain ay isinasagawa sa mas kaunti sa 5 mga thread. Tandaan na CachedThreadPoollumilikha ng isang thread para sa bawat gawain? Iyon ay dahil wait()na-block ang thread, gustong makumpleto ang mga kasunod na gawain, at gumawa ng mga bagong thread para sa kanila sa pool. Sa isang pagnanakaw na pool, ang mga thread ay hindi walang ginagawa magpakailanman. Sinimulan nilang gawin ang mga gawain ng kanilang mga kapitbahay. Ano ang pinagkaiba WorkStealingPoolng ibang thread pool? Ang katotohanan na ang mahiwagangForkJoinPoolnakatira sa loob nito:

public static ExecutorService newWorkStealingPool() {
        return new ForkJoinPool
            (Runtime.getRuntime().availableProcessors(),
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
             null, true);
}

Sa totoo lang, may isa pang pagkakaiba. Bilang default, ang mga thread na ginawa para sa isang ForkJoinPoolay mga daemon thread, hindi katulad ng mga thread na ginawa sa pamamagitan ng isang onrdinary ThreadPool. Sa pangkalahatan, dapat mong tandaan ang mga thread ng daemon, dahil, halimbawa, CompletableFutureay gumagamit din ng mga thread ng daemon maliban kung tinukoy mo ang iyong sarili ThreadFactoryna lumilikha ng mga hindi-daemon na thread. Ito ang mga sorpresa na maaaring magtago sa mga hindi inaasahang lugar! :)

ForkJoinPool

Sa bahaging ito, muli nating pag-uusapan ang ForkJoinPool(tinatawag ding fork/join framework), na nabubuhay "sa ilalim ng talukbong" ng WorkStealingPool. Sa pangkalahatan, lumitaw ang fork/join framework sa Java 1.7. At kahit na malapit na ang Java 11, sulit pa rin itong alalahanin. Hindi ito ang pinakakaraniwang pagpapatupad, ngunit ito ay medyo kawili-wili. Mayroong magandang pagsusuri tungkol dito sa web: Pag-unawa sa Java Fork-Join Framework na may Mga Halimbawa . Ang ForkJoinPoolumaasa sa java.util.concurrent.RecursiveTask. Meron din java.util.concurrent.RecursiveAction. RecursiveActionhindi nagbabalik ng resulta. Kaya, RecursiveTaskay katulad ng Callable, at RecursiveActionkatulad ng unnable. Makikita natin na kasama sa pangalan ang mga pangalan ng dalawang mahahalagang pamamaraan: forkat join. Angforkang pamamaraan ay nagsisimula ng ilang gawain nang asynchronous sa isang hiwalay na thread. At joinhinahayaan ka ng pamamaraan na maghintay para matapos ang trabaho. Upang makuha ang pinakamahusay na pag-unawa, dapat mong basahin ang Mula sa Imperative Programming hanggang Fork/Join sa Parallel Streams sa Java 8 .

Buod

Buweno, tinatapos nito ang bahaging ito ng pagsusuri. Natutunan namin na Executororihinal na naimbento para magsagawa ng mga thread. Pagkatapos ay nagpasya ang mga tagalikha ng Java na ipagpatuloy ang ideya at gumawa ng ExecutorService. ExecutorServicehinahayaan kaming magsumite ng mga gawain para sa pagpapatupad gamit submit()ang at invoke(), at isara din ang serbisyo. Dahil ExecutorServicenangangailangan ng mga pagpapatupad, nagsulat sila ng isang klase na may mga pamamaraan ng pabrika at tinawag itong Executors. Hinahayaan ka nitong lumikha ng mga thread pool ( ThreadPoolExecutor). Bukod pa rito, may mga thread pool na nagbibigay-daan din sa amin na tumukoy ng iskedyul ng pagpapatupad. At isang ForkJoinPoolnagtatago sa likod ng isang WorkStealingPool. Umaasa ako na natagpuan mo ang isinulat ko sa itaas hindi lamang kawili-wili, ngunit naiintindihan din :) Lagi akong natutuwa na marinig ang iyong mga mungkahi at komento. Mas mahusay na magkasama: Java at ang klase ng Thread. Bahagi I — Mga Thread ng execution Mas mahusay na magkasama: Java at ang Thread class. Bahagi II — Pag-synchronize Mas mahusay na magkasama: Java at ang Thread na klase. Part III — Mas Mahusay na Pakikipag-ugnayan: Java at ang Thread class. Bahagi IV — Matatawagan, Hinaharap, at mga kaibigan Mas mahusay na magkasama: Java at ang Thread na klase. Bahagi VI - Sunog!