通知不同对象中的线程

Question

在下面的生产者和消费者问题中，如何在不使用同步方法的情况下，在下面的程序中将线程从一个对象通知到另一个对象。

我为put 和get 方法使用queue 类，并在Producer 类和Consumer 类的run() 方法中使用wait() 和notify()。

我的目标是在Producer 类和Consumer 类中使用wait() 和notify() 方法，而不是在Queue 类中使用它们。

它给出了一个IllegalMonitorStateException。

程序：

package threads;

class Queue{
    int num;

    int get(int number)
    {

        System.out.println("The Consumer "+number+" got "+num);
        return num;
    }

    void put(int n,int number)
    {

        this.num=n;
        System.out.println("The producer "+number+" put "+this.num);

    }
}

public class producerandconsumer{
    boolean flag=false;
class Producer implements Runnable{
    Queue q;
    int number;
    Producer(Queue q,int number)
    {
        this.q=q;
        this.number = number;
        new Thread(this,"Producer").start();
    }

    public void run()
    {
        for(int i=0;i<10;i++)
        {
            while(flag)
                try{
                    wait();
                }
                catch(InterruptedException e){
                    System.out.println("InterruptedException caught ");
                }
            q.put(i,number);
            flag=true;
            notify();
        }
    }

}

class Consumer implements Runnable{
    Queue q;
    int number;
    Consumer(Queue q,int number)
    {
        this.q=q;
        this.number=number;
        new Thread(this,"Consumer").start();
    }

    public void run()
    {
        for(int i=0;i<10;i++)
        {
            while(!flag)
                try{
                    wait();
                    }
                    catch(InterruptedException e){
                        System.out.println("InterruptedException caught ");
                    }
                flag=false;
                notify();
                q.get(number);
        }
    }
}



public static void main(String[] args) {
    // TODO Auto-generated method stub
    producerandconsumer pc= new producerandconsumer();
    Queue q=new Queue();
    pc.new Producer(q,1);
    pc.new Consumer(q,1);


}

}

程序的输出：它给出了一个IllegalMonitorStateException。

 The producer 1 put 0 Exception in thread "Producer" 
 java.lang.IllegalMonitorStateException
at java.lang.Object.notifyAll(Native Method)
at threads.producerandconsumer$Producer.run(producerandconsumer.java:48)
at java.lang.Thread.run(Unknown Source)

Exception in thread "Consumer" java.lang.IllegalMonitorStateException
at java.lang.Object.notifyAll(Native Method)
at threads.producerandconsumer$Consumer.run(producerandconsumer.java:76)
at java.lang.Thread.run(Unknown Source)

Answer 1

方法wait 和notify 必须在synchronized 块内调用。这就是你得到 illegelMonitorStateException 的原因。

Answer 2

线程之间共享的状态是您的Queue 类，因此您必须通过使用适当的同步使其成为线程安全的。将同步代码放在 Producer 和 Consumer 中并不是一个好主意，因为这需要它们之间的额外通信并且无法扩展。

下面是一个同步整数队列的简单示例，它只能容纳一个int（类似于您的Queue）。当take 在空队列上被调用以及put 在满队列上被调用时，它会阻塞。

请注意，您应该将现有的同步数据结构（例如 BlockingQueue 的实现）用于实际应用程序！

public class IntQueue {
    int data;
    boolean filled = false;

    public synchronized int take() throws InterruptedException
    {
        while (!filled) { // wait for filled condition
            wait();
        }

        filled = false; // set not-filled condition
        notifyAll(); // notify (other) waiting threads

        return data;
    }

    public synchronized void put(int data) throws InterruptedException
    {
        while (filled) { // wait for not-filled condition
            wait();
        }

        filled = true; // set filled condition
        notifyAll(); // notify (other) waiting threads

        this.data = data;
    }
}

使用此队列，您可以拥有任意数量的不需要任何进一步同步的生产者和消费者：

static final IntQueue queue = new IntQueue();
static final int POISON_PILL = -1; // stops Consumer

class Producer implements Runnable {
    public void run() {
        try {
            for(int i = 0; i < 100; i++) {
                System.out.println("producing " + i);
                queue.put(i);
            }
        } catch (InterruptedException ex) { /* done */ }
    }
}

class Consumer implements Runnable {
    public void run() {
        try {
            int n = queue.take();
            // poison pill causes Consumer to stop
            while (n != POISON_PILL) {
                System.out.println("consuming " + i);
                n = queue.take();
            }
        } catch (InterruptedException ex) { /* done */ }
    }
}

public static void main() throws Exception {
    // create threads
    Thread p1 = new Thread(new Producer());
    Thread p2 = new Thread(new Producer());
    Thread c = new Thread(new Consumer());

    // start threads
    p1.start();
    p2.start();
    c.start();

    // wait for producers to complete
    p1.join();
    p2.join();

    // queue poison pill to stop consumer
    queue.put(POISON_PILL);

    // wait for consumer to complete
    c.join();
}