Шумо наметавонед Java-ро бо ришта вайрон кунед: Қисми V - Иҷрокунанда, ThreadPool, Fork ҳамроҳ шавед

Муқаддима

Ҳамин тавр, мо медонем, ки дар Java риштаҳо мавҷуданд, ки шумо метавонед дар баррасии " Шумо Java-ро бо ришта вайрон карда наметавонед: Қисми I - Threads " хонда метавонед. Биёед бори дигар ба рамзи намуна назар кунем:

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

Тавре ки мо мебинем, рамзи оғози вазифа хеле стандартӣ аст, аммо барои ҳар як оғози нав мо бояд онро такрор кунем. Як ҳалли он аст, ки онро ба усули алоҳида интиқол диҳед, масалан execute(Runnable runnable). Аммо таҳиягарони Java аллакай дар бораи мо хавотир шудаанд ва интерфейсеро пешниҳод кардаанд Executor:

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

Тавре ки шумо мебинед, code мухтасартар шудааст ва ба мо имкон дод, ки танҳо code нависем, то онро Runnableдар ришта иҷро кунем. Аҷоиб, ҳамин тавр не? Аммо ин танҳо ибтидо аст:

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

Тавре ки шумо мебинед, интерфейс Executorинтерфейси насли дорад ExecutorService. JavaDoc-и ин интерфейс мегӯяд, ки ExecutorServiceон тавсифи махсуси Executor'a' мебошад, ки усулҳои қатъи кори Executor'a'-ро таъмин мекунад ва ба шумо имкон медиҳад, ки java.util.concurrent.Futureпешрафти иҷроро пайгирӣ кунед. Пештар, дар " Шумо Java-ро бо ришта вайрон карда наметавонед: Қисми IV - даъватшаванда, оянда ва дӯстон ", мо имконотро мухтасар дида баромадем Future. Агар шумо онро фаромӯш карда бошед ё нахонда бошед, ман ба шумо маслиҳат медиҳам, ки хотираатонро тароват диҳед;) Дар JavaDoc боз кадом чизҳои ҷолиб навишта шудаанд? Мо як корхонаи махсус дорем java.util.concurrent.Executors, ки ба мо имкон медиҳад, ки татбиқҳоеро эҷод кунем, ки бо нобаёнӣ дастрасанд ExecutorService.

Хадамоти иҷрокунанда

Боз ба хотир меорем. Мо бояд Executorвазифаи муайянеро дар ришта иҷро кунем (яъне иҷро кунем), вақте ки амалисозии эҷоди ришта аз мо пинҳон аст. Мо ExecutorServiceяк махсус дорем Executor, ки дорои маҷмӯи қобorятҳо барои идоракунии пешрафти иҷро мебошад. Ва мо як завод дорем Executors, ки ба шумо имкон медиҳад, ки эҷод кунед ExecutorService. Биёед ҳоло худамон ин корро кунем:

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

Тавре ки мо мебинем, мо як ҳавзи риштаи собит ( Fixed Thread Pool)-и андозаи 2-ро муайян кардем. Баъд аз он мо вазифаҳоро як ба як ба ҳавз мефиристем. Ҳар як супориш сатри ( String)-ро бар мегардонад, ки дорои номи ришта ( currentThread().getName()). Муҳим аст, ки дар ниҳоят қатъ карда шавад ExecutorService, зеро дар акси ҳол барномаи мо хориҷ намешавад. Дар завод Executorsдигар усулхои заводй мавчуданд . Масалан, мо метавонем як ҳавзи танҳо як ришта эҷод кунем - newSingleThreadExecutorё ҳавзи дорои кэш newCachedThreadPool, ки дар он риштаҳо дар сурати 1 дақиқа бекор мондан аз ҳавз хориҷ карда мешаванд. Дарвоқеъ, дар паси инҳо як навбати басташавӣExecutorService мавҷуд аст , ки дар он вазифаҳо гузошта мешаванд ва аз онҳо ин вазифаҳо иҷро мешаванд. Маълумоти бештарро дар бораи бастани навбатҳо дар видеои " Бастани навбат - Коллексияҳои #5 - Java Advanced " дидан мумкин аст. Шумо инчунин метавонед баррасии " Бастани навбатҳои бастаи ҳамзамон " ва ҷавобро ба саволи " Кай LinkedBlockingQueue аз ArrayBlockingQueue афзалтар дониста шавад? " -ро хонед. Super соддакардашуда - (навбати басташавӣ) риштаро дар ду ҳолат блок мекунад: BlockingQueue

• ришта кӯшиш мекунад, ки элементҳоро аз навбати холӣ гирад
• ришта кӯшиш мекунад, ки элементҳоро ба навбати пурра гузорад

Агар ба татбики усулхои завод назар кунем, мебинем, ки онхо чй тавр сохта шудаанд. Барои намуна:

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

ё

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

Тавре ки мо мебинем, татбиқҳо дар дохor усулҳои завод сохта шудаанд ExecutorService. Ва ин асосан аст ThreadPoolExecutor. Танҳо сифатҳое, ки ба кор таъсир мерасонанд, тағир меёбанд.

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

