آپ جاوا کو تھریڈ سے خراب نہیں کر سکتے: حصہ V - ایگزیکیوٹر، تھریڈ پول، فورک جوائن

تعارف

لہذا، ہم جانتے ہیں کہ جاوا میں تھریڈز ہیں، جن کے بارے میں آپ جائزہ میں پڑھ سکتے ہیں " آپ جاوا کو تھریڈ کے ساتھ خراب نہیں کر سکتے: حصہ اول - تھریڈز "۔ آئیے نمونہ کوڈ کو دوبارہ دیکھیں:

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 کہتا ہے کہ ExecutorServiceیہ ایک خاص Executor'a' کی تفصیل ہے جو کام Executor'a' کو روکنے کے طریقے فراہم کرتا ہے اور آپ کو java.util.concurrent.Futureعملدرآمد کی پیشرفت کو ٹریک کرنے کی اجازت دیتا ہے۔ اس سے پہلے، " آپ تھریڈ کے ساتھ جاوا کو خراب نہیں کر سکتے: حصہ IV - کال ایبل، فیوچر اینڈ فرینڈز، " میں ہم نے مختصراً امکانات کا جائزہ لیا تھا Future۔ اگر آپ اسے بھول گئے ہیں یا نہیں پڑھا ہے تو، میں آپ کو مشورہ دیتا ہوں کہ آپ اپنی یادداشت کو تازہ کریں؛) JavaDoc میں اور کیا دلچسپ ہے؟ کہ ہمارے پاس ایک خاص کارخانہ ہے java.util.concurrent.Executorsجو ہمیں ایسے نفاذات بنانے کی اجازت دیتا ہے جو بطور ڈیفالٹ دستیاب ہوں ExecutorService۔

ایگزیکیوٹر سروس

چلو پھر یاد کرتے ہیں۔ ہمیں Executorتھریڈ میں ایک خاص کام کو انجام دینا ہوتا ہے، جب تھریڈ بنانے کا عمل ہم سے پوشیدہ ہوتا ہے۔ ہمارے پاس 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();
}

جیسا کہ ہم دیکھ سکتے ہیں، ہم نے ایک فکسڈ تھریڈ پول ( Fixed Thread Pool) سائز 2 کا تعین کیا ہے۔ جس کے بعد ہم پول کو ایک ایک کرکے کام بھیجتے ہیں۔ ہر کام ایک سٹرنگ ( String) لوٹاتا ہے جس میں تھریڈ کا نام ( currentThread().getName()) ہوتا ہے۔ بالکل آخر میں بند کرنا ضروری ہے ExecutorService، کیونکہ بصورت دیگر ہمارا پروگرام نہیں نکلے گا۔ فیکٹری میں Executorsفیکٹری کے دیگر طریقے ہیں ۔ مثال کے طور پر، ہم صرف ایک دھاگے کا ایک پول بنا سکتے ہیں - newSingleThreadExecutorیا کیشنگ کے ساتھ ایک پول newCachedThreadPool، جہاں دھاگوں کو پول سے ہٹا دیا جائے گا اگر وہ 1 منٹ تک بیکار رہیں۔ درحقیقت، ان کے پیچھے ایک بلاکنگ قطارExecutorService ہے جس میں کام رکھے جاتے ہیں اور جہاں سے ان کاموں کو انجام دیا جاتا ہے۔ قطاروں کو مسدود کرنے کے بارے میں مزید معلومات ویڈیو میں دیکھی جا سکتی ہیں " Blocking queue - Collection #5 - Advanced Java "۔ آپ جائزہ " سمورتی پیکج کی قطاروں کو مسدود کرنا " اور اس سوال کا جواب بھی پڑھ سکتے ہیں " ArrayBlockingQueue پر LinkedBlockingQueue کو کب ترجیح دیں؟ " انتہائی آسان - (قطار کو مسدود کرنا) ایک دھاگے کو روکتا ہے، دو صورتوں میں: 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

جیسا کہ ہم نے پہلے دیکھا، فیکٹری کے طریقوں کے اندر ThreadPoolExecutor، . فعالیت اس بات سے متاثر ہوتی ہے کہ کن اقدار کو زیادہ سے زیادہ اور کم سے کم دھاگوں کے طور پر پاس کیا جاتا ہے، ساتھ ہی کون سی قطار استعمال ہوتی ہے۔ اور انٹرفیس کے کسی بھی نفاذ کو استعمال کیا جا سکتا ہے java.util.concurrent.BlockingQueue۔ 'ahs کی بات کرتے ہوئے 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();
}

یہ کوڈ ایک غلطی کے ساتھ ناکام ہو جائے گا جیسے:

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۔ یہ ٹائمر پر کام انجام دینے کی صلاحیت فراہم کرتا ہے۔

