File d'attente simultanée et bloquante en Java

Question

J'ai le problème classique d'un thread qui envoie des événements à la file d'attente entrante d'un second thread. Seulement cette fois, je suis très intéressé par la performance. Ce que je veux réaliser est:

• Je souhaite un accès simultané à la file d'attente, au producteur qui pousse, au destinataire qui s'ouvre.
• Lorsque la file d'attente est vide, je souhaite que le consommateur se bloque dans la file d'attente, en attente du producteur.

Mon idée première était d’utiliser un LinkedBlockingQueue , mais j’ai vite réalisé que ce n’était pas concurrent et que les performances en souffraient. Par contre, j’utilise maintenant un ConcurrentLinkedQueue , mais je paie quand même le coût de wait () / notify () sur chaque publication. . Étant donné que le consommateur, lorsqu'il trouve une file d'attente vide, ne bloque pas, je dois synchroniser et wait () sur un verrou. D'autre part, le producteur doit obtenir ce verrou et notify () à chaque publication. Le résultat global est que je paie le coût de sycnhronized (lock) {lock.notify ()} dans chaque publication, même si vous n'en avez pas besoin.

Ce dont je suppose qu’il est nécessaire ici, c’est une file d’attente bloquante et simultanée. J'imagine qu'une opération push () fonctionne comme dans ConcurrentLinkedQueue , avec un notify () supplémentaire à l'objet lorsque l'élément poussé est le premier. dans la liste. Une telle vérification, selon moi, existe déjà dans ConcurrentLinkedQueue , car pousser nécessite de se connecter à l'élément suivant. Ainsi, cela serait beaucoup plus rapide que la synchronisation à chaque fois sur le verrou externe.

Quelque chose comme cela est-il disponible / raisonnable?

Answer 1

Je pense que vous pouvez vous en tenir à java.util.concurrent.LinkedBlockingQueue indépendamment de vos doutes. C'est simultané. Bien que, je n'ai aucune idée de ses performances. Il est probable qu'une autre implémentation de BlockingQueue vous conviendra mieux. Ils ne sont pas nombreux, alors faites des tests de performance et mesurez-les.

Answer 2

Semblable à cette réponse https://stackoverflow.com/a/1212515/1102730 mais un peu différent. J'ai fini par utiliser un ExecutorService . Vous pouvez en instancier un en utilisant Executors.newSingleThreadExecutor () . J'avais besoin d'une file d'attente simultanée pour lire / écrire BufferedImages dans des fichiers, ainsi que de l'atomicité avec des lectures et des écritures. Je n'ai besoin que d'un seul thread parce que le fichier IO est beaucoup plus rapide que le fichier source, net IO. De plus, j'étais plus préoccupé par l'atomicité des actions et l'exactitude que par les performances, mais cette approche peut également être utilisée avec plusieurs threads du pool pour accélérer les choses.

Pour obtenir une image (Try-Catch-Finally est omis):

Future<BufferedImage> futureImage = executorService.submit(new Callable<BufferedImage>() {
    @Override
        public BufferedImage call() throws Exception {
            ImageInputStream is = new FileImageInputStream(file);
            return  ImageIO.read(is);
        }
    })

image = futureImage.get();

Pour enregistrer une image (Try-Catch-Finally est omis):

Future<Boolean> futureWrite = executorService.submit(new Callable<Boolean>() {
    @Override
    public Boolean call() {
        FileOutputStream os = new FileOutputStream(file); 
        return ImageIO.write(image, getFileFormat(), os);  
    }
});

boolean wasWritten = futureWrite.get();

Il est important de noter que vous devez vider et fermer vos flux dans un bloc finally. Je ne sais pas comment il se comporte par rapport aux autres solutions, mais il est assez polyvalent.

Answer 3

Je vous suggère de consulter ThreadPoolExecutor newSingleThreadExecutor. Il se chargera de garder vos tâches ordonnées pour vous et si vous soumettez Callables à votre exécuteur, vous pourrez également obtenir le comportement de blocage que vous recherchez.

Answer 4

Vous pouvez essayer LinkedTransferQueue depuis jsr166: http : //gee.cs.oswego.edu/cgi-bin/viewcvs.cgi/jsr166/src/jsr166y/

Il répond à vos besoins et génère moins de frais généraux pour les opérations d'offre / de scrutation. Comme je peux le constater d'après le code, lorsque la file d'attente n'est pas vide, elle utilise des opérations atomiques pour les éléments d'interrogation. Et lorsque la file d'attente est vide, il tourne pendant un certain temps et parque le fil en cas d'échec. Je pense que cela peut aider dans votre cas.

Answer 5

J'utilise ArrayBlockingQueue chaque fois que j'ai besoin de transmettre des données d'un thread à un autre. Utiliser les méthodes put and take (qui bloqueront si elles sont pleines / vides).

Answer 6

Voici un liste des classes implémentant BlockingQueue .

Je recommanderais de vérifier SynchronousQueue .

Comme @Rorick l'a mentionné dans son commentaire, je pense que toutes ces implémentations sont simultanées. Je pense que vos préoccupations avec LinkedBlockingQueue ne sont peut-être pas à sa place.