java - 即使参数已经满足，Notify() 也不起作用

Question

我的程序运行并正常等待，但没有到达通知。

 public static void main(String[] args) {
        LinkedBlockingQueue<Can> beltQ = new LinkedBlockingQueue<>();
        LinkedBlockingQueue<Can> sterilizeQ = new LinkedBlockingQueue<>();
        LinkedBlockingQueue<Can> sterilize1Q = new LinkedBlockingQueue<>();
        LinkedBlockingQueue<Can> fillingQ = new LinkedBlockingQueue<>();
        LinkedBlockingQueue<Can> sealingQ = new LinkedBlockingQueue<>();
        LinkedBlockingQueue<Can> labellingQ = new LinkedBlockingQueue<>();
        LinkedBlockingQueue<Box> packagingQ = new LinkedBlockingQueue<>();
        LinkedBlockingQueue<Bigbox> boxingQ = new LinkedBlockingQueue<>();
        LinkedBlockingQueue<Bigbox> loadingQ = new LinkedBlockingQueue<>();
        LinkedBlockingQueue<Bigbox> deliveryQ = new LinkedBlockingQueue<>();
        LinkedBlockingQueue<Bigbox> customer = new LinkedBlockingQueue<>();

        Belt b = new Belt(beltQ, boxingQ);
        Sterilize s = new Sterilize(beltQ, sterilizeQ, boxingQ);
        Sterilize1 s1 = new Sterilize1(sterilizeQ, sterilize1Q, boxingQ);
        Filling f = new Filling(sterilize1Q, fillingQ, boxingQ);
        Sealing seal = new Sealing(fillingQ, sealingQ, boxingQ);
        Labelling l = new Labelling(sealingQ, labellingQ, boxingQ);
        Packaging p = new Packaging(labellingQ, packagingQ, boxingQ);
        Boxing box = new Boxing(packagingQ, boxingQ);
        b.start();
        s.start();
        s1.start();
        f.start();
        seal.start();
        l.start();
        p.start();
        box.start();
        for (int i = 1; i < 4; i++) {
            ForkliftTruck ft = new ForkliftTruck(boxingQ, loadingQ, i);
            ft.start();
        }
        LoadingArea la = new LoadingArea(2);
        for (int i = 1; i < 4; i++) {
            DeliveryVan dv = new DeliveryVan(la, i, loadingQ, deliveryQ, customer);
            dv.start();
        }
    }

class Packaging extends Thread {

    LinkedBlockingQueue<Can> labellingQ;
    LinkedBlockingQueue<Box> packagingQ;
    LinkedBlockingQueue<Bigbox> boxingQ;

    public Packaging(LinkedBlockingQueue<Can> labellingQ, LinkedBlockingQueue<Box> packagingQ, LinkedBlockingQueue<Bigbox> boxingQ) {
        this.labellingQ = labellingQ;
        this.packagingQ = packagingQ;
        this.boxingQ = boxingQ;
    }

    synchronized public void run() {
        int boxNo = 1;
        while (true) {
            try {
                synchronized (boxingQLock) {
                while (boxingQ.size() == 1) {
                    System.out.println("Production is paused");
                    boxingQLock.wait();
                }
            }
                if (labellingQ.size() == 6) {
                    System.out.println("Packaging cans into box ");
                    Can can1 = labellingQ.take();
                    Can can2 = labellingQ.take();
                    Can can3 = labellingQ.take();
                    Can can4 = labellingQ.take();
                    Can can5 = labellingQ.take();
                    Can can6 = labellingQ.take();
                    System.out.println("Cans have been shrink wrapped");
                    Box bx = new Box(can1, can2, can3, can4, can5, can6, boxNo);
                    packagingQ.put(bx);
                    System.out.println("Cans " + can1.CanNo + " " + can2.CanNo + " " + can3.CanNo + " " + can4.CanNo + " " + can5.CanNo + " " + can6.CanNo + " has been packed into packet " + bx.BoxNo);
                    boxNo++;
                }
            } catch (Exception e) {
            }
        }
    }
}


