生产者消费者应用程序上的 Java 多线程查询

Question

我写了一个简单的java多线程程序。

我对代码有一些疑问。请帮我解决这些问题。

提前致谢！

这是我的代码：

Producer.java

package com.prodcon;
import java.util.Stack;
public class Producer extends Thread {
    DataStorage data;
    MainProcess tempmp;
    public Producer(DataStorage dst, MainProcess mp){
        data = dst;
        tempmp = mp;
    }
    public void run(){
        for(int i = 0; i < 3; i++){
            System.out.println("Thread:"+this.getName()+"called");
            data.PutData();
            /*-------------current states---------------------*/
            System.out.println("Current states of the threads:");
            System.out.println("p1->"+tempmp.p1.getState());
            System.out.println("p2->"+tempmp.p2.getState());
            System.out.println("p3->"+tempmp.p3.getState());
            System.out.println("c1->"+tempmp.c1.getState());
            System.out.println("c2->"+tempmp.c2.getState());
            System.out.println("c3->"+tempmp.c3.getState());
            /*-------------current states---------------------*/

        }
    }
}

consumer.java

  package com.prodcon;

    public class Consumer extends Thread {
        DataStorage data;
        MainProcess tempmp;
        public Consumer(DataStorage dst, MainProcess mp){
            data = dst;
            tempmp = mp;
        }
        public void run(){
            for(int i = 0; i < 3; i++){
                System.out.println("Thread:"+this.getName()+"called");
                data.GetData();
                /*-------------current states---------------------*/
                System.out.println("Current states of the threads:");
                System.out.println("p1->"+tempmp.p1.getState());
                System.out.println("p2->"+tempmp.p2.getState());
                System.out.println("p3->"+tempmp.p3.getState());
                System.out.println("c1->"+tempmp.c1.getState());
                System.out.println("c2->"+tempmp.c2.getState());
                System.out.println("c3->"+tempmp.c3.getState());
                /*-------------current states---------------------*/
            }
        }
    }

DataStorage.java

    package com.prodcon;

import java.util.Random;
import java.util.Stack;

import javax.xml.crypto.Data;

public class DataStorage {

    int countofdata;
    Stack<Double> data;

    public DataStorage() {
        countofdata = 0;
        data = new Stack<Double>();
    }

    public synchronized void GetData() {
        while (data.isEmpty()) {
            try {
                wait();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
            }

        }
        double temp = (double) data.pop();
        //System.out.println("Data poped out:" + temp);
        countofdata++;
        notifyAll();
    }

    public synchronized void PutData() {
        while (true) {
            if (data.size() == 3) {
                try {
                    wait();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            } else {
                break;
            }
        }
        double temp = Math.random();
        data.push(temp);
        //System.out.println("Data inserted in storage:" + temp);
        countofdata--;
        notifyAll();
    }
}

MainProcess.java

    package com.prodcon;

public class MainProcess {

    /**
     * @param args
     */
    DataStorage ProcessData;
    public Producer p1, p2, p3, p4;
    public Consumer c1, c2, c3, c4;
    public MainProcess(){
        ProcessData = new DataStorage();
        p1 = new Producer(ProcessData, this);
        p2 = new Producer(ProcessData, this);
        p3 = new Producer(ProcessData, this);
        c1 = new Consumer(ProcessData, this);
        c2 = new Consumer(ProcessData, this);
        c3 = new Consumer(ProcessData, this);
        p1.setName("p1");
        p2.setName("p2");
        p3.setName("p3");
        c1.setName("c1");
        c2.setName("c2");
        c3.setName("c3");
    }
    public void startprocess(){
        p1.start();
        p2.start();
        p3.start();
        c1.start();
        c2.start();
        c3.start();
    }
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        MainProcess mp1 = new MainProcess();
        mp1.startprocess();

    }

}

这是这个程序的输出：

    Thread:p2called
