बेहतर एक साथ: जावा और थ्रेड क्लास। भाग वी - एक्ज़ीक्यूटर, थ्रेडपूल, फोर्क / जॉइन

परिचय

तो, हम जानते हैं कि जावा में थ्रेड्स हैं। आप इसके बारे में बेहतर एक साथ समीक्षा में पढ़ सकते हैं : जावा और थ्रेड क्लास। भाग I - निष्पादन के सूत्र । आइए विशिष्ट कोड पर एक और नज़र डालें:

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)। लेकिन जावा के रचनाकारों ने हमारी दुर्दशा पर विचार किया और इंटरफ़ेस के साथ आए 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);
}

यह कोड स्पष्ट रूप से अधिक संक्षिप्त है: अब हम केवल Runnableथ्रेड पर प्रारंभ करने के लिए कोड लिखते हैं। यह बहुत अच्छा है, है ना? लेकिन यह महज़ एक शुरुआत है:

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

जैसा कि आप देख सकते हैं, Executorइंटरफ़ेस में एक ExecutorServiceसबइंटरफ़ेस है। इस इंटरफ़ेस के लिए Javadoc का कहना है कि a ExecutorServiceकिसी विशेष का वर्णन करता है Executorजो Executor. java.util.concurrent.Futureनिष्पादन प्रक्रिया को ट्रैक करने के लिए यह एक क्रम प्राप्त करना भी संभव बनाता है । पहले, बेटर टुगेदर: Java और थ्रेड क्लास में। भाग IV - कॉल करने योग्य, भविष्य और दोस्तों , हमने संक्षेप में की क्षमताओं की समीक्षा की Future। यदि आप इसे भूल गए हैं या इसे कभी नहीं पढ़ा है, तो मेरा सुझाव है कि आप अपनी स्मृति को ताज़ा करें;) जावाडोक और क्या कहता है? यह हमें बताता है कि हमारे पास एक विशेष java.util.concurrent.Executorsकारखाना है जो हमें ExecutorService.

निष्पादक सेवा

पिछली समीक्षा। हमें एक थ्रेड पर एक निश्चित कार्य Executorनिष्पादित करना है (यानी कॉल करना है execute()), और थ्रेड बनाने वाला कोड हमसे छिपा हुआ है। हमारे पास ExecutorService— एक विशिष्ट Executorजिसमें प्रगति को नियंत्रित करने के लिए कई विकल्प हैं। और हमारे पास वह 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();
}

आप देख सकते हैं कि हमने एक निश्चित थ्रेड पूल निर्दिष्ट किया है जिसका आकार 2 है। फिर हम एक-एक करके कार्यों को पूल में जमा करते हैं। प्रत्येक कार्य एक लौटाता है Stringजिसमें थ्रेड नाम ( currentThread().GetName()) होता है। ExecutorServiceइसे बिल्कुल अंत में बंद करना महत्वपूर्ण है , क्योंकि अन्यथा हमारा कार्यक्रम समाप्त नहीं होगा। कारखाने Executorsमें अतिरिक्त कारखाने के तरीके हैं। उदाहरण के लिए, हम एक पूल बना सकते हैं जिसमें केवल एक थ्रेड ( newSingleThreadExecutor) या एक पूल शामिल है जिसमें एक कैश ( newCachedThreadPool) शामिल है जिसमें से थ्रेड्स को 1 मिनट के लिए निष्क्रिय रहने के बाद हटा दिया जाता है। वास्तव में, इन्हें ExecutorServiceएक अवरुद्ध कतार द्वारा समर्थित किया जाता है , जिसमें कार्यों को रखा जाता है और जिनसे कार्यों को निष्पादित किया जाता है। क्यू ब्लॉक करने के बारे में अधिक जानकारी इस वीडियो में पाई जा सकती है । आप इसे भी पढ़ सकते हैंBlockingQueue के बारे में समीक्षा करें । और प्रश्न का उत्तर देखें "कब ArrayBlockingQueue पर LinkedBlockingQueue को प्राथमिकता दें?" सबसे सरल शब्दों में, a 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>());
}

ExecutorServiceजैसा कि हम देख सकते हैं, फैक्ट्री विधियों के अंदर कार्यान्वयन बनाए जाते हैं। और अधिकांश भाग के लिए, हम बात कर रहे हैं ThreadPoolExecutor। केवल कार्य को प्रभावित करने वाले पैरामीटर बदले जाते हैं।

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

