Java 多线程：慢的生产者和快的消费者

Question

我正在尝试理解同步块。这里我实现了一个生产者线程和两个消费者线程。

由于 LinkedList 为空，我在线程中不断收到异常。

package com.Main;

import com.qed.Consumer;
import com.qed.Producer;
import com.qed.Store;

public class Main {

public static void main(String[] args) throws InterruptedException {
    Store st = new Store();

    Thread populate = new Thread(new Producer(st)); 
    Thread con1 = new Thread(new Consumer(st));
    Thread con2 = new Thread(new Consumer(st));
    con1.setName("A");
    con2.setName("B");
    populate.start();
    con1.start();
    con2.start();

    populate.join();
    con1.join();
    con2.join();
    if(populate.isAlive()){
        con1.interrupt();
        con2.interrupt();
    }
  }
}

package com.qed;

import java.util.LinkedList;

public class Store {

private LinkedList<Integer> qu = new LinkedList<Integer>();
private final Object lock = new Object();

public void add(int data){
    try{
        while(qu.size() ==10){
            Thread.sleep(1);
            }
        qu.add(data);
    }catch(InterruptedException ie){
        ie.printStackTrace();
    }
}

public int remove(){
    int data=0;
    try{
        synchronized(lock){
            while(qu.size() == 0){
                Thread.sleep(1);
                }
            data = qu.removeFirst();
         }
    }catch(InterruptedException ie){
        ie.printStackTrace();
    }
    return data;
  }
}

package com.qed;

public class Consumer implements Runnable{

private Store st;
public Consumer(Store st){
    this.st=st;
}


public void run(){
    while(true){
        System.out.println(Thread.currentThread().getName() + ". " +st.remove());
    }
  }
}

package com.qed;

public class Producer implements Runnable{

private Store st;
private final int runs = 5000;
public Producer(Store st){
    this.st = st;
}

public void run(){
    int data = 0;
    int curRun =0;
    while(++curRun < runs){
        st.add(data+=200);
        }
    System.out.println("DONE.");
  }
}

堆栈跟踪：

Exception in thread "B" Exception in thread "A"  
java.util.NoSuchElementException    
    at java.util.LinkedList.removeFirst(Unknown Source)    
    at com.qed.Store.remove(Store.java:46)    
    at com.qed.Consumer.run(Consumer.java:20)    
    at java.lang.Thread.run(Unknown Source)    
java.util.NoSuchElementException    
    at java.util.LinkedList.removeFirst(Unknown Source)    
    at com.qed.Store.remove(Store.java:46)    
    at com.qed.Consumer.run(Consumer.java:20)    
    at java.lang.Thread.run(Unknown Source)    

Answer 1

您还必须锁定添加。您的代码允许生产者更新队列，而消费者可能想要删除条目！

当两个线程并行修改同一个队列时，所有的赌注都没有了！

单个锁的使用只会防止多个消费者相互踩踏！

因此：为添加值的部分添加相同类型的锁定。

除此之外，EJP 是正确的——真正的解决方案是使用低级别的信令方法，例如 wait() 和 notify()。但是，当然，使用这些会导致非常不同的行为。

鉴于您的评论：请记住，这是两件不同的事情：A) 消费者/生产者相互发送信号 B) 消费者/生产者以相同的外观同步。

我知道您不想要“A)” - 但您需要“B)”；否则您的队列会损坏，并且会发生意外。

Answer 2

你应该在这里调用 wait() 方法。
wait() 让你的线程等到其他线程调用 notify 来唤醒他。
sleep() 只是在指定的时间段内不执行下一条语句。
如果你看到你的程序 sn-p，你正在使用同步块，它使用一个对象来检查监视器的可用性。但是您没有使用任何对象监视器方法 wait/notify/notifyAll 并且您试图在不调用这些方法的情况下获取和释放锁。由于消费者和生产者都使用了列表对象，因此您应该使用此列表对象监视器来同步所有线程。如果一个线程获得了它的监视器，那么其他线程将无法访问它。因为每个对象只有一个监视器。这种方法将解决所有工作线程之间的同步问题。

Answer 3

问题是您的Store 类实现。您需要在添加和删除元素时在那里实现wait() 和notify 机制，而不是休眠。

您在所有消费者和生产者之间共享单个 Store 实例是正确的，但您的商店需要表现得像 BlockingQueue

因此，要么使用 JDK 中现有的 BlockingQueue 实现，要么修改 Store 类以实现类似的机制。

implement-your-own blocking queue in java

希望对你有帮助！！