producteur/consommateur files d'attente de travail

Je suis à la lutte avec la meilleure façon de mettre en œuvre mes pipeline de traitement.

Mes producteurs flux de travail à un BlockingQueue. Du côté du consommateur, je interrogation de la file d'attente, l'envelopper de ce que je reçois dans un Praticable de la tâche, et la soumettre à un ExecutorService.

while (!isStopping())
{
    String work = workQueue.poll(1000L, TimeUnit.MILLISECONDS);
    if (work == null)
    {
        break;
    }
    executorService.execute(new Worker(work));   //needs to block if no threads!
}

Ce n'est pas l'idéal; le ExecutorService a sa propre file d'attente, bien sûr, de sorte que ce qui se passe réellement, c'est que je suis toujours pleinement vidange de ma file d'attente de travail et de remplissage de la file d'attente des tâches, qui se vide lentement que les tâches sont terminées.

Je me rends compte que je pouvais tâches de file d'attente à la production de la fin, mais j'aurais vraiment préféré ne pas le faire - j'aime l'indirection/isolation de ma file d'attente de travail étant muet cordes; ce n'est pas vraiment toute entreprise du producteur, ce qui va leur arriver. En forçant le producteur de la file d'attente d'un Exécutable ou Remboursables par anticipation les sauts de une abstraction, à mon humble avis.

Mais je ne veux pas le travail en commun de la file d'attente pour représenter l'état actuel du traitement. Je veux être en mesure de bloquer les producteurs si les consommateurs ne suivent pas.

J'aimerais utiliser les Exécuteurs, mais je sens que je suis en lutte contre leur conception. Puis-je partiellement boire le Kool-ade, ou dois-je avaler? Suis-je mal posé, à résister à des files d'attente des tâches? (Je pense que je pourrait mettre en place ThreadPoolExecutor d'utiliser un 1-tâche de la file d'attente et de remplacer la méthode execute pour bloquer plutôt que de le rejeter-sur-file d'attente-plein, mais qui se sent brut.)

Suggestions?

Exactement ma pensée. Merci de mettre cela par écrit. Intéressant qu'il n'y a pas de bibliothèque.

OriginalL'auteur Jolly Roger | 2010-02-10

  1. 17

    Je veux que le travail en commun de la file d'attente à
    représentent le traitement en cours
    état.

    Essayez d'utiliser un partagées BlockingQueue et de disposer d'un pool de threads de travail en prenant les éléments de travail de la File d'attente.

    Je veux être en mesure de bloquer l'
    producteurs si les consommateurs ne sont pas
    en gardant en place.

    Les deux ArrayBlockingQueue et LinkedBlockingQueue support borné files d'attente, de sorte qu'ils se bloquer sur les mettez quand il est plein. En utilisant le blocage de put() méthodes garantit que les producteurs sont bloqués si la file d'attente est pleine.

    Voici un début difficile. Vous pouvez régler le nombre de travailleurs et de la taille de file d'attente:

    public class WorkerTest<T> {

    private final BlockingQueue<T> workQueue;
    private final ExecutorService service;

    public WorkerTest(int numWorkers, int workQueueSize) {
        workQueue = new LinkedBlockingQueue<T>(workQueueSize);
        service = Executors.newFixedThreadPool(numWorkers);

        for (int i=0; i < numWorkers; i++) {
            service.submit(new Worker<T>(workQueue));
        }
    }

    public void produce(T item) {
        try {
            workQueue.put(item);
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
        }
    }


    private static class Worker<T> implements Runnable {
        private final BlockingQueue<T> workQueue;

        public Worker(BlockingQueue<T> workQueue) {
            this.workQueue = workQueue;
        }

        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    T item = workQueue.take();
                    //Process item
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }
}
    Merci; mon précédent de mise en œuvre a été un peu comme ça, même si il a juste utilisé un ThreadFactory - une fois que vous réduire à un ensemble fixe de threads qui essayer de vider la file d'attente de travail, il y a peu d'intérêt à l'aide de ExecutorService plus. J'ai été de commutation à ExecutorService afin de profiter d'une plus accordable pool de threads, avec la sémantique de "trouver une disposition existante thread de travail si celle-ci existe, créez-en un si nécessaire, de les tuer si ils vont en veille".
    Les Exécuteurs Testamentaires.newCachedThreadPool() va faire quelque chose de semblable. Vous pouvez aussi vraiment affiner la politique de la piscine sur ThreadPoolExecutor lui-même. Qu'est-ce que vous êtes après?
    C'est l'idée... il peut être paramétrée exactement de la façon dont je l'aime, si j'étais disposé à utiliser son travail dans les tâches de la file d'attente. Ce que je veux vraiment, c'est de tailler le pool de threads smarts de l'exécuteur testamentaire et de mettre en œuvre mon propre pool de threads client, mais il n'est pas vraiment mis en place pour cela.
    Pourquoi n'êtes-vous pas prêt à utiliser son travail dans les tâches de la file d'attente?
    Pas besoin de plusieurs Worker cas, non?

    OriginalL'auteur Kevin

  2. 1

    "trouver un existant thread de travail si celle-ci existe, créez-en un si nécessaire, de les tuer si ils vont en veille".

    La gestion de tous ces travailleurs unis est aussi inutile que c'est périlleux. Je voudrais créer un thread de contrôle qui fonctionne en permanence en arrière-plan, qui est la seule tâche est de remplir la file d'attente et de frayer les consommateurs... pourquoi ne pas faire les threads de travail démons de sorte qu'ils meurent dès qu'ils complets? Si vous attachez tous à un seul ThreadGroup vous pouvez redimensionner dynamiquement la piscine... par exemple:

      **for(int i=0; i<queue.size()&&ThreadGroup.activeCount()<UPPER_LIMIT;i++ { 
         spawnDaemonWorkers(queue.poll());
  }**

    OriginalL'auteur Melissa

  3. 0

    Vous pourriez avoir votre consommation d'exécuter Runnable::run directement au lieu de commencer un nouveau thread. Combinez cela avec une file d'attente de blocage avec une taille maximum et je pense que vous allez obtenir ce que vous voulez. Votre consommateur devient un travailleur qui est de l'exécution des tâches inline basé sur les éléments de travail dans la file d'attente. Ils ne file d'attente des éléments aussi vite qu'ils les traitent de sorte que votre producteur lors de vos consommateurs de cesser de consommer de l'.

    Qui ne fera que donner un ouvrier, et je tiens à le paramétrer pour optimiser le traitement d'une donnée de configuration du système. Travaux de traitement va durer entre une fraction de seconde et des dizaines de secondes, et d'un mélange de I/O-lié et le CPU de travail.
    J'ai été en supposant que plusieurs consommateurs associés à un ou plusieurs producteurs. Si vous n'avez qu'un seul consommateur, alors, pourquoi ne pas le producteur de vidage du travail directement dans l'exécuteur testamentaire?

    OriginalL'auteur D.Shawley