求助，线程控制的问题

A线程取一批数据处理，然后将结果交给B，C，D三个线程继续处理，
等B、C、D三个线程处理完成后A才能取下一批数据，同样如果A没有取下一批数据，
B、C、D不能去执行。
如何用wait、notify、notifyAll合理控制，此处四个线程都是在循环执行

求指点，要怎么修改

 package test;

public class A extends Thread {

    public static Object my = null;
    public static Object my1 = null;
    public static Object my2 = null;
    public static Object my3 = null;

    @Override
    public void run() {
        while (true) {
            my = new Object();
            my1 = new Object();
            my2 = new Object();
            my3 = new Object();

            System.out.println("执行A");
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            synchronized(A.my) {
                A.my.notifyAll();
            }
            synchronized(my1) {
                try {
                    A.my1.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            synchronized(A.my2) {
                try {
                    A.my2.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            synchronized(A.my3) {
                try {
                    A.my3.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void main(String[] args) {
        A a = new A();
        a.start();
        B1 b1 = new B1();
        b1.start();
        B2 b2 = new B2();
        b2.start();
        B3 b3 = new B3();
        b3.start();
    }
}

class B1 extends Thread {
    @Override
    public void run() {
        while (true) {
            try {
                synchronized(A.my) {
                    A.my.wait();
                }
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
            System.out.println("执行B");
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            synchronized(A.my1) {
                A.my1.notify();
            }
        }
    }
}

class B2 extends Thread {
    @Override
    public void run() {
        while (true) {
            try {
                synchronized(A.my) {
                    A.my.wait();
                }
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
            System.out.println("执行C");
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            synchronized(A.my2) {
                A.my2.notify();
            }
        }
    }
}

class B3 extends Thread {
    @Override
    public void run() {
        while (true) {
            try {
                synchronized(A.my) {
                    A.my.wait();
                }
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
            System.out.println("执行D");
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            synchronized(A.my3) {
                A.my3.notify();
            }
        }
    }
}

简单的生产者消费者模式的例子，可以参考：http://blog.csdn.net/wojiushiwo945you/article/details/42262149
只要弄清楚这几个方法的用法，这个需求其实不难的。

import java.util.concurrent.CountDownLatch;

public class A extends Thread {

public static Object my = new Object();// 不要使用多个锁，使控制复杂导致死锁，尽量使用1个锁。
public static CountDownLatch countDownLatch = new CountDownLatch(3);

@Override
public void run() {
    while (true) {

// my = new Object();// 锁的实例化不要放到循环里，会使锁的引用一直变化。锁应该始终保持引用同一对象。
// my1 = new Object();
// my2 = new Object();
// my3 = new Object();

        try {
            countDownLatch.await();
        } catch (InterruptedException e1) {

            e1.printStackTrace();
        }

        System.out.println("执行A");
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        synchronized(A.my) {
            countDownLatch = new CountDownLatch(3);
            A.my.notifyAll();
        }
    }
}

public static void main(String[] args) {
    A a = new A();
    a.start();
    B1 b1 = new B1();
    b1.start();
    B2 b2 = new B2();
    b2.start();
    B3 b3 = new B3();
    b3.start();
}

}

class B1 extends Thread {
@Override
public void run() {
while (true) {
try {
synchronized(A.my) {
A.countDownLatch.countDown();
A.my.wait();
}
} catch (InterruptedException e1) {
e1.printStackTrace();
}
System.out.println("执行B");
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

class B2 extends Thread {
@Override
public void run() {
while (true) {
try {
synchronized(A.my) {
A.countDownLatch.countDown();
A.my.wait();
}
} catch (InterruptedException e1) {
e1.printStackTrace();
}
System.out.println("执行C");
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

class B3 extends Thread {
@Override
public void run() {
while (true) {
try {
synchronized(A.my) {
A.countDownLatch.countDown();
A.my.wait();
}
} catch (InterruptedException e1) {
e1.printStackTrace();
}
System.out.println("执行D");
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

这个问题我也不是很清楚，不过看上面的答案学习了下CountDownLatch和CyclicBarrier，楼主可以看看。