ایک ساتھ بہتر: جاوا اور تھریڈ کلاس۔ حصہ V — ایگزیکیوٹر، تھریڈ پول، فورک/جوائن

تعارف

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

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

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()کہ تھریڈ کو بلاک کر دیا گیا ہے، اس کے بعد کے کام مکمل ہونا چاہتے ہیں، اور پول میں ان کے لیے نئے تھریڈز بنائے گئے تھے۔ چوری کے تالاب کے ساتھ، دھاگے ہمیشہ کے لیے بیکار نہیں رہتے۔ وہ اپنے پڑوسیوں کے کام انجام دینے لگتے ہیں۔ کیا چیز WorkStealingPoolدوسرے دھاگے کے تالابوں سے اتنا مختلف ہے؟ حقیقت یہ ہے کہ جادو ForkJoinPoolاس کے اندر رہتا ہے:

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

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

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

اس حصے میں، ہم دوبارہ بات کریں گے ForkJoinPool(جسے فورک/جوائن فریم ورک بھی کہا جاتا ہے)، جو کہ "ہڈ کے نیچے" رہتا ہے WorkStealingPool۔ عام طور پر، جاوا 1.7 میں فورک/جوائن فریم ورک دوبارہ ظاہر ہوا۔ اور اگرچہ جاوا 11 قریب ہے، یہ اب بھی یاد رکھنے کے قابل ہے۔ یہ سب سے عام نفاذ نہیں ہے، لیکن یہ کافی دلچسپ ہے۔ ویب پر اس کے بارے میں ایک اچھا جائزہ ہے: Java Fork-Join Framework with Examples کو سمجھنا ۔ ForkJoinPoolپر انحصار کرتا ہے java.util.concurrent.RecursiveTask۔ وہاں بھی ہے java.util.concurrent.RecursiveAction۔ RecursiveActionنتیجہ واپس نہیں کرتا. اس طرح، RecursiveTaskسے ملتا جلتا ہے Callable، اور RecursiveActionاس سے ملتا جلتا ہے unnable۔ ہم دیکھ سکتے ہیں کہ نام میں دو اہم طریقوں کے نام شامل ہیں: forkاور join. یہ forkطریقہ کچھ کام کو ایک الگ تھریڈ پر متضاد طور پر شروع کرتا ہے۔ اور joinطریقہ آپ کو کام کرنے کا انتظار کرنے دیتا ہے۔ بہترین تفہیم حاصل کرنے کے لیے، آپ کو جاوا 8 میں Imperative Programming سے Fork/Join to Parallel Streams تک پڑھنا چاہیے ۔

خلاصہ

ٹھیک ہے، یہ جائزہ کے اس حصے کو سمیٹتا ہے۔ ہم نے سیکھا ہے کہ Executorاصل میں دھاگوں کو چلانے کے لیے ایجاد کیا گیا تھا۔ پھر جاوا کے تخلیق کاروں نے اس خیال کو جاری رکھنے کا فیصلہ کیا اور اس کے ساتھ آئے ExecutorService۔ ہمیں اور ExecutorServiceکا استعمال کرتے ہوئے عمل درآمد کے لیے ٹاسک جمع کرنے دیتا ہے ، اور سروس کو بھی بند کرتا ہے۔ چونکہ عمل درآمد کی ضرورت ہے، انہوں نے فیکٹری کے طریقوں کے ساتھ ایک کلاس لکھی اور اسے بلایا ۔ یہ آپ کو تھریڈ پول ( ) بنانے دیتا ہے۔ مزید برآں، تھریڈ پولز ہیں جو ہمیں عمل درآمد کا شیڈول بتانے کی بھی اجازت دیتے ہیں۔ اور ایک کے پیچھے چھپ جاتا ہے ۔ مجھے امید ہے کہ آپ نے جو کچھ میں نے اوپر لکھا ہے وہ نہ صرف دلچسپ بلکہ قابل فہم بھی ہے :) مجھے آپ کی تجاویز اور تبصرے سن کر ہمیشہ خوشی ہوتی ہے۔ ایک ساتھ بہتر: جاوا اور تھریڈ کلاس۔ حصہ اول — عمل درآمد کے دھاگے ایک ساتھ بہتر: جاوا اور تھریڈ کلاس۔ حصہ دوم — ہم آہنگی ایک ساتھ بہتر: جاوا اور تھریڈ کلاس۔ حصہ III - ایک دوسرے کے ساتھ بہتر تعامل: جاوا اور تھریڈ کلاس۔ حصہ چہارم — کال کے قابل، مستقبل، اور دوست ایک ساتھ بہتر: جاوا اور تھریڈ کلاس۔ حصہ VI - آگ دور! submit()invoke()ExecutorServiceExecutorsThreadPoolExecutorForkJoinPoolWorkStealingPool