Concurrentlinkedqueue con wait () y notificar ()

Question

No estoy bien versado en múltiples subprocesos. Estoy tratando de tomar una captura de pantalla repetidamente por un hilo de productor, que agrega el BufferedImage oponerse a ConcurrentLinkedQueue y un hilo del consumidor poll cola para BufferedImage OBJETO para guardarlos en el archivo. Podría consumirlos mediante encuestas repetidas (mientras bucle), pero no sé cómo consumirlos usando notify() y wait(). He intentado usar wait() y notify En programas más pequeños, pero no pudo implementarlo aquí.

Tengo el siguiente código:

class StartPeriodicTask implements Runnable {
    public synchronized void run() {
        Robot robot = null;
        try {
            robot = new Robot();
        } catch (AWTException e1) {
            e1.printStackTrace();
        }
        Rectangle screenRect = new Rectangle(Toolkit.getDefaultToolkit()
                .getScreenSize());
        BufferedImage image = robot.createScreenCapture(screenRect);
        if(null!=queue.peek()){
            try {
                System.out.println("Empty queue, so waiting....");
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }else{
            queue.add(image);
            notify();
        }
    }
}

public class ImageConsumer implements Runnable {
        @Override
        public synchronized void run() {
            while (true) {
                BufferedImage bufferedImage = null;
                if(null==queue.peek()){
                    try {
                        //Empty queue, so waiting....
                        wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }else{
                    bufferedImage = queue.poll();
                    notify();
                }
                File imageFile = getFile();
                if (!imageFile.getParentFile().exists()) {
                    imageFile.getParentFile().mkdirs();
                }
                    try {
                        ImageIO.write(bufferedImage, extension, imageFile);
                        //Image saved
                    catch (IOException e) {
                        tracer.severe("IOException occurred. Image is not saved to file!");
                    }
                }
            }

Anteriormente tenía una encuesta repetida para verificar la existencia de BufferedImage Objeto. Ahora he cambiado run método como synchronised e intenté implementar wait() y notify(). ¿Estoy correctamente? Por favor ayuda. Gracias.

Answer 1

Estás usando el mal Queue para el trabajo. los ConcurrentLinkedQueue es una cola sin bloqueo, lo que significa que no hay semántica del consumidor de productores. Si solo está haciendo un lector y un escritor eche un vistazo a Synchronousqueue

Simplemente coloque su código se puede reescribir como tal

BlockingQueue<?> queue = new SynchrnousQueue<?>();
class StartPeriodicTask implements Runnable {
    public void run() {
        Robot robot = null;
        try {
            robot = new Robot();
        } catch (AWTException e1) {
            e1.printStackTrace();
        }
        Rectangle screenRect = new Rectangle(Toolkit.getDefaultToolkit()
                .getScreenSize());
        BufferedImage image = robot.createScreenCapture(screenRect);
        queue.offer(image); //1
}
public class ImageConsumer implements Runnable {
        @Override
        public void run() {
            while (true) {
                BufferedImage bufferedImage = queue.poll(); //2

                File imageFile = getFile();
                if (!imageFile.getParentFile().exists()) {
                    imageFile.getParentFile().mkdirs();
                }
                    try {
                        ImageIO.write(bufferedImage, extension, imageFile);
                        //Image saved
                    catch (IOException e) {
                        tracer.severe("IOException occurred. Image is not saved to file!");
                    }
            }

Eso es realmente todo.

Dejame explicar. En línea // 1 el hilo productor 'colocará' la imagen en la cola. Yo cita el lugar porque un synchrnousqueue no tiene profundidad. Lo que realmente sucede es que el hilo le dice a la cola "Si hay algún hilo pidiendo un elemento de esta cola, entonces dale el hilo y déjame continuar. Si no, esperaré hasta que otro hilo esté listo"

La línea // 2 es similar a 1 donde el hilo consumidor solo espera hasta que se ofrece un hilo. Esto funciona muy bien con un solo escritor de lectores

Answer 2

El primer problema es la espera innecesaria que tiene en su productor:

    if(null!=queue.peek()){ // You are the producer, you don't care if the queue is empty
        try {
            System.out.println("Empty queue, so waiting....");
            wait(); // This puts you to bed, your waiting and so is your consumer
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }else{
        queue.add(image);
        notify();
    }

Esto es todo lo que debes necesitar:

        queue.add(image);
        notify();

El siguiente problema es el innecesario notify en tu consumidor. Produce el control de su procesamiento en ese momento, lo que creo que pretendía como una forma de poner en marcha a su productor, pero por supuesto, su código nunca llega a ese punto. Así que esto:

            }else{
                bufferedImage = queue.poll();
                notify();
            }
            File imageFile = getFile();
            if (!imageFile.getParentFile().exists()) {
                imageFile.getParentFile().mkdirs();
            }
                try {
                    ImageIO.write(bufferedImage, extension, imageFile);
                    //Image saved
                catch (IOException e) {
                    tracer.severe("IOException occurred. Image is not saved to file!");
                }
            }

Debería verse más así:

            }else{
                bufferedImage = queue.poll();

                File imageFile = getFile();
                if (!imageFile.getParentFile().exists()) {
                   imageFile.getParentFile().mkdirs();
                }

                try {
                    ImageIO.write(bufferedImage, extension, imageFile);
                    //Image saved
                catch (IOException e) {
                    tracer.severe("IOException occurred. Image is not saved to file!");
                }
            }

Answer 3

Una vez el java.util.concurrent La biblioteca entró en el JDK1.5, la necesidad de escribir su propia espera/notificación Logic salió por la puerta. En 2012, si está haciendo su propia espera/notificación, está trabajando demasiado duro y debe considerar fuertemente los equivalentes probados y verdaderos Java.util.concurrent.

Dicho esto, creo que las encuestas es la idea detrás del incorporado java.util.concurrent.ConcurrentLinkedQueue. En otras palabras, los consumidores se sientan en su propio hilo y. !isEmpty(). La mayoría de las implementaciones que he visto arrojan algún tipo de sueño de un segundo entre las pruebas de la !isEmpty(), pero no creo que eso sea realmente necesario. Además, paga nota al comentario del chico vint sobre mi respuesta, .poll() puede regresar null. Considere implementaciones alternativas de java.util.AbstractQueue Eso puede tener un comportamiento de bloqueo más cerca de lo que está buscando.

Este tipo tiene un ejemplo simple: http://www.informit.com/articles/article.aspx?p=1339471&seqnum=4

Finalmente, obtenga el libro de Goetz "Java Concurrencia en la práctica" y léelo. Estoy casi seguro de que tiene una receta para qué usar para reemplazar sus propias esperas/notificaciones locales.