Thread:p3called
Current states of the threads:
Thread:p1called
Current states of the threads:
Thread:c3called
Current states of the threads:
p1->RUNNABLE
p2->BLOCKED
p3->RUNNABLE
c1->BLOCKED
c2->BLOCKED
c3->BLOCKED
Thread:p3called
Current states of the threads:
p1->RUNNABLE
p2->BLOCKED
p3->RUNNABLE
c1->BLOCKED
c2->BLOCKED
c3->BLOCKED
Thread:p3called
Thread:c2called
Thread:c1called
Current states of the threads:
Current states of the threads:
p1->RUNNABLE
Current states of the threads:
p2->BLOCKED
p1->RUNNABLE
p2->BLOCKED
p3->BLOCKED
c1->BLOCKED
c2->BLOCKED
c3->BLOCKED
Thread:p1called
Current states of the threads:
p1->RUNNABLE
p2->BLOCKED
p3->BLOCKED
c1->BLOCKED
c2->BLOCKED
c3->BLOCKED
Thread:p1called
Current states of the threads:
p1->RUNNABLE
p2->BLOCKED
p3->BLOCKED
c1->RUNNABLE
c2->RUNNABLE
c3->BLOCKED
Thread:c1called
Current states of the threads:
p1->BLOCKED
p2->BLOCKED
p3->BLOCKED
c1->RUNNABLE
c2->RUNNABLE
c3->BLOCKED
Thread:c1called
Current states of the threads:
p1->BLOCKED
p2->BLOCKED
p3->BLOCKED
c1->RUNNABLE
c2->RUNNABLE
c3->BLOCKED
Current states of the threads:
p1->RUNNABLE
p2->BLOCKED
p3->BLOCKED
c1->TERMINATED
c2->RUNNABLE
c3->BLOCKED
p1->RUNNABLE
p2->BLOCKED
p3->BLOCKED
c1->TERMINATED
c2->BLOCKED
c3->RUNNABLE
Thread:c3called
Current states of the threads:
p1->TERMINATED
p2->BLOCKED
p3->BLOCKED
c1->TERMINATED
c2->BLOCKED
c3->RUNNABLE
Thread:c3called
Current states of the threads:
p1->TERMINATED
p2->BLOCKED
p3->BLOCKED
c1->TERMINATED
c2->BLOCKED
c3->RUNNABLE
p3->RUNNABLE
p1->BLOCKED
p2->RUNNABLE
p3->RUNNABLE
c1->TERMINATED
c2->BLOCKED
c3->TERMINATED
Thread:p2called
Current states of the threads:
p1->TERMINATED
p2->RUNNABLE
p3->RUNNABLE
c1->TERMINATED
c2->BLOCKED
c3->TERMINATED
Thread:p2called
Current states of the threads:
p1->TERMINATED
p2->RUNNABLE
p3->RUNNABLE
c1->TERMINATED
c2->BLOCKED
c3->TERMINATED
p1->TERMINATED
p2->TERMINATED
p3->RUNNABLE
c1->TERMINATED
c2->RUNNABLE
c3->TERMINATED
Thread:c2called
Current states of the threads:
p1->TERMINATED
p2->TERMINATED
p3->RUNNABLE
c1->TERMINATED
c2->RUNNABLE
c3->TERMINATED
Thread:c2called
Current states of the threads:
p1->TERMINATED
p2->TERMINATED
p3->RUNNABLE
c1->TERMINATED
c2->RUNNABLE
c3->TERMINATED
c1->TERMINATED
c2->TERMINATED
c3->TERMINATED

我的问题是：

1.根据程序，这个进程应该永远不会停止......但是一些线程会自动终止，为什么？？

2.即使在要求某些线程进入等待状态之后..根据输出，没有线程进入等待状态。为什么以及如何？

3.根据这段代码：生产者-消费者问题和 读写器问题？

再次感谢！！

Answer 1

第一个和第二个问题 - 由于逻辑原因，此代码自然退出。尝试在此处添加调试输出以查看生产者何时退出。

