المهام الغير منفذة حين إستخدام ThreadPoolExecutor

ما المقصود بقولك وقد يحدث رفض لبعض المهام

ضع الشيفرة التي نتكلم عنها حتى يتضح الموضوع أكثر

تحياتي

هذا كود بتعديل بسيط مأخوذ من http://bobah.net/d4d/source-code/misc/thread-pool-executor-example-j2ee

سنلاحظ أن يحصل reject لبعض المهام

/**
 * @author Vladimir Lysyy
 */

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
//import java.util.concurrent.ExecutorService;
//import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class PoolDemo1 {
	  private static class RejectedHandler implements RejectedExecutionHandler {
		    @Override
		    public void rejectedExecution(Runnable arg0, ThreadPoolExecutor arg1) {
		      // TODO Auto-generated method stub
		      System.err.println(Thread.currentThread().getName() + " execution rejected: " + arg0);     
		    }
		  }

		  private static class Task implements Runnable {
		    private static SimpleDateFormat fmt = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss.SSS");
		    private String name;
		    private Date created;

		    public Task(String name) {
		      this.name = name;
		      this.created = new Date();
		    }

		    @Override
		    public void run() {
		      final boolean wantOverflow = true;
		      System.out.println(Thread.currentThread().getName() + " executing " + this);
		      try {
		        Thread.sleep(wantOverflow ? 50 : 10);
		      } catch (InterruptedException e) {
		        e.printStackTrace();
		      }
		      System.out.println(Thread.currentThread().getName() + " executed " + this);
		    }

		    @Override
		    public String toString() {
		      return name + ", created " + fmt.format(created);
		    }

		  }

		  public static void main(String[] args) throws InterruptedException {
		    final boolean wantExceptionOnReject = false;

//		    // fixed pool, unlimited queue
//		    ExecutorService service = Executors.newFixedThreadPool(10 /* size */);
//		    ThreadPoolExecutor executor = (ThreadPoolExecutor) service;

		    // fixed pool fixed queue
		    BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(100, true);
		    ThreadPoolExecutor executor = new ThreadPoolExecutor(
		      10, // core size
		      20, // max size
		      1, // keep alive time
		      TimeUnit.MINUTES, // keep alive time units
		      queue // the queue to use
		    );

		    // set rejected execution handler
		    // or catch exception from executor.execute (see below)
		    if (!wantExceptionOnReject) executor.setRejectedExecutionHandler(new RejectedHandler());

		    for(long i = 0;i<500 ; ++i) {
		      Task t = new Task(String.valueOf(i));
		      System.out.println(Thread.currentThread().getName() + " submitted " + t + ", queue size = " + executor.getQueue().size());
		      try {
		        executor.execute(t);
		      } catch (RejectedExecutionException e) {
		        // will be thrown if rejected execution handler
		        // is not set with executor.setRejectedExecutionHandler
		        e.printStackTrace();
		      }
		      Thread.sleep(1);
		    }
		    executor.shutdown();
		  }
		}

هناك من يستخدم WaitPolicy لمعالجة المشكلة ,كمثال http://kickjava.com/src/org/mule/config/ThreadingProfile.java.htm

لكن إذا نفذ الوقت المحدد في WaitPolicy كيف سيتم التعامل مع المهام التي لم تنفذ ؟

لماذا يتم رفض المهام يا أخي؟

يهيألي أن هذا غير منطقي

والمنطق الوحيد لعدم اتمام المهام هو وجود run-time exception

وهذا ما أعتقد أنه موجود لديك

على كل حتى تحصل على إجابة جيدة مني

يجب عليك أن تقرأ عن الموضوع وتخبرني عن أسباب الرفض وسنتحدث بعدها بإذن الله

تحياتي

اقتباس

والمنطق الوحيد لعدم اتمام المهام هو وجود run-time exception

ليس الامر كذلك .. بإمكانك كتابة مثال آخر للتأكد. في المثال لأن عدد المهمات كبير لذا يحصل رفض لبعضها بسبب ما سأذكره لاحقاً. حتى وإن كان عدد المهمات قليل لكن الوقت المستغرق طويل قد يحصل رفض لبعض المهمات.

لماذا يحصل رفض لبعض المهات ؟

حسب ما ورد في وثائق Java

اقتباس

If a request cannot be queued, a new thread is created unless this would exceed maximumPoolSize, in which case, the task will be rejected.

وهنا أيضاً:

اقتباس

Rejected tasks

New tasks submitted in method execute(java.lang.Runnable) will be rejected when the Executor has been shut down, and also when the Executor uses finite bounds for both maximum threads and work queue capacity, and is saturated. In either case, the execute method invokes the RejectedExecutionHandler.rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) method of its RejectedExecutionHandler. Four predefined handler policies are provided:

وهناك شرح عن أنواع الـpolicies المتاحة.

في المثال نحن إستخدمنا RejectedExecutionHandler قمنا بتعريفه.

بالطبع يمكننا إستخدام ThreadPoolExecutor.CallerRunsPolicy وهنا تقوم ال Thread التي إستدعت execute بتنفيذ المهام المرفوضة شرط أن الـ executor لم يغلق .. في المثال السابق سنجد أنها تفي بالغرض. لكن هذا لا يعني بأنه لن تبقى بعض المهام المرفوضة غير منفذة في حالات أخرى..

اقتباس

ThreadPoolExecutor.CallerRunsPolicy

A handler for rejected tasks that runs the rejected task directly in the calling thread of the execute method, unless the executor has been shut down, in which case the task is discarded.

أثناء بحثي في النت وجدت كتاب يشير إلى طريقة نصح بها في أكثر من نقاش.

الكتاب Java Concurrency In Practice

في الجزء الخاص بـ Saturation Policies

اقتباس

There is no predefined saturation policy to make execute block when the work queue is full. However, the same effect can be accomplished by using a Semaphore to bound the task injection rate, as shown in BoundedExecutor in Listing 8.4. In such an approach, use an unbounded queue (there's no reason to bound both the queue size and the injection rate) and set the bound on the semaphore to be equal to the pool size plus the number of queued tasks you want to allow, since the semaphore is bounding the number of tasks both currently executing and awaiting execution.

على العموم سأحاول إستخدام ما ذكر أو أقوم بتجربة الطريقة التي أشرت إليها في سؤالي .

سبب التعديل:

أخطاء في التنسيق

الصراحة أني لم أستخدم الموضوع في بيئة عمل

لذا لم أواجه هذه المشاكل

سأحاول أن أقرأ الموضوع في أقرب وقت

تحياتي