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
}

И в другом потоке, который изменяет боксQ:

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 раз и позволить очереди позаботиться об ожидании, если этого недостаточно. Или используйте блокировку для обозначения Q, проверьте, достаточно ли элементов и соберите все 6 в синхронизированном блоке, чтобы, когда вы знали, что элементов было достаточно, никакой другой поток не мог получить их до вас.