class Boxing extends Thread {

LinkedBlockingQueue<Box> packagingQ;
LinkedBlockingQueue<Bigbox> boxingQ;

public Boxing(LinkedBlockingQueue<Box> packagingQ, LinkedBlockingQueue<Bigbox> boxingQ) {
    this.packagingQ = packagingQ;
    this.boxingQ = boxingQ;
}

synchronized public void run() {
    int bigboxNo = 1;

    while (true) {
        try {
            LinkedBlockingQueue<Box> packet = new LinkedBlockingQueue<Box>(2);
            if (boxingQ.size() == 1) {
                System.out.println("No space!! Waiting for boxed to be taken!");
                System.out.println("No space!! Waiting for boxed to be taken!");
            }
            for (int i = 1; i < 3; i++) {
                Box b1 = packagingQ.take();
                System.out.println("Loading packets into boxes");
                packet.put(b1);
            }
            Bigbox big = new Bigbox(packet, bigboxNo);
            bigboxNo++;
            System.out.println("Packets  have been loaded into box " + big.BigboxNo);
            boxingQ.put(big);
            System.out.println("BoxingQ size: " + boxingQ.size());
            System.out.println("Box has been stored into loading area");
        } catch (Exception e) {
        }
    }
}
}
class ForkliftTruck extends Thread {

    LinkedBlockingQueue<Bigbox> boxingQ;
    LinkedBlockingQueue<Bigbox> loadingQ;
    int forkno;

    public ForkliftTruck(LinkedBlockingQueue<Bigbox> boxingQ, LinkedBlockingQueue<Bigbox> loadingQ, int forkno) {
        this.boxingQ = boxingQ;
        this.loadingQ = loadingQ;
        this.forkno = forkno;
    }

    public synchronized void run() {
        while (true) {
            try {
                 synchronized (boxingQLock) {
                Bigbox bb = boxingQ.take();
                boxingQLock.notifyAll();
                Thread.sleep(500);
                System.out.println("Production is resumed");
                System.out.println("Forklift " + forkno + " has picked box " + bb.BigboxNo);
                Random rand = new Random();
                int fdefect = rand.nextInt(4) + 1;
                if (fdefect == 2) {
                    Thread.sleep(1000);
                    System.out.println("Forklift " + forkno + " has broken down......waiting for repair");
                    Thread.sleep(5000);
                    System.out.println("Forklift " + forkno + " is repaired......successfully placed box " + bb.BigboxNo + " in the loading bay");
                    loadingQ.put(bb);
                    System.out.println("LoadingQ size: " + loadingQ.size());
                } else {
                    Thread.sleep(1000);
                    System.out.println("Forklift " + forkno + " is in good condition");
                    Thread.sleep(500);
                    System.out.println("Forklift " + forkno + " has successfully placed box " + bb.BigboxNo + " in the loading bay");
                    loadingQ.put(bb);
                    System.out.println("LoadingQ size: " + loadingQ.size());
                }
            }
            } catch (Exception e) {
            }
        }
    }
}

从输出中可以看到，一旦 boxingQ.size 达到 1，它就会暂停，但是一旦从队列中取出并减少到 0，notify()功能似乎不起作用

这是我的输出

Loading packets into boxes
Cans 8 9 11 12 14 16 has been packed into packet 2 //2 packets go into one box
Packets  have been loaded into box 1
BoxingQ size: 1
Box has been stored into loading area
Production is paused
Forklift 1 has picked box 1

如您所见，由于叉车已拾取箱子，boxingQ 目前队列中有 0 个元素。但生产并未恢复

Answer 1

调用notify将释放等待您调用notify的对象的线程。也就是说，如果一个线程调用 obj.wait()，另一个线程必须调用 obj.notify() 来唤醒等待线程。

就您而言，看起来您需要等到 boxingQ 满足某个条件。您可以为此使用对象:

static final Object boxingQLock=new Object();

synchronized (boxingQLock) {
   while ( boxingQ does not satisfy the condition) {
      boxingQLock.wait(); 
   } 
   // when you come here, boxingQ satisfies the condition, and you are
   // holding a lock on it
}

并且，在另一个修改 boxingQ 的线程中:

synchronized (boxingQLock) {
   // Modify boxingQ
   // Notify a thread
   boxingQLock.notify(); //or notifyAll()
}

抛开所有关于等待/通知的讨论:您正在使用BlockingQueue。这意味着，您可以简单地调用 take 并等待有人将某些内容放入队列，而根本不用担心同步/等待/通知。然而，像下面这样的代码本质上是活泼的:

if (labellingQ.size() == 6) {
  System.out.println("Packaging cans into box ");
  Can can1 = labellingQ.take();
  Can can2 = labellingQ.take();
  ...

因为如果没有显式锁定，队列大小可能会在您读取后发生变化。无法保证在读取长度后其他线程不会执行相同操作并尝试从中获取元素。一种方法是取 6 次，如果不够就让队列负责等待。或者，使用 labellingQ 锁，检查是否有足够的项目并在同步块(synchronized block)中获取所有 6 个项目，这样当您知道有足够的项目时，其他线程就无法在您之前获取它们。