ThreadPoolExecutor

Чунон ки пештар дидем, дар дохor завод усулхои ThreadPoolExecutor, . Функсия аз он вобаста аст, ки кадом арзишҳо ҳамчун риштаҳои максималӣ ва минималӣ интиқол дода мешаванд ва инчунин кадом навбат истифода мешаванд. Ва ҳама гуна татбиқи интерфейс метавонад истифода шавад java.util.concurrent.BlockingQueue. Дар бораи ThreadPoolExecutor'ахс сухан ронда, дар вакти кор хусусиятхои ачоибро кайд кардан лозим аст. Масалан, ThreadPoolExecutorагар дар он ҷо ҷой набошад, шумо наметавонед супоришҳоро ба:

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

Ин code бо хатогӣ ба монанди:

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]

Яъне taskшумо пешниҳод карда наметавонед, зеро SynchronousQueueон тавре тарҳрезӣ шудааст, ки он воқеан аз як унсур иборат аст ва ба шумо имкон намедиҳад, ки дар он ҷо бештар ҷойгир кунед. Тавре ки мебинем, queued tasksдар ин ҷо 0 вуҷуд дорад ва дар ин ҳеҷ чизи аҷибе нест, зеро ин мушаххас аст SynchronousQueue- дар асл, он навбат аз 1 элемент аст, ки ҳамеша холӣ аст. (!) Вақте ки як ришта элементеро ба навбат мегузорад, он интизор мешавад, ки риштаи дигар элементро аз навбат мегирад. Аз ин рӯ, мо метавонем бо иваз кунем new LinkedBlockingQueue<>(1)ва он чизе, ки дар хато нишон дода мешавад, тағир меёбад queued tasks = 1. Зеро навбат танҳо 1 элемент аст, пас мо наметавонем дуюмашро илова кунем. Ва мо ба ин меафтем. Мавзӯи навбатро идома дода, бояд қайд кард, ки синф ThreadPoolExecutorусулҳои иловагии хидматрасонии навбат дорад. Масалан, ин усул threadPoolExecutor.purge()ҳамаи вазифаҳои бекоршударо аз навбат хориҷ мекунад, то ҷой дар навбат холӣ шавад. Боз як хусусияти ҷолиби марбут ба навбат ин коркарди вазифаҳои қабулнашаванда мебошад:

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

Масалан, коркардкунанда танҳо калимаро Rejectedбарои ҳар як рад кардани қабули супориш дар навбат чоп мекунад. Қулай, ҳамин тавр не? Илова бар ин, ThreadPoolExecutorӯ як вориси ҷолиб дорад - ScheduledThreadPoolExecutorкист ScheduledExecutorService. Он қобorяти иҷрои вазифаро дар таймер таъмин мекунад.

ScheduledExecutorService

ExecutorServiceнавъи ScheduledExecutorServiceба шумо имкон медиҳад, ки вазифаҳоро мувофиқи ҷадвал иҷро кунед. Биёед як мисолро дида бароем:

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

Дар ин ҷо ҳама чиз оддӣ аст. Супоришҳо фиристода мешаванд, мо "вазифаи нақшавӣ" мегирем java.util.concurrent.ScheduledFuture. Ҳолати зерин низ метавонад бо ҷадвал муфид бошад:

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

Дар ин ҷо мо Runnableвазифаеро мефиристем, ки бо суръати муқарраршуда (Фикси Собит) бо таъхири муайян иҷро карда шавад. Дар ин ҳолат, пас аз 1 сония ҳар 2 сония, ба иҷрои вазифа шурӯъ кунед. Варианти монанд вуҷуд дорад:

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

Аммо дар ин ҷо супоришҳо бо фосилаи додашуда БАЙНИ иҷрои вазифаҳои гуногун иҷро карда мешаванд. Яъне супориш taskдар 1 сония ичро мешавад. Баъдан, ҳамин ки ба анҷом расид, 2 сония мегузарад ва сипас вазифаи нав оғоз мешавад. Шумо метавонед маводҳои зеринро дар ин мавзӯъ мутолиа кунед:

• Муқаддима ба ҳавзҳои ришта
• Муқаддима ба ҳавзҳои ришта
• Java Multithreading Steeplechase: Бекор кардани вазифаҳо дар иҷрокунандагон
• Интихоби иҷрокунандагони дурусти Java барои вазифаҳои замина

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

WorkStealing Pool

Илова ба ҳавзҳои риштаи дар боло зикршуда, боз як ҳавз вуҷуд дорад. Шумо метавонед бигӯед, ки вай каме махсус аст. Номи он ҳавзи дуздии корӣ аст. Хулоса, дуздии корӣ як алгоритми кор аст, ки дар он риштаҳои бекор ба гирифтани вазифаҳо аз риштаҳои дигар ё супоришҳо аз навбати умумӣ оғоз мекунанд. Биёед як мисолро дида бароем:

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