थ्रेडपूल एक्ज़ीक्यूटर

जैसा कि हमने पहले देखा, ThreadPoolExecutorआमतौर पर फ़ैक्टरी विधियों के अंदर क्या बनाया जाता है। कार्यक्षमता उन तर्कों से प्रभावित होती है जिन्हें हम थ्रेड्स की अधिकतम और न्यूनतम संख्या के रूप में पास करते हैं, साथ ही किस प्रकार की कतार का उपयोग किया जा रहा है। लेकिन java.util.concurrent.BlockingQueueइंटरफ़ेस के किसी भी कार्यान्वयन का उपयोग किया जा सकता है। बोलते हुए ThreadPoolExecutor, हमें कुछ रोचक विशेषताओं का जिक्र करना चाहिए। ThreadPoolExecutorउदाहरण के लिए, यदि कोई स्थान उपलब्ध नहीं है, तो आप a को कार्य सबमिट नहीं कर सकते :

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

यह कोड इस तरह की त्रुटि के साथ क्रैश हो जाएगा:

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. यह टाइमर के आधार पर कार्य करने की क्षमता प्रदान करता है।

अनुसूचित निष्पादक सेवा

ScheduledExecutorService(जो एक प्रकार का है ExecutorService) हमें कार्यों को एक समय पर चलाने देता है। आइए एक उदाहरण देखें:

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

लेकिन इस मामले में, कार्यों को प्रत्येक निष्पादन के बीच एक विशिष्ट अंतराल के साथ किया जाता है। यानी task1 सेकंड के बाद वसीयत को निष्पादित किया जाएगा। फिर उसके पूरा होते ही 2 सेकंड बीत जायेंगे और फिर एक नया काम शुरू हो जायेगा। इस विषय पर कुछ अतिरिक्त संसाधन यहां दिए गए हैं:

• जावा में थ्रेड पूल का परिचय
• जावा में थ्रेड पूल का परिचय
• जावा मल्टीथ्रेडिंग स्टीपलचेज़: निष्पादकों में कार्य रद्द करना
• पृष्ठभूमि कार्यों के लिए जावा निष्पादकों का उपयोग करना

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

वर्कस्टीलिंगपूल

उपरोक्त थ्रेड पूल के अलावा, एक और पूल है। हम ईमानदारी से कह सकते हैं कि यह थोड़ा खास है। इसे कामचोरी करने वाला पूल कहा जाता है। संक्षेप में, कार्य-चोरी एक एल्गोरिथ्म है जिसमें निष्क्रिय थ्रेड्स अन्य थ्रेड्स से कार्य लेना शुरू करते हैं या साझा कतार से कार्य करते हैं। आइए एक उदाहरण देखें:

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

यदि हम इस कोड को चलाते हैं, तो ExecutorServiceहमारे लिए 5 थ्रेड्स बनेंगे, क्योंकि प्रत्येक थ्रेड को लॉक ऑब्जेक्ट के लिए प्रतीक्षा कतार में रखा जाएगा। हम पहले से ही बेटर में एक साथ मॉनिटर और लॉक का पता लगा चुके हैं: जावा और थ्रेड क्लास। भाग II - तुल्यकालन । Executors.newCachedThreadPool()अब के साथ बदलें Executors.newWorkStealingPool()। क्या बदलेगा? हम देखेंगे कि हमारे कार्यों को 5 से कम धागों पर निष्पादित किया जाता है। याद रखें कि CachedThreadPoolप्रत्येक कार्य के लिए एक सूत्र बनाता है? ऐसा इसलिए है क्योंकि wait()धागे को अवरुद्ध कर दिया गया है, बाद के कार्यों को पूरा करना चाहते हैं, और पूल में उनके लिए नए धागे बनाए गए हैं। चोरी करने वाले पूल के साथ, धागे हमेशा के लिए बेकार नहीं रहते। वे अपने पड़ोसियों के कार्य करने लगते हैं। a को WorkStealingPoolअन्य थ्रेड पूल से इतना अलग क्या बनाता है? तथ्य यह है कि जादुईForkJoinPoolइसके अंदर रहता है:

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

