消费者线程停止的问题

Question

这个是队列目前存放个标记位对象，得到这个标记后，消费者线程停止；
private volatile BlockingQueue<KaiFaBan> queue=new LinkedBlockingQueue<>();
KaiFaBan k=new KaiFaBan();

                    k.setName("开发板1");
                    k.setCount(-1);
                    //"毒丸"标记
                    queue.put(k);
                    

@RequestMapping("/consumer")

public void consumer(HttpServletRequest request, HttpServletResponse response) {
    Runnable consumer = new Runnable() {
        @Override
        public void run() {
            System.out.println("开启《消费者》...");
            while(true){
                try {
                    KaiFaBan takeValue = queue.take();//阻塞
                    //这是“毒丸”对象
                    synchronized (waitLock){
                        if(takeValue.getCount() == -1){
                            //把biaoJi变为true
                            biaoJi=true;
                            System.out.println("《消费者已经停下来了,,,,,,,,,,,,,,,,,,,,,,,,,,,,》");
                            //消费者线程等待(让消费者线程停下来)
                            waitLock.wait();
                        }
                    }
                    System.out.println("《消费者被重新启动,,,,,,,,,,,,,,,,,,,,,,,,,,,,》");
                    synchronized (myLock) {
                        if (takeValue != null) {
                            if (takeValue.getName().equals("开发板1")) {
                                if (KAFABANNUMBER1 < takeValue.getCount()) {
                                    KAFABANNUMBER1 = takeValue.getCount();
                                    System.out.println("KAFABANNUMBER1：" + KAFABANNUMBER1);
                                }else{
                                    System.out.println("else方法中，KAFABANNUMBER1："+ KAFABANNUMBER1);
                                }
                            } else if (takeValue.getName().equals("开发板2")) {
                                if (KAFABANNUMBER2 < takeValue.getCount()) {
                                    KAFABANNUMBER2 = takeValue.getCount();
                                    System.out.println("KAFABANNUMBER2：" + KAFABANNUMBER2);
                                }else{
                                    System.out.println("else方法中，KAFABANNUMBER2："+ KAFABANNUMBER2);
                                }
                            } else if (takeValue.getName().equals("开发板3")) {
                                if (KAFABANNUMBER3 < takeValue.getCount()) {
                                    KAFABANNUMBER3 = takeValue.getCount();
                                    System.out.println("KAFABANNUMBER3：" + KAFABANNUMBER3);
                                }else{
                                    System.out.println("else方法中，KAFABANNUMBER3："+ KAFABANNUMBER3);
                                }
                            }
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    };
    exec.execute(consumer);
}


下边摘录是疑惑的地方，这个是咋么让所有的消费者停下来的？不知道之是让一个线程进入等待吗？
//这是“毒丸”对象
                synchronized (waitLock){
                    if(takeValue.getCount() == -1){
                        //把biaoJi变为true
                        biaoJi=true;
                        System.out.println("《消费者已经停下来了,,,,,,,,,,,,,,,,,,,,,,,,,,,,》");
                        //消费者线程等待(让消费者线程停下来)
                        waitLock.wait();
                    }
                }

这个应该只能让一个线程停下来吧。我的理解是，takeValue.getCount() == -1是其中的一个元素，第一个线程拿到这个队列中的对象后进入了等待状态，下一个线程再来那就取不到takeValue.getCount() == -1，状态无法满足，之后的线程都无法进入等待，所以为了让所有的线程都停下来，我是这么改动的，不知道对不（就增加了biaoJi == true）?????????????

synchronized (waitLock){

                       if(takeValue.getCount() == -1 || biaoJi == true){
                            //把biaoJi变为true
                            biaoJi=true;
                            System.out.println("《消费者已经停下来了,,,,,,,,,,,,,,,,,,,,,,,,,,,,》");
                            //消费者线程等待(让消费者线程停下来)
                            waitLock.wait();
                        }
                    }

Answer 1

你这种情况最好还是别用毒丸对象，毒丸对象一般放在最后一个，一般能够确定生产和消费的数量。一般使用毒丸对象，是不需要消费者同步的，因为毒丸对象就是最后一个。
你的理解是没有问题的，但感觉你的测试有问题。我不相信还特地去测试了下。

public class PoisonTest {

    private static String poison = "poison";

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> blockingQueue = new LinkedBlockingDeque<>();

        Thread producer = new Thread(() -> {
            for (int i = 0; i < 100; i++) {
                if (i == 50) {
                    try {
                        blockingQueue.put(poison);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    blockingQueue.add(String.valueOf(i));
                }
            }
        });
        Runnable runnable = ()->{

            while (true) {
                try {
                    String s = blockingQueue.take();
                    synchronized (poison) {
                        if (s.equals(poison)) {
                            poison.wait();
                        }
                    }
                    System.out.println(Thread.currentThread().getName() + ":" + s);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };

        Thread consumer1 = new Thread(runnable);

        Thread consumer2 = new Thread(runnable);

        producer.start();

        consumer1.start();

        consumer2.start();

        TimeUnit.SECONDS.sleep(10);
    }

}

从打印结果就能看出，只有一个线程被停止了。

还有，你这消费者基本全部同步了，多线程的意义不太大。

Answer 2

简单看了下
queue队列里面你就放了一个KaiFaBan?
当之后的请求到来的时候,KaiFaBan takeValue = queue.take();//这里都会等着吧?
还是代码没贴全
假设你是在某个时机放入的KaiFaBan Count==1这个对象
注意这一段代码

//尝试获取锁,当Count!=1的时候,代码块迅速结束
//当有一个count==-1时,代码被阻塞,等待notify方法
//那么这个代码块不会完成,下一个请求到来的时候都会阻塞在尝试获取锁
//所以有没有biaoJi == true这个判断都会形成你所谓的了暂停
//你打断点会发现,当第一个count==-1出现之后,接下来的请求待会在这里便阻塞
//不会到达判断count是否等于-1
synchronized (waitLock){ 

                        if(takeValue.getCount() == -1){
                            //把biaoJi变为true
                            biaoJi=true;
                            System.out.println("《消费者已经停下来了,,,,,,,,,,,,,,,,,,,,,,,,,,,,》");
                            //消费者线程等待(让消费者线程停下来)
                            //当第获取到==-1的bean之后这里wait,意味着代码会等在这里
                            waitLock.wait(); 
                        }
                    }