java - 使用 wait/notifyAll 正确同步 Java 线程？

Question

这是我的应用程序的简化版本，显示了我正在做的事情。

/*
in my app's main():

    Runner run = new Runner();

    run.dowork();

*/

class Runner
{
    private int totalWorkers = 2;
    private int workersDone = 0;

    public synchronized void workerDone()
    {
        workersDone++;
        notifyAll();
    }

    public synchronized void dowork()
    {
        workersDone = 0;

        //<code for opening a file here, other setup here, etc>

        Worker a = new Worker(this);
        Worker b = new Worker(this);

        while ((line = reader.readLine()) != null)
        {
            //<a large amount of processing on 'line'>

            a.setData(line);
            b.setData(line);

            while (workersDone < totalWorkers)
            {
                wait();
            }               
        }
    }
}

class Worker implements Runnable
{
    private Runner runner;
    private String data;

    public Worker(Runner r)
    {
        this.runner = r;
        Thread t = new Thread(this);
        t.start();
    }

    public synchronized void setData(String s)
    {
        this.data = s;
        notifyAll();
    }

    public void run
    {
        while (true)
        {
            synchronized(this)
            {
                wait();

                //<do work with this.data here>

                this.runner.workerDone();
            }
        }
    }
}

这里的基本概念是，我有一群工作人员，他们都独立地对传入的数据行进行一些处理，并在他们喜欢的任何地方写出数据 - 他们不需要将任何数据报告回主线程或彼此共享数据。

我遇到的问题是这段代码死锁。我正在读取一个超过 100 万行的文件，很幸运在我的应用停止响应之前读取了 100 行。

实际上， worker 所做的工作量都不同，因此我想等到他们全部完成后再转到下一行。

我不能让工作人员以不同的速度处理数据并在内部对数据进行排队，因为我正在处理的文件对于此来说太大并且不适合内存。

我无法为每个工作人员提供自己的 FileReader 来独立获取“行”，因为在工作人员看到该行之前我会对该行进行大量处理，并且不希望在每个工作人员中重新进行处理。

我知道我错过了 Java 同步的一些相当简单的方面，但我陷入了这一点。如果有人能解释我在这里做错了什么，我将不胜感激。我相信我误解了同步的某些方面，但我没有办法尝试修复它。

Answer 1

直接与 synchronized 合作, wait() ，和notify()绝对是棘手的。

幸运的是Java Concurrency API为此类事情提供了一些更直观的优秀控制对象。特别是看 CyclicBarrier 和 CountDownLatch ;其中之一几乎肯定就是您正在寻找的。

您还可以找到 ThreadPoolExecutor 可以方便地应对这种情况。

这是一个简单的示例/代码片段的转换，它会产生以下输出(当然没有死锁):

Read line: Line 1
Waiting for work to be complete on line: Line 1
Working on line: Line 1
Working on line: Line 1
Read line: Line 2
Waiting for work to be complete on line: Line 2
Working on line: Line 2
Working on line: Line 2
Read line: Line 3
Waiting for work to be complete on line: Line 3
Working on line: Line 3
Working on line: Line 3
All work complete!

public class Runner
{

    public static void main(String args[]) {
        Runner r = new Runner();
        try {
            r.dowork();
        } catch (IOException e) {
            // handle
            e.printStackTrace();
        }
    }

    CyclicBarrier barrier;
    ExecutorService executor;
    private int totalWorkers = 2;

    public Runner() {
        this.barrier = new CyclicBarrier(this.totalWorkers + 1);
        this.executor = Executors.newFixedThreadPool(this.totalWorkers);
    }

    public synchronized void dowork() throws IOException
    {
        //<code for opening a file here, other setup here, etc>
        //BufferedReader reader = null;
        //String line;

        final Worker worker = new Worker();

        for(String line : new String[]{"Line 1", "Line 2", "Line 3"})
        //while ((line = reader.readLine()) != null)
        {
            System.out.println("Read line: " + line);
            //<a large amount of processing on 'line'>

            for(int c = 0; c < this.totalWorkers; c++) {
                final String curLine = line;
                this.executor.submit(new Runnable() {
                    public void run() {
                        worker.doWork(curLine);
                    }
                });
            }

            try {
                System.out.println("Waiting for work to be complete on line: " + line);
                this.barrier.await();
            } catch (InterruptedException e) {
                // handle
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                // handle
                e.printStackTrace();
            }
        }

        System.out.println("All work complete!");
    }

    class Worker
    {
        public void doWork(String line)
        {
            //<do work with this.data here>
            System.out.println("Working on line: " + line);

            try {
                Runner.this.barrier.await();
            } catch (InterruptedException e) {
                // handle
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                // handle
                e.printStackTrace();
            }
        }
    }    
}