दरअसल, एक और अंतर है। डिफ़ॉल्ट रूप से, a के लिए बनाए गए थ्रेड ForkJoinPoolडेमन थ्रेड होते हैं, एक ऑनरडिनरी के माध्यम से बनाए गए थ्रेड के विपरीत ThreadPool। सामान्य तौर पर, आपको डेमन थ्रेड्स को याद रखना चाहिए, क्योंकि, उदाहरण के लिए, CompletableFutureडेमन थ्रेड्स का भी उपयोग करता है जब तक कि आप अपने स्वयं के थ्रेड्स निर्दिष्ट नहीं करते हैं ThreadFactoryजो गैर-डेमन थ्रेड्स बनाता है। ये आश्चर्य हैं जो अप्रत्याशित स्थानों में दुबक सकते हैं! :)

फोर्कजॉइनपूल

इस भाग में, हम फिर से ForkJoinPool(जिसे फोर्क/जॉइन फ्रेमवर्क भी कहा जाता है) के बारे में बात करेंगे, जो WorkStealingPool. सामान्य तौर पर, फोर्क/जॉइन फ्रेमवर्क जावा 1.7 में वापस दिखाई दिया। और भले ही Java 11 निकट है, फिर भी यह याद रखने योग्य है। यह सबसे आम कार्यान्वयन नहीं है, लेकिन यह काफी दिलचस्प है। वेब पर इसके बारे में एक अच्छी समीक्षा है: उदाहरण के साथ जावा फोर्क-जॉइन फ्रेमवर्क को समझना । ForkJoinPoolपर निर्भर करता है java.util.concurrent.RecursiveTask। वहाँ भी है java.util.concurrent.RecursiveAction। RecursiveActionपरिणाम वापस नहीं करता है। इस प्रकार, RecursiveTaskके समान है Callable, और RecursiveActionके समान है unnable। हम देख सकते हैं कि नाम में दो महत्वपूर्ण विधियों के नाम शामिल हैं: forkऔर join। forkविधि कुछ कार्य को एक अलग थ्रेड पर अतुल्यकालिक रूप से प्रारंभ करती है। और joinविधि आपको काम पूरा होने की प्रतीक्षा करने देती है। सर्वोत्तम समझ प्राप्त करने के लिए, आपको Java 8 में इंपीरेटिव प्रोग्रामिंग से फोर्क/ज्वाइन टू पैरेलल स्ट्रीम पढ़ना चाहिए ।

सारांश

खैर, यह समीक्षा के इस भाग को समाप्त करता है। हमने सीखा है कि Executorमूल रूप से धागे को निष्पादित करने के लिए आविष्कार किया गया था। तब जावा के रचनाकारों ने विचार जारी रखने का फैसला किया और साथ आए ExecutorService। ExecutorServiceहमें submit()और का उपयोग करके निष्पादन के लिए कार्य सबमिट करने देता है invoke(), और सेवा को बंद भी कर देता है। क्योंकि ExecutorServiceकार्यान्वयन की आवश्यकता है, उन्होंने फ़ैक्टरी विधियों के साथ एक वर्ग लिखा और इसे Executors. यह आपको थ्रेड पूल ( ThreadPoolExecutor) बनाने देता है। इसके अतिरिक्त, थ्रेड पूल हैं जो हमें निष्पादन शेड्यूल निर्दिष्ट करने की अनुमति भी देते हैं। और ForkJoinPoola, a के पीछे छिप जाता है WorkStealingPool। मुझे आशा है कि आपने ऊपर जो लिखा है वह न केवल दिलचस्प है, बल्कि समझने योग्य भी है :) मुझे आपके सुझावों और टिप्पणियों को सुनकर हमेशा खुशी होती है। बेहतर एक साथ: जावा और थ्रेड क्लास। भाग I - थ्रेड्स ऑफ़ एक्जीक्यूशन बेहतर एक साथ: जावा और थ्रेड क्लास। भाग II - तुल्यकालन बेहतर एक साथ: जावा और थ्रेड वर्ग। भाग III - इंटरेक्शन बेटर टुगेदर: जावा और थ्रेड क्लास। भाग IV - कॉल करने योग्य, भविष्य और मित्र बेहतर एक साथ: जावा और थ्रेड वर्ग। भाग VI - आग बुझाओ!