java - 生产者线程和消费者线程之间的线程间通信？

Question

我正在尝试学习使用 BlockingQueue 的线程间通信。

我编写了一个生成器，它生成 TaskId 并将其插入到 BlockingQueue 中。

现在我有 2 个消费者线程(名称“1”和“0”)。如果taskId是奇数，则它被线程“1”消耗，否则“2”消耗。

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

                while (queue.peek() != null && !name.equals(String.valueOf(queue.peek().intValue() % 2 ))) {

                try {
                    System.out.println(name + ",consumed," + queue.take());
                } catch (InterruptedException e) {
                    e.printStackTrace();

                }
            }
        }
    }

我怎样才能在这里进行检查？

Answer 1

我想的一种方法，也可能有其他更好的方法:

@Override
    public void run() {
        String name = Thread.currentThread().getName();
        while (true) {

            while (queue.peek() == null) {
                //some sleep time
            }

            synchronized (lock) {
                while (queue.peek() != null && !name.equals(String.valueOf(queue.peek().intValue() % 2 ))) {
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                if(queue.peek() != null) {
                    try {
                        System.out.println(name + ",consumed," + queue.take());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                lock.notify();
            }
        }
    }

另一种方式:每当元素添加到队列时，生产者线程都会通知 anotherLock。

@Override
    public void run() {
        String name = Thread.currentThread().getName();
        while (true) {

            synchronized (anotherLock) {
                while (queue.peek() == null) {
                    anotherLock.wait();
                }
            }

            synchronized (lock) {
                while (queue.peek() != null && !name.equals(String.valueOf(queue.peek().intValue() % 2 ))) {
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                if(queue.peek() != null) {
                    try {
                        System.out.println(name + ",consumed," + queue.take());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                lock.notify();
            }
        }
    }