شیڈولڈ ایگزیکیوٹرسروس

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ایک مقررہ شرح (فکسڈ ریٹ) پر ایک خاص تاخیر کے ساتھ کام کو انجام دینے کے لیے بھیجتے ہیں۔ اس صورت میں، ہر 2 سیکنڈ میں 1 سیکنڈ کے بعد، کام کو انجام دینا شروع کریں۔ اسی طرح کا ایک آپشن ہے:

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 تھریڈز بنائے گا، کیونکہ ہر تھریڈ آبجیکٹ کے مقام پر انتظار کی قطار میں شامل ہو جائے گا lock۔ ہم پہلے ہی اس پر مانیٹر اور لاک کے بارے میں بات کر چکے ہیں " آپ جاوا کو تھریڈ کے ساتھ خراب نہیں کر سکتے: حصہ II - مطابقت پذیری ۔" اور اب ہم اسے Executors.newCachedThreadPoolسے بدل دیں گے Executors.newWorkStealingPool()۔ کیا تبدیلی آئے گی؟ ہم دیکھیں گے کہ ہمارے کام 5 دھاگوں میں نہیں بلکہ اس سے کم میں انجام پاتے ہیں۔ یاد رکھیں کہ cachedThreadPoolآپ نے ہر کام کے لیے اپنا تھریڈ بنایا ہے؟ کیونکہ waitاس نے تھریڈ کو بلاک کردیا تھا، لیکن اگلے کاموں کو انجام دینا چاہتے تھے اور ان کے لیے پول میں نئے تھریڈز بنائے گئے تھے۔ دھاگوں کی صورت میں StealingPool، وہ ہمیشہ کے لیے بیکار نہیں رہیں گے wait، وہ پڑوسی کے کاموں کو انجام دینا شروع کر دیں گے۔ یہ دوسرے تھریڈ پولز سے اتنا مختلف کیسے ہے WorkStealingPool؟ کیونکہ حقیقت میں اس کے اندر ایک جادوئی چیز موجود ہے ForkJoinPool:

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

اصل میں ایک اور فرق ہے۔ وہ تھریڈز جو ForkJoinPoolڈیفالٹ کے ذریعے بنائے جاتے ہیں ڈیمون تھریڈز ہوتے ہیں، ریگولر کے ذریعے بنائے گئے تھریڈز کے برعکس ThreadPool۔ عام طور پر، یہ ڈیمون تھریڈز کے بارے میں یاد رکھنے کے قابل ہے، کیونکہ... مثال کے طور پر، CompletableFutureڈیمون تھریڈز بھی استعمال کیے جاتے ہیں، اگر آپ اپنی اپنی وضاحت نہیں کرتے ہیں ThreadFactory، جو غیر ڈیمون تھریڈز بنائے گا۔ یہ اس قسم کی حیرتیں ہیں جو کسی غیر متوقع جگہ پر آپ کا انتظار کر سکتی ہیں!)

فورک/جوائن پول

اس حصے میں ہم اسی کے بارے میں بات کریں گے ForkJoinPool(جسے فورک/جوائن فریم ورک بھی کہا جاتا ہے) جو "ہڈ کے نیچے" رہتا ہے WorkStealingPool۔ عام طور پر، جاوا 1.7 میں فورک جوائن فریم ورک ظاہر ہوا۔ اور یہاں تک کہ اگر جاوا 11 پہلے ہی صحن میں ہے، یہ اب بھی یاد رکھنے کے قابل ہے۔ سب سے عام کام نہیں، لیکن کافی دلچسپ۔ انٹرنیٹ پر اس موضوع پر ایک اچھا جائزہ ہے: " جاوا 7 میں فورک/جوائن فریم ورک "۔ Fork/JoinPoolاپنے کام میں اس طرح کے تصور کے ساتھ کام کرتا ہے java.util.concurrent.RecursiveTask۔ ایک ینالاگ بھی ہے - java.util.concurrent.RecursiveAction. Recursive Actions نتیجہ واپس نہیں کرتے۔ اس طرح RecursiveTaskسے مماثل ہے Callable، اور RecursiveActionاس سے ملتا جلتا ہے Runnable۔ ٹھیک ہے، نام کو دیکھتے ہوئے، ہمیں دو اہم طریقے نظر آتے ہیں - forkاور join. یہ طریقہ forkایک الگ تھریڈ میں ایک کام کو متضاد طور پر چلاتا ہے۔ اور طریقہ joinآپ کو کام مکمل ہونے کا انتظار کرنے کی اجازت دیتا ہے۔ اسے استعمال کرنے کے کئی طریقے ہیں: یہ تصویر Alexey Shipilev کی رپورٹ " فورک/جوائن: نفاذ، استعمال، کارکردگی " کی ایک سلائیڈ کا حصہ ہے۔ اسے واضح کرنے کے لیے، JEE CONF میں ان کی رپورٹ کو دیکھنا ضروری ہے: " فورک جوائن کے نفاذ کی خصوصیات ۔"

خلاصہ کرنا

تو، ہم یہاں ہیں، جائزہ کے اگلے حصے کو ختم کر رہے ہیں۔ ہم نے سوچا کہ ہم سب سے پہلے Executorدھاگوں کو چلانے کے لیے کیا لے کر آئے تھے۔ پھر ہم نے اس خیال کو جاری رکھنے کا فیصلہ کیا اور اس کے ساتھ آئے ExecutorService۔ ExecutorServiceآپ کو استعمال کرتے ہوئے عمل درآمد کے لیے کام بھیجنے کی اجازت دیتا ہے submitاور invokeاس کے ساتھ ساتھ سروس کو آف کر کے اس کا نظم بھی کرتا ہے۔ کیونکہ ExecutorService'ہمیں نفاذ کی ضرورت ہے، ہم نے فیکٹری کے طریقوں کے ساتھ ایک کلاس لکھی اور اسے بلایا Executors۔ یہ آپ کو تھریڈ پول بنانے کی اجازت دیتا ہے ThreadPoolExecutor۔ ایک ہی وقت میں، تھریڈ پول ہیں جو آپ کو عمل درآمد کے لیے ایک شیڈول بتانے کی بھی اجازت دیتے ہیں، لیکن WorkStealingPoolچھپاتے ہیں ForkJoinPool۔ مجھے امید ہے کہ اوپر جو کچھ لکھا گیا ہے وہ آپ کے لیے نہ صرف دلچسپ تھا بلکہ قابل فہم بھی تھا) مجھے تجاویز اور تبصرے موصول ہونے میں ہمیشہ خوشی ہوتی ہے۔ #ویاچسلاو