在 Java 中实现多消费者多生产者的问题

Question

我写了一个简单的消费者-生产者问题，它有一个阻塞队列，有多个生产者，多个消费者获取和放入队列中的整数。但是，当我尝试对其进行测试时，结果并不理想，例如队列的大小不正确。我不认为消费者和生产者的规模是同步的。此外，我对生产者和消费者都设置了 2 秒的睡眠时间，但在测试时，它每两秒打印出 2 个生产者和 2 个消费者的结果。有谁知道我做错了什么？也许我启动线程错误？我评论了另一种方法，但结果仍然是错误的。

结果：

run:
Producing 425     Thread-0 size left 0
Consuming 890     Thread-3 size left 0
Consuming 425     Thread-2 size left 0
Producing 890     Thread-1 size left 0
Consuming 192     Thread-2 size left 0
Consuming 155     Thread-3 size left 0
Producing 155     Thread-1 size left 0
Producing 192     Thread-0 size left 0
Consuming 141     Thread-2 size left 1
Producing 141     Thread-0 size left 0
Producing 919     Thread-1 size left 0
Consuming 919     Thread-3 size left 0
Producing 361     Thread-1 size left 0
Producing 518     Thread-0 size left 0
Consuming 518     Thread-3 size left 0
Consuming 361     Thread-2 size left 0
Producing 350     Thread-0 size left 1
Consuming 350     Thread-3 size left 0
Consuming 767     Thread-2 size left 0
Producing 767     Thread-1 size left 0

制片人

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

public class Producer implements Runnable {

    BlockingQueue<Integer> items = new LinkedBlockingQueue<>();

    public Producer(BlockingQueue<Integer> q) {
        this.items = q;
    }

    private int generateRandomNumber(int start, int end) {
        Random rand = new Random();
        int number = start + rand.nextInt(end - start + 1);
        return number;
    }