if (data.size() == 3) {
    try {
        wait();
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
} else {
    System.out.printf("%s: breaking as data size is %d, %d%n", Thread.currentThread().getName(), countofdata, data.size());

    break;
}

另一个可能的问题是：

for(int i = 0; i < 3; i++){
    data.PutData();
}

你只放入了 3 次队列。

提示：Java 中生产者-消费者问题的经典实现是LinkedBlockingQueue。

第三个问题。我将尝试向生产者和消费者解释这一点。读写器问题是当您没有生产者但有许多消费者访问单个共享资源时。其中一些可以读取资源，一些可以写入资源。可以同时读取，并且写入锁是独占的。

更新

要模拟多次读取，您可以尝试使用

• LinkedBlockingQueue 与您的 DataStorage 相当。
• ReentrantReadWriteLock
• 或更新您的解决方案。这是研究示例，我没有测试，但应该给你一个大致的想法：

最终对象 readLock = new Object(); //使用对象作为锁
final Object writeLock = new Object();

public void GetData() {
    synchronized (readLock) {             // acquire only the read lock
        while (data.isEmpty()) {
            try {
                wait();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
            }

        }
        double temp = (double) data.pop();
        //System.out.println("Data poped out:" + temp);
        countofdata++;
        System.out.printf("%s: %d, %d%n", Thread.currentThread().getName(), countofdata, data.size());
        notifyAll();        
     }
 }

public void PutData() {
    synchronized (readLock) {       //first, acquire the read lock
        synchronized (writeLock) {  // then acquire the write lock
            while (true) {
                if (data.size() == 3) {
                    try {
                        wait();
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                } else {
                    System.out.printf("%s: breaking as data size is %d, %d%n", Thread.currentThread().getName(), countofdata, data.size());

                    break;
                }
            }
            double temp = Math.random();
            data.push(temp);
            //System.out.println("Data inserted in storage:" + temp);
            countofdata--;
            notifyAll();
        }
    }
}

Answer 2

我可以看到您非常热衷于使用线程，这是问题的主题，但您是否考虑过 Observer/Observable？

数据存储是您的 Observable。

消费者是观察者。他们正在寻找改变。

我修改了您的代码，只有一个定期运行的线程。

制作人：

 public class Producer{
    DataStorage data;
    MainProcess tempmp;

    public Producer(DataStorage dst, MainProcess mp){
        data = dst;
        tempmp = mp;
    }
    public void changeTheData(){

            data.PutData();
            System.out.println("INFO :: Producer :: Just put data");
    }
}

消费者：

 import java.util.Observable;
 import java.util.Observer;
 import java.util.Stack;
 import java.util.concurrent.atomic.AtomicReference;

 public class Consumer implements Observer {
    DataStorage data;
    MainProcess tempmp;
    public Consumer(DataStorage dst, MainProcess mp){
        data = dst;
        tempmp = mp;
    }


 public void update(Observable arg0, Object arg1) {
    System.out.println("INFO :: class is " + arg1.getClass().toString());
    System.out.println("INFO :: Consumer :: data is " + ((AtomicReference<Double>)arg1).get());


}
}

数据存储：

 import java.util.Observable;
 import java.util.concurrent.atomic.AtomicReference;

 public class DataStorage extends Observable {

   int countofdata;
   AtomicReference<Double> data = new AtomicReference<Double>();

   public DataStorage() {
       countofdata = 0;

   }

   public AtomicReference<Double> GetData() {

       // System.out.println("Data poped out:" + temp);
       countofdata--;
       return data;

   }

   public synchronized void PutData() {

      this.data.set(Math.random());
      System.out.println("INFO :: DataStorage :: Data inserted in storage:" + this.data.get());
      this.setChanged();
      this.notifyObservers(data);
      countofdata++;

  }
 }

主进程：

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class MainProcess implements Runnable{

/**
 * @param args
 */
DataStorage processData;
private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
private Producer p1, p2, p3, p4;
private Consumer c1, c2, c3, c4;

public MainProcess(){

    processData = new DataStorage();
    p1 = new Producer(processData, this);
    p2 = new Producer(processData, this);
    p3 = new Producer(processData, this);
    c1 = new Consumer(processData, this);
    c2 = new Consumer(processData, this);
    c3 = new Consumer(processData, this);
    processData.addObserver(c1);
    processData.addObserver(c2);
    processData.addObserver(c3);


}
public void startprocess(){
     this.scheduledExecutorService.scheduleAtFixedRate(this, 5, 10, TimeUnit.SECONDS);

}
public static void main(String[] args) {
    // TODO Auto-generated method stub
    MainProcess mp1 = new MainProcess();
    mp1.startprocess();

}
public void run() {
    System.out.println("INFO :: p1 Changing the data");
    p1.changeTheData();
    System.out.println("INFO :: p2 Changed the data");
    p2.changeTheData();
    System.out.println("INFO :: p3 Changed the data");
    p3.changeTheData();
}

}