带有wait()和notify()的ConcurrentLinkedQueue

Question

我并不精通多线程.我试图通过一个生产者线程重复截取屏幕截图,该线程将BufferedImage对象添加到,ConcurrentLinkedQueue并且消费者线程将为对象poll排队BufferedImage以将它们保存在文件中.我可以通过重复轮询(while循环)来消耗它们,但我不知道如何使用notify()和消耗它们wait().我曾尝试使用wait(),并notify在较小的项目,但不能在这里实现它.

我有以下代码:

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!");
                    }
                }
            }

以前我有一个重复的轮询来检查BufferedImageObject的存在.现在我已经改变了run方法,synchronised并试图实现wait()和notify().我做得对吗？请帮忙.谢谢.

Answer 1

第一个问题是您在生产者中不必要的等待:

    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();
    }

这就是你应该需要的:

        queue.add(image);
        notify();

下一个问题是notify您的消费者不必要的.它在那时产生了对其处理的控制,我相信你的目的是让你的生产者继续前进,但当然你的代码永远不会达到这一点.所以这:

            }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!");
                }
            }

应该看起来更像这样:

            }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 2

你正在使用错误Queue的工作.这ConcurrentLinkedQueue是一个非阻塞队列,这意味着没有生成者消费者语义.如果您只是做一个读者和一个作家,请看看SynchronousQueue

简单地说你的代码可以像这样重写

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!");
                    }
            }

真的是这样的.

让我解释.在第//行,生产线程将"放置"在队列上的图像.我引用的地方是因为SynchrnousQueue没有深度.实际发生的是线程告诉队列"如果有任何线程要求来自此队列的元素然后给它该线程并让我继续.如果不是,我会等到另一个线程准备好"

第// 2行与1类似,其中消费线程只等待线程提供.这对于单读者单作者来说非常有用