    public void run() {
        for (int i = 0; i < 5; i++) {
            int rand = generateRandomNumber(100, 1000);
            try {
                items.put(rand);
                System.out.println("Producing " + rand + "     " + Thread.currentThread().getName() + " size left " + items.size());
                Thread.sleep(3000);
            } catch (InterruptedException ex) {
                Logger.getLogger(ProducerConsumer.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }
}

消费者

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

public class Consumer implements Runnable {

    BlockingQueue<Integer> items = new LinkedBlockingQueue<>();

    public Consumer(BlockingQueue<Integer> q) {
        this.items = q;
    }

    public void run() {
        while (true) {
            try {
                System.out.println("Consuming " + items.take() + "     " + Thread.currentThread().getName() + " size left " + items.size());
                Thread.sleep(3000);
            } catch (InterruptedException ex) {
                Logger.getLogger(ProducerConsumer.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }
}

测试

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class ProducerConsumer {
    public static void main(String args[]) {
        BlockingQueue<Integer> items = new LinkedBlockingQueue<>();

        Producer producer = new Producer(items);
        Consumer consumer = new Consumer(items);
        Thread t1 = new Thread(producer);
        Thread t2 = new Thread(producer);
        Thread t3 = new Thread(consumer);
        Thread t4 = new Thread(consumer);
        /*
        Thread t1 = new Thread(new Producer());
        Thread t2 = new Thread(new Producer());
        Thread t3 = new Thread(new Consumer());
        Thread t4 = new Thread(new Consumer());
        */
        t1.start();
        t2.start();
        t3.start();
        t4.start();
    }
}

更新：我试图实现可重入锁，但我的程序在锁行处停止。有什么帮助吗？ 消费者

import java.util.concurrent.locks.ReentrantReadWriteLock;

public class Consumer implements Runnable { 

    //private BlockingQueue<Integer> items = new LinkedBlockingQueue<>(); 
    private MyBlockingQ items;

    public Consumer(MyBlockingQ q) { 
        this.items = q; 
    } 

    public void run() { 
        while (true) { 
            items.remove();
            //Thread.sleep(1000);
        }
    }
} 

制片人

import java.util.Random;

public class Producer implements Runnable {
    private MyBlockingQ items;
    public Producer(MyBlockingQ q) {
        this.items = q;
    }

    private int generateRandomNumber(int start, int end) {
        Random rand = new Random();
        int number = start + rand.nextInt(end - start + 1);
        return number;
    }

    public void run() {
        for (int i = 0; i < 5; i++) {
            int rand = generateRandomNumber(100, 1000);
            items.add(rand);
        }
    }
}

MyBlockingQ（共享资源）

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import java.util.logging.Logger;

public class MyBlockingQ {

    private BlockingQueue<Integer> items = new LinkedBlockingQueue<>();
    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

    public MyBlockingQ() {
    }

    public void add(Integer i) {
        lock.writeLock().lock();
        try {
            items.put(i);
            System.out.println("Producing " + i + "     " + Thread.currentThread().getName() + " size left " + items.size());
        } catch (InterruptedException ex) {
            Logger.getLogger(ProducerConsumer.class.getName()).log(Level.SEVERE, null, ex);
        } finally {
            lock.writeLock().unlock();
        }
    }

    public void remove() {
        lock.writeLock().lock();
        try {
            int taken = items.take();
            System.out.println("Consuming " + taken + "     " + Thread.currentThread().getName() + " size left " + items.size());
        } catch (InterruptedException ex) {
            Logger.getLogger(ProducerConsumer.class.getName()).log(Level.SEVERE, null, ex);
        } finally {
            lock.writeLock().unlock();
        }
    }
}

测试

import java.util.concurrent.BlockingQueue; 
import java.util.concurrent.LinkedBlockingQueue; 

public class ProducerConsumer { 
    public static void main(String args[]) { 
        MyBlockingQ items = new MyBlockingQ(); 

        System.out.println("starting");
        Thread t1 = new Thread(new Producer(items)); 
        Thread t2 = new Thread(new Producer(items)); 
        Thread t3 = new Thread(new Consumer(items)); 
        Thread t4 = new Thread(new Consumer(items)); 
        t1.start(); 
        t2.start(); 
        t3.start(); 
        t4.start(); 
    } 
} 

Answer 1

这两行

items.put(rand);
System.out.println("Producing " + rand + "     " + Thread.currentThread().getName() + " size left " + items.size());

不同步。生产者可能会在队列中放入数字，但是当从放入队列的线程显示队列的大小时，消费者可能已经消费了一个数字。

Answer 2

您可能对这部分输出感到困惑：

Producing 425     Thread-0 size left 0
Consuming 890     Thread-3 size left 0
Consuming 425     Thread-2 size left 0
Producing 890     Thread-1 size left 0

问题： Thread-3 为何在 Thread-1 生产它们之前消耗了 890 个项目？

答案： Thread-3 不消耗物品，在它们被 Thread-1 生产之前。

原因： 当 Thread-1 将项目放入队列时，Thread-3 可能已经在等待从队列中取出项目。所以 Thread-1 放了物品：

items.put(rand);

在 Thread-1 跳到下一行并打印有关它生成的项目的信息之前，Thread-3 执行以下行：

System.out.println("Consuming " + items.take() + "     " + Thread.currentThread().getName() + " size left " + items.size());

只有这样 Thread-1 才会执行它的 println：

System.out.println("Producing " + rand + "     " + Thread.currentThread().getName() + " size left " + items.size());

因此，您可以在控制台中看到这些有趣的结果。

您可能想了解synchronizing。有两种方法可以解决您的问题：

• 同步方法
• 同步语句（brimborium 使用的方法）

同步锁定对同步块内对象的访问。这意味着所有其他方法都必须等待轮到它，然后才能访问对象。

因此，如果您在生产者和消费者中的项目上使用同步，那么：

• 生产者放置物品时，消费者无法取走物品。
• 生产者无法在消费者取走物品时放入物品。

如果项目为空并且消费者的方法锁定项目，程序将陷入所谓的死锁。生产者必须等待消费者解锁，但这永远不会发生，因为消费者正在等待获取物品（这些物品必须由生产者放置在那里）。

此外，我让生产者和消费者都睡了 2 秒，但是当 测试，每两秒打印出 2 个生产者的结果 和 2 个消费者。

这正是您应该期待的。在 Test 类中，您将创建 2 个生产者和 2 个消费者。

Thread t1 = new Thread(producer);
Thread t2 = new Thread(producer);
Thread t3 = new Thread(consumer);
Thread t4 = new Thread(consumer);

t1.start();
t2.start();
t3.start();
t4.start();

Answer 3

您需要同步items 访问。我只是稍微改变了你的例子，结果看起来不错。由于同步，您还必须处理死锁。在这种情况下，只要您不通过Consumer 中的items.take() 同步就应该没问题。

你的新测试：

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class ProducerConsumer {
    public static void main(String args[]) {
        BlockingQueue<Integer> items = new LinkedBlockingQueue<>();

        Thread t1 = new Thread(new Producer(items));
        Thread t2 = new Thread(new Producer(items));
        Thread t3 = new Thread(new Consumer(items));
        Thread t4 = new Thread(new Consumer(items));
        t1.start();
        t2.start();
        t3.start();
        t4.start();
    }
}

消费者

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

public class Consumer implements Runnable {

    BlockingQueue<Integer> items = new LinkedBlockingQueue<>();

    public Consumer(BlockingQueue<Integer> q) {
        this.items = q;
    }

    public void run() {
        while (true) {
            try {
                System.out.println("Consuming " + items.take() + "     " + Thread.currentThread().getName() + " size left " + items.size());
                Thread.sleep(1000);
            } catch (InterruptedException ex) {
                Logger.getLogger(ProducerConsumer.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }
}

还有制作人

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

public class Producer implements Runnable {

    BlockingQueue<Integer> items = new LinkedBlockingQueue<>();

    public Producer(BlockingQueue<Integer> q) {
        this.items = q;
    }

    private int generateRandomNumber(int start, int end) {
        Random rand = new Random();
        int number = start + rand.nextInt(end - start + 1);
        return number;
    }

    public void run() {
        for (int i = 0; i < 5; i++) {
            int rand = generateRandomNumber(100, 1000);
            try {
                synchronized (items) {
                    items.put(rand);
                    System.out.println("Producing " + rand + "     " + Thread.currentThread().getName() + " size left " + items.size());
                }
                Thread.sleep(1000);
            } catch (InterruptedException ex) {
                Logger.getLogger(ProducerConsumer.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }
}