尝试用多线程解决java中的消费者-生产者

Question

我正在尝试使用 java 中的线程解决生产者消费者问题，但代码不会并行/并发运行。生产者总是在消费者开始消费之前完全填满缓冲区，我不明白为什么。重点是尝试仅使用同步块、wait() 和 notify() 来完成。

主要：

    String [] data = {"Fisk", "Katt", "Hund", "Sau", "Fugl", "Elg", "Tiger", 
               "Kameleon", "Isbjørn", "Puma"};
    ProducerConsumer pc = new ProducerConsumer(5);
    Thread[] thrds = new Thread[2];
    thrds[0] = new Thread(new MyThread1(pc, data)); // producer
    thrds[1] = new Thread(new MyThread2(pc)); // consumer
    thrds[0].start();
    thrds[1].start();
    for(int i = 0; i < 2; i++) { // wait for all threads to die
        try { 
            thrds[i].join(); 
        } 
        catch (InterruptedException ie) {}
    }
    System.exit(0);

ProducerConsumer.java：

import java.util.LinkedList;
import java.util.Queue;

public class ProducerConsumer implements Runnable {
    private int bufferSize;
    private Queue<String> buffer;

public ProducerConsumer(int size) {

    bufferSize = size;
    buffer = new LinkedList<String>();
}

public void produce(String item) throws InterruptedException {
        synchronized(buffer) {
            while (buffer.size() >= bufferSize) {
                try {
                    System.out.println("Full buffer. Waiting for consumer...");
                    buffer.wait();
                }catch (Exception e) {}
            }
            buffer.add(item);
            System.out.println("Producer is putting " + item + " in the buffer");
            buffer.notify();
        }   
}

public void consume() throws InterruptedException {
    synchronized (buffer) {
        while (buffer.size() == 0) {
            try {
                System.out.println("Empty buffer. Waiting for production...");
                buffer.wait();
            }catch (Exception e) {}
        }
        System.out.println("Consumer is consuming " +  buffer.remove() + ".");
        buffer.notify();
    }
}

@Override
public void run() {
}

}

MyThread1：

/*
 * PRODUCER - Thread
 */
public class MyThread1 implements Runnable {

private String [] data;
private ProducerConsumer pc;

public MyThread1(ProducerConsumer pc, String [] data) {
    this.pc = pc;
    this.data = data;
}
@Override
public void run() {
    for (int i = 0; i < data.length; i++) {
        try {
            pc.produce(data[i]);
        } catch (InterruptedException ex) {}
    }
}

}

MyThread2：

//消费者 - 线程

public class MyThread2 implements Runnable{

private ProducerConsumer pc;

public MyThread2(ProducerConsumer pc) {
    this.pc = pc;
}

//Run consume
@Override
public void run() {
    while (true) {
        try {
            pc.consume();
            Thread.sleep(2);
        }
        catch(InterruptedException e) {}

    }

}
}

Answer 1

在最近的机器上，像这样的短队列，你永远不会看到真正的多线程效果，比如在这种情况下，生产者和消费者轮流进行，除非你稍微放慢它们的速度。你只会减慢消费者的速度。不要使用短数组，而是将一百万个整数放入队列中，看看会发生什么。

Answer 2

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

public class ProduserConsumerDemo {

    public static void main(String[] args) {
        List<Integer> list = new CopyOnWriteArrayList<>();
        int size = 5;
        Producer producer = new Producer(list, size);
        Consumer consumer = new Consumer(list);
        Thread t1 = new Thread(producer, "Producer");
        Thread t2 = new Thread(consumer, "Consumer");
        t1.start();
        t2.start();
    }
}

class Producer implements Runnable {
    private final List<Integer> list;
    private final int size;

    public Producer(List<Integer> list, final int size) {
        this.list = list;
        this.size = size;
    }

    public void run() {
        try {
            produce();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void produce() throws InterruptedException {
        int i = 0;
        while (i >= 0) {
            synchronized (list) {
                while (list.size() == size) {
                    System.out.println(
                            "List is full." + Thread.currentThread().getName() + " is waiting. Size:" + list.size());
                    list.wait();
                }
                System.out.println("Produce :" + i);
                list.add(i++);
                Thread.sleep(50);
                list.notify();
            }
        }
    }
}

class Consumer implements Runnable {
    private final List<Integer> list;

    public Consumer(List<Integer> list) {
        this.list = list;
    }

    public void run() {
        try {
            consume();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void consume() throws InterruptedException {
        while (true) {
            synchronized (list) {
                while (list.isEmpty()) {
                    System.out.println(
                            "List is empty. " + Thread.currentThread().getName() + " is waiting. Size:" + list.size());
                    list.wait();
                }
                System.out.println("Consumed item:" + list.remove(0));
                Thread.sleep(50);
                list.notify();
            }
        }
    }
}