comment ne pas accabler java executorservice tâche de la file d'attente?

J'ai le code ci-dessous l'extrait de code, qui fonctionne très bien. Mais le problème est qu'il crée et mis sur plus de 2000 des tâches sur l'exécuteur de la file d'attente droite au large de la chauve-souris.

Je besoin de vérifier si les tâches déjà à l'exécuteur de la file d'attente sont complets, alors seulement lui donner plus de tâches. Il n'a pas à être exact, c'est à dire si la file d'attente a <10 tâches à gauche, ajouter 50 plus.

Donc l'exécuteur de tâches de la file d'attente n'ont donc de nombreuses tâches en attente, ce qui permettra également à l'arrêt() pour des travaux en temps opportun, autrement, même si elle est appelée, l'exécuteur testamentaire encore en train d'essayer de terminer tous les 2000 tâches dans la file d'attente de la première.

Quelle est la meilleure façon d'accomplir cette? merci

executor = Executors.newFixedThreadPool(numThreads);

while(some_condition==true)
{
    //if(executor < 10 tasks pending)  <---- how do i do this?
    //{                             
        for(int k=0;k<20;k++)
        {  
            Runnable worker = new MyRunnable();
            executor.execute(worker);
        }
    //}
    //else 
    //{
    //     wait(3000);
    //}
}

Mise à jour à l'aide de sémaphore:

private final Semaphore semaphore = new Semaphore(10)
executor = new ThreadPoolExecutorWithSemaphoreFromJohnExample();

while(some_condition==true)
{

        Runnable worker = new MyRunnable();
        //So at this point if semaphore is full, then while loop would PAUSE(??) until
        //semaphore frees up again.
          executor.execute(worker);   
}

OriginalL'auteur user1539050 | 2012-07-19

6

Vous pouvez utiliser un simple Sémaphore. Lors de l'envoi d'acquérir un nouveau permis et après son achèvement, le permis de permettre à quiconque attente à soumettre.
```
private final Semaphore semaphore = new Semaphore(10);//or however you want max queued at any given moment
ThreadPoolExecutor tp= new ThreadPoolExecutor(...){
      public void execute(Runnable r){
          semaphore.acquire();
          super.execute(r);
      }    
      public void afterExecute(Runnable r, Thread t){
         semaphore.release();  
         super.afterExecute(r,t);
      }
};
```
Voici donc la soumission de threads sera suspendu si il n'y a pas plus de permis disponibles.

Bon Jean. Il serait pro être recommandé à attraper et à manipuler RejectedExecutionException ou de faire attention à ce que votre sémaphore de la valeur est plus petite que le blocage de la limite de file d'attente (le cas échéant).
Aussi le sémaphore numéro (10) n'est pas le nombre de max-file d'attente, c'est le nombre de l'exécution de + en file d'attente.
Vous avez raison sur les deux, de bons points. J'ai mis cela comme un point de départ, mais ces recommandations devrait être envisagée que si les OP ont été pour le mettre en œuvre.
merci john et gary, je pense que c'est exactement ce dont j'ai besoin. Mais pour confirmer pouvez-vous prendre un coup d'oeil à ma réponse avec quelques questions supplémentaires dans le code. merci

OriginalL'auteur John Vint
8

J'ai le code ci-dessous l'extrait de code, qui fonctionne très bien. Mais le problème est qu'il crée et mis sur plus de 2000 des tâches sur l'exécuteur de la file d'attente droite au large de la chauve-souris.

Une façon de le faire est de créer votre propre ThreadPoolExecutor avec une limite de file d'attente de travail et de personnaliser la RejectedExecutionHandler sur elle. Cela vous permet d'avoir un contrôle précis sur la façon dont de nombreux emplois à la file d'attente.

Vous avez besoin du gestionnaire personnalisé parce que par défaut, si la file d'attente est pleine, le ThreadPoolExecutor.submit(...) va jeter un RejectedExecutionException. Avec le gestionnaire personnalisé ci-dessous, lorsqu'elle est rejetée par la file d'attente, le rejet de gestionnaire met juste de la lire dans, le blocage jusqu'à ce que la file d'attente a l'espace. Donc pas d'emplois sera rejetée ou abandonnée.