Агар мо ин codeро иҷро кунем, ExecutorServiceон 5 ришта эҷод мекунад, зеро ҳар як ришта ба навбати интизорӣ дар маҳалли ҷойгиршавии an object ҳамроҳ мешавад lock. Мо аллакай дар бораи мониторҳо ва қулфҳои он дар " Шумо Java-ро бо ришта вайрон карда наметавонед: Қисми II - Синхронизатсия " муҳокима кардем. Executors.newCachedThreadPoolВа ҳоло мо онро бо иваз мекунем Executors.newWorkStealingPool(). Чӣ тағир хоҳад ёфт? Мо мебинем, ки супоришхои мо на дар 5 ришта, балки камтар ичро мешаванд. Дар хотир доред, ки cachedThreadPoolшумо барои ҳар як вазифа риштаи шахсии худро офаридаед? Зеро waitон риштаро бастааст, аммо вазифаҳои навбатӣ мехостанд иҷро шаванд ва дар ҳавз барои онҳо риштаҳои нав эҷод карда шуданд. Дар мавриди StealingPoolриштаҳо, онҳо ҳамеша дар , бекор намемонанд wait, онҳо ба иҷрои вазифаҳои ҳамсоя шурӯъ мекунанд. Ин чӣ гуна аз дигар ҳавзҳои ришта фарқ мекунад WorkStealingPool? Зеро дар дохor ӯ воқеан чизи ҷодугаре зиндагӣ мекунад ForkJoinPool:

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

Дар асл як фарқияти дигар вуҷуд дорад. Риштаҳое, ки бо нобаёнӣ сохта мешаванд ForkJoinPool, риштаҳои демон мебошанд, бар хилофи риштаҳое, ки тавассути муқаррарӣ сохта шудаанд ThreadPool. Умуман, дар бораи риштаҳои демон ёдовар шудан бамаврид аст, зеро... масалан, CompletableFutureриштаҳои демон низ истифода мешаванд, агар шумо худатонро нишон надиҳед ThreadFactory, ки риштаҳои ғайридемониро эҷод мекунанд. Инҳоянд сюрпризҳое, ки шуморо дар ҷои ғайричашмдошт интизоранд!)

Фарқ / Ҳавзи ҳамроҳ

Дар ин бахш мо дар бораи ҳамон яке ForkJoinPool(инчунин чаҳорчӯбаи fork/join номида мешавад) сӯҳбат хоҳем кард, ки "дар зери сарпӯш" -и WorkStealingPool. Умуман, чаҳорчӯбаи Fork Join дар Java 1.7 пайдо шуд. Ва ҳатто агар Java 11 аллакай дар ҳавлӣ бошад, он ҳанӯз ҳам ба ёдоварист. На вазифаи маъмултарин, балки хеле ҷолиб. Дар Интернет баррасии хубе дар ин мавзӯъ мавҷуд аст: " Fork/Join Framework дар Java 7 ". Fork/JoinPoolдар кори худ бо чунин концепция амал мекунад java.util.concurrent.RecursiveTask. Аналоги низ вуҷуд дорад - java.util.concurrent.RecursiveAction. RecursiveActions натиҷаро барнамегардонад. Ҳамин тавр RecursiveTaskба , монанд Callableва RecursiveActionмонанд ба Runnable. Хуб, ба ном нигоҳ карда, мо ду усули асосиро мебинем - forkва join. Усул forkвазифаро ба таври асинхронӣ дар риштаи алоҳида иҷро мекунад. Ва усул joinба шумо имкон медиҳад, ки то анҷоми кор интизор шавед. Якчанд роҳҳои истифодаи он вуҷуд доранд: Ин расм як қисми слайд аз гузориши Алексей Шипилев аст " Ҷойгир/Ҳамроҳ: татбиқ, истифода, иҷроиш ." Барои равшантар кардани он, бояд гузориши ӯро дар JEE CONF тамошо кунед: " Хусусиятҳои татбиқи Fork Join ."

Ҷамъбаст кардан

Инак, мо қисми навбатии баррасиро анҷом медиҳем. Мо фаҳмидем, ки Executorбарои иҷрои риштаҳо чӣ кор карда будем. Баъд мо тасмим гирифтем, ки ин идеяро идома диҳем ва онро ба миён овардем ExecutorService. ба шумо имкон медиҳад, ки бо истифода аз ва ExecutorServiceвазифаҳоро барои иҷро фиристед , инчунин хидматро бо хомӯш кардани он идора кунед. Зеро 'ба мо татбиқҳо лозиманд, мо синфро бо усулҳои заводӣ навиштем ва онро . Он ба шумо имкон медиҳад, ки ҳавзҳои ришта эҷод кунед . Дар айни замон, ҳавзҳои ришта мавҷуданд, ки ба шумо имкон медиҳанд ҷадвали иҷроро муайян кунед, аммо он дар паси . Умедворам, ки он чизе, ки дар боло навишта шудааст, барои шумо на танҳо ҷолиб, балки фаҳмо ҳам буд) Ман ҳамеша аз гирифтани пешниҳодҳо ва эродҳо хушҳолам. #ВячеславsubmitinvokeExecutorServiceExecutorsThreadPoolExecutorWorkStealingPoolForkJoinPool