Voici environ la façon dont vous démarrez votre propre thread-piscine et de définir votre propre rejeter gestionnaire.
```
//you can tune the blocking queue size which is the number of jobs to queue
//when the NUM_THREADS are all working
final BlockingQueue<MyRunnable> queue =
    new ArrayBlockingQueue<MyRunnable>(NUM_JOBS_TO_QUEUE);
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(NUM_THREADS, NUM_THREADS,
       0L, TimeUnit.MILLISECONDS, queue);
//by default (unfortunately) the ThreadPoolExecutor will throw an exception
//when you submit the job that fills the queue, to have it block you do:
threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
   public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
      //this will block if the queue is full as opposed to throwing
      executor.getQueue().put(r);
   }
});
...
//now submit all of your jobs and it will block if the queue is full
for(int k = 0; k < 20000000; k++) {  
   Runnable worker = new MyRunnable();
   threadPool.execute(worker);
}
```
Voir ma réponse ici pour plus de détails sur le blocage de fil-piscines:

Comment puis-je faire ThreadPoolExecutor commande d'attente si il y a trop de données dont il a besoin pour travailler?

Vous pouvez également utiliser le ThreadPoolExecutor.CallerRunsPolicy qui serait la cause de l'appelant qui est de soumettre le travail dans le fil de la piscine pour exécuter le job. Je n'aime pas cette solution, cependant, car il bloque l'appelant jusqu'à ce que le travail de finitions qui risquaient de mourir de faim les autres threads de travail. Aussi, s'il y a plusieurs déposants, il pourrait encore causer trop de threads pour exécuter les travaux.

Enfin, notez que j'ai mis le cœur et max nombre de threads dans le ThreadPoolExecutor pour le même nombre. Malheureusement, par défaut, l'exécuteur démarre le noyau fils, puis remplit la file d'attente, et alors seulement, il n'allouer des threads supplémentaires jusqu'à max. C'est complètement contre-intuitif.

mais le problème est que je ne veux pas rejeter les tâches ou les jeter n'importe quelle question, je veux juste attendre jusqu'à ce que la tâche de la file d'attente de moins en attente puis ajouter ces tâches. Il sera okie pour créer un compteur (int tasksRunning = 0), alors à chaque fois execute() est appelée tasksRunning++ et dans l'exécutable.run() quand il finit de faire tasksRunning-- merci
Ma solution ne rejettera pas les tâches. C'est l'ensemble de point. Je vais faire ce que plus clair @user1539050.
voir aussi stackoverflow.com/questions/2001086/...

OriginalL'auteur Gray
3

J'ai l'habitude de gaz de tels systèmes à l'aide d'un objet "de la piscine de la file d'attente" pour la tâche objets BlockingQueue qui est rempli avec de la X des tâches au démarrage. Tout ce qui veut soumettre une tâche pour le fils doit en obtenir un à partir de la file d'attente du pool, de le charger avec des données et ensuite le soumettre.

Lorsque la tâche est terminée et les résultats dans ce traité, il est repoussé sur la piscine de la file d'attente pour les ré-utiliser.

Si la piscine se vide, en soumettant les blocs de threads sur le bassin de la file d'attente jusqu'à ce que certaines tâches sont retournés.

C'est essentiellement une forme de sémaphore de contrôle, tel que suggéré par @John Vint, mais il a quelques avantages - pas de continue de créer/GC de la runnables, par exemple. J'aime dump PooolQueue.la taille à une interface graphique barre d'état sur une minuterie, de sorte que je peux voir comment "occupé" le système est, (et aussi de détecter rapidement n'importe quel objet de fuites:)

OriginalL'auteur Martin James
2

Vous serez mieux de l'établissement d'un Rejet de la politique, puisque vous ne voulez pas de submerger le pool de threads, j'ai trouvé que la meilleure façon d'atteindre cet objectif sans complication bien est en faisant quelque chose comme ceci:
```
final ThreadPoolExecutor executor=(ThreadPoolExecutor)Executors.newFixedThreadPool(THREADS_COUNT);
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
```
Ce qui va arriver, c'est qu'une fois que tous les Fils sont occupées, l'appelant du thread exécute la tâche. Ici est une référence à une telle politique CallerRunsPolicy JavaDoc

Ne va pas downvote, mais veux juste noter que le casting c'est dangereux parce qu'il n'est pas nécessairement un ThreadPoolExecutor
ExecutorService qui est le type de la déclaration de l'instance ne dispose pas d'une définition pour le Rejet de l'Exécution de la politique, si vous n'avez pas de voter pour elle, vous ne serez pas en mesure de le faire. Dans ce cas, à troutle l'exécution par pas de files d'attente de plus d'emplois..
Les exécuteurs en général sont difficiles et besoin de quelques travaux de Oracle/Sun, comme le fait qu'il n'y a pas un Rejet de la politique de bloquer à la file d'attente tandis que les fils ramassent le Travail, je dois faire quelque chose de recommandé de ne PAS soutenir une telle fonctionnalité au sein de notre société, jusqu'à ce cadre écrivains à le rendre meilleur.

OriginalL'auteur Guido Medina

Vous devez vous connecter pour publier un commentaire.