消费者、生产者-互斥体、同步-临界区答案

【问题标题】：Consumer, producer- mutex, sync- critical section消费者、生产者-互斥体、同步-临界区
【发布时间】：2016-01-19 08:57:21
【问题描述】：

我尝试了一些典型的多线程示例，之后我想尝试典型的生产者-消费者问题。

（如果有空间，生产者可以生产，如果消费者不消费，反之亦然）

但是我对共享资源有疑问，在 java 中是否类似于 C 中的信号量？（与等待和发布功能一起使用）

我找到了同步示例
但我想尝试一些我可以手动控制的东西（例如 C 中的信号量）

我有：

class MyThread implements Runnable- 我的线程的基本类
class Producer extends MyThread- 生产者线程
class Consumer extends MyThread- 消费类
class ThreadContainer- 共享资源（库存）

在ThreadContainer 中，我准备了一些锁，我找到并尝试了，但它并没有按预期工作：

java.lang.IllegalMonitorStateException
    at java.lang.Object.notify(Native Method)Running Consumer 0 [1]
java.lang.IllegalMonitorStateException
    at java.lang.Object.wait(Native Method)
    at java.lang.Object.wait(Unknown Source)
(etc.)

如果有人向我解释“如何”，我将不胜感激。

下面的类

我的线程：

public class MyThread implements Runnable {
    private Thread t;
    private String threadName;

    private ThreadContainer container;

    MyThread(String name, ThreadContainer cont) {
        threadName = name;
        this.container = cont;
        System.out.println("Creating " + threadName);
    }

    public void run() {

    }

    public void start() {
        System.out.println("Starting " + threadName);
        if (t == null) {
            t = new Thread(this, threadName);
            t.start();
        }
    }

    public ThreadContainer getContainer() {
        return container;
    }

    public String getThreadName() {
        return threadName;
    }
}

制作人：

 public class Producer extends MyThread {

    Producer(String name, ThreadContainer cont) {
        super(name, cont);
    }

    public void produce(int amount) {
        super.getContainer().produce(amount);
    }

    @Override
    public void run() {
        System.out.println("Running " + super.getThreadName());
        try {
            for (int i = 10; i > 0; i--) {

                synchronized (super.getContainer().lock) {
                    System.out.println(super.getThreadName()
                            + " want to produce: " + i);
                    while (!super.getContainer().canProduce(i)) {
                        super.getContainer().lock.wait();
                    }
                    System.out.println(super.getThreadName() + " producing: "
                            + i);
                    super.getContainer().produce(i);
                    System.out.println("Container state: "
                            + super.getContainer());
                }

            }

            Thread.sleep(50);
        } catch (InterruptedException e) {
            System.out.println("Thread " + super.getThreadName()
                    + " interrupted.");
        }

        System.out.println("Thread " + super.getThreadName() + " exiting.");
    }

}

消费者：

 public class Consumer extends MyThread {

    Consumer(String name, ThreadContainer cont) {
        super(name, cont);
    }

    public void consume(int am) {
        super.getContainer().consume(am);
    }

    @Override
    public void run() {
        System.out.println("Running " + super.getThreadName());
        try {
            for (int i = 10; i > 0; i--) {
                synchronized (super.getContainer().lock) {
                    System.out.println(super.getThreadName()
                            + " want to consume: " + i);
                    while (!super.getContainer().canConsume(i)) {
                        super.getContainer().lock.wait();
                    }
                    System.out.println(super.getThreadName() + " consuming: "
                            + i);
                    super.getContainer().consume(i);
                    System.out.println("Container state: "
                            + super.getContainer());
                }
            }

            Thread.sleep(50);
        } catch (InterruptedException e) {
            System.out.println("Thread " + super.getThreadName()
                    + " interrupted.");
        }

        System.out.println("Thread " + super.getThreadName() + " exiting.");
    }

}

容器：

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ThreadContainer {
    private int capacity;
    private int value;

    private Lock locky = new ReentrantLock(true);

    public ThreadContainer(int capacity) {
        this.capacity = capacity;
        this.value = 0; 
    }

    public void produce(int amount){
        if(this.value + amount <= this.capacity){
            this.value += amount;
        }else{
            this.value = capacity;
        }
    }

    public void consume(int amount){
        if(this.value - amount >= 0 ){
            this.value -= amount;
        }else{
            this.value =0;
        } 
    }

    public boolean canProduce(int am){
        return (this.value + am) <= this.capacity;
    }

    public boolean canConsume(int am){
        return (this.value - am) >= 0;
    }

    public boolean tryLock(){
        if(this.locky.tryLock()){
            this.locky.lock();
            return true;
        }else{
            return false;
        }
    }

    public void unlock(){
        this.locky.unlock();
        this.locky.notify();
    }

    public void waitLock() throws InterruptedException{
        this.locky.wait();
    }


    @Override
    public String toString() {
        return "capacity: " + this.capacity + ", value: " + this.value;
    }

}

主类：

public class RunFrom {
    public static void main(String args[]) {
        ThreadContainer container = new ThreadContainer(25);

        /*
        Producer prod = new Producer("Producer", container);
        prod.start();

        Consumer cons = new Consumer("Consumer", container);
        cons.start();
        */

        int prodCount =0;
        int conCount =0;
        for (int i = 0; i < 5; i++) {
            if(i%2 == 0){
                Producer prod = new Producer("Producer " + prodCount + " [" + i + "]", container);
                prodCount++;
                prod.start();
            }else{
                Consumer cons = new Consumer("Consumer " + conCount + " [" + i + "]", container);
                conCount++;
                cons.start();
            }
        }
    }
}

所以，我在帖子中做了如下修改@fildor link 看起来它是否适用于 2 个线程（1 个消费者和 1 个生产者），但是当我创建更多线程时仍然存在问题..

MyThread 与原版相同
消费者只是消费
生产者只是生产
在库存容器中解决了锁定问题

消费者

//...
    try {
                for (int i = 10; i > 0; i--) {
                    System.out.println(super.getThreadName() + " want to consume: "
                            + i);
                    System.out.println(super.getThreadName() + " consuming: " + i);
                    super.getContainer().consume(i);
                    System.out.println("Container state: " + super.getContainer());
                    Thread.sleep(100);
                }

            } catch (InterruptedException e) {
                System.out.println("Thread " + super.getThreadName()
                        + " interrupted.");
            }
//...

制片人

//...
try {
            for (int i = 10; i > 0; i--) {
                System.out.println(super.getThreadName() + " want to produce: "
                        + i);
                System.out.println(super.getThreadName() + " producing: " + i);
                super.getContainer().produce(i);
                System.out.println("Container state: " + super.getContainer());
                Thread.sleep(100);
            }

        } catch (InterruptedException e) {
            System.out.println("Thread " + super.getThreadName()
                    + " interrupted.");
        }
//...

库存容器

//...
final Lock lock = new ReentrantLock();
    final Condition notFull = lock.newCondition();
    final Condition notEmpty = lock.newCondition();
//...

public void produce(int amount) {
        lock.lock();
        try {
            while (!canProduce(amount)) {
                notFull.wait();
            }

            if (this.value + amount <= this.capacity) {
                this.value += amount;
            } else {
                this.value = capacity;
            }
            notEmpty.signal();
        } catch (InterruptedException e) {
            System.out.println("InterruptedException" + e);
        } finally {
            lock.unlock();
        }
    }

    public void consume(int amount) {
        lock.lock();
        try {
            while (!canConsume(amount)) {
                notEmpty.wait();
            }

            if (this.value - amount >= 0) {
                this.value -= amount;
            } else {
                this.value = 0;
            }
            notFull.signal();
        } catch (InterruptedException e) {
            System.out.println("InterruptedException" + e);
        } finally {
            lock.unlock();
        }
    }

对于 4 个线程（2 个生产者和 2 个消费者）的输出如下所示：

Creating Producer 0 [0]
Starting Producer 0 [0]
Running Producer 0 [0]
Producer 0 [0] want to produce: 10
Producer 0 [0] producing: 10
Container state: capacity: 25, value: 10
Creating Consumer 0 [1]
Starting Consumer 0 [1]
Creating Producer 1 [2]
Starting Producer 1 [2]
Creating Consumer 1 [3]
Running Consumer 0 [1]
Starting Consumer 1 [3]
Creating Producer 2 [4]
Starting Producer 2 [4]
Running Producer 1 [2]
Producer 1 [2] want to produce: 10
Producer 1 [2] producing: 10
Consumer 0 [1] want to consume: 10
Consumer 0 [1] consuming: 10
Container state: capacity: 25, value: 20
Container state: capacity: 25, value: 10
Running Consumer 1 [3]
Consumer 1 [3] want to consume: 10
Running Producer 2 [4]
Producer 2 [4] want to produce: 10
Producer 2 [4] producing: 10
Container state: capacity: 25, value: 20
Consumer 1 [3] consuming: 10
Container state: capacity: 25, value: 10
Producer 0 [0] want to produce: 9
Producer 0 [0] producing: 9
Container state: capacity: 25, value: 19
Consumer 0 [1] want to consume: 9
Consumer 0 [1] consuming: 9
Container state: capacity: 25, value: 10
Producer 1 [2] want to produce: 9
Producer 1 [2] producing: 9
Container state: capacity: 25, value: 19
Producer 2 [4] want to produce: 9
Producer 2 [4] producing: 9
Exception in thread "Producer 2 [4]" Consumer 1 [3] want to consume: 9
Consumer 1 [3] consuming: 9
Container state: capacity: 25, value: 10
java.lang.IllegalMonitorStateException
    at java.lang.Object.wait(Native Method)
    at java.lang.Object.wait(Unknown Source)
    at test.ThreadContainer.produce(ThreadContainer.java:24)
    at test.Producer.run(Producer.java:21)
    at java.lang.Thread.run(Unknown Source)
Producer 0 [0] want to produce: 8
Producer 0 [0] producing: 8
Container state: capacity: 25, value: 18
Consumer 0 [1] want to consume: 8
Consumer 0 [1] consuming: 8
Container state: capacity: 25, value: 10
Producer 1 [2] want to produce: 8
Producer 1 [2] producing: 8
Container state: capacity: 25, value: 18
Consumer 1 [3] want to consume: 8
Consumer 1 [3] consuming: 8
Container state: capacity: 25, value: 10
Producer 0 [0] want to produce: 7
Producer 0 [0] producing: 7
Container state: capacity: 25, value: 17
Consumer 0 [1] want to consume: 7
Consumer 0 [1] consuming: 7
Container state: capacity: 25, value: 10
Producer 1 [2] want to produce: 7
Producer 1 [2] producing: 7
Container state: capacity: 25, value: 17
Consumer 1 [3] want to consume: 7
Consumer 1 [3] consuming: 7
Container state: capacity: 25, value: 10
Producer 0 [0] want to produce: 6
Producer 0 [0] producing: 6
Container state: capacity: 25, value: 16
Consumer 0 [1] want to consume: 6
Consumer 0 [1] consuming: 6
Container state: capacity: 25, value: 10
Producer 1 [2] want to produce: 6
Producer 1 [2] producing: 6
Container state: capacity: 25, value: 16
Consumer 1 [3] want to consume: 6
Consumer 1 [3] consuming: 6
Container state: capacity: 25, value: 10
Producer 0 [0] want to produce: 5
Producer 0 [0] producing: 5
Container state: capacity: 25, value: 15
Consumer 0 [1] want to consume: 5
Consumer 0 [1] consuming: 5
Container state: capacity: 25, value: 10
Producer 1 [2] want to produce: 5
Producer 1 [2] producing: 5
Container state: capacity: 25, value: 15
Consumer 1 [3] want to consume: 5
Consumer 1 [3] consuming: 5
Container state: capacity: 25, value: 10
Producer 0 [0] want to produce: 4
Producer 0 [0] producing: 4
Container state: capacity: 25, value: 14
Consumer 0 [1] want to consume: 4
Consumer 0 [1] consuming: 4
Container state: capacity: 25, value: 10
Producer 1 [2] want to produce: 4
Producer 1 [2] producing: 4
Container state: capacity: 25, value: 14
Consumer 1 [3] want to consume: 4
Consumer 1 [3] consuming: 4
Container state: capacity: 25, value: 10
Producer 0 [0] want to produce: 3
Producer 0 [0] producing: 3
Container state: capacity: 25, value: 13
Consumer 0 [1] want to consume: 3
Consumer 0 [1] consuming: 3
Container state: capacity: 25, value: 10
Producer 1 [2] want to produce: 3
Producer 1 [2] producing: 3
Container state: capacity: 25, value: 13
Consumer 1 [3] want to consume: 3
Consumer 1 [3] consuming: 3
Container state: capacity: 25, value: 10
Producer 0 [0] want to produce: 2
Producer 0 [0] producing: 2
Container state: capacity: 25, value: 12
Consumer 0 [1] want to consume: 2
Consumer 0 [1] consuming: 2
Container state: capacity: 25, value: 10
Producer 1 [2] want to produce: 2
Producer 1 [2] producing: 2
Container state: capacity: 25, value: 12
Consumer 1 [3] want to consume: 2
Consumer 1 [3] consuming: 2
Container state: capacity: 25, value: 10
Producer 0 [0] want to produce: 1
Producer 0 [0] producing: 1
Container state: capacity: 25, value: 11
Consumer 0 [1] want to consume: 1
Consumer 0 [1] consuming: 1
Container state: capacity: 25, value: 10
Producer 1 [2] want to produce: 1
Producer 1 [2] producing: 1
Container state: capacity: 25, value: 11
Consumer 1 [3] want to consume: 1
Consumer 1 [3] consuming: 1
Container state: capacity: 25, value: 10
Thread Producer 0 [0] exiting.
Thread Consumer 0 [1] exiting.
Thread Producer 1 [2] exiting.
Thread Consumer 1 [3] exiting.

难道不是因为更多的消费者在等待相同的信号，而更多的生产者在等待另一个信号吗？

【问题讨论】：

示例 here 我猜应该对您有所帮助...
谷歌搜索一下可以找到几十种不同类型的示例。

标签： java multithreading mutex thread-synchronization

【解决方案1】：

您真的应该在某处阅读Java 101 中的线程 教程。您得到的异常是因为您正在等待一个对象而没有获取内在锁。给定任何用lock 标识的对象，惯用代码是：

synchronized (lock) {
  while (!condition) {
    lock.wait();
  }
}

【讨论】：

当您可以在 SO 上发布每条错误消息时，为什么还要阅读教程。
那么，同步的解决方案也不好？ :D
顺便说一句 @Kayaman 与其发现我做错了，不如写下我做错了什么；）仅供参考
在利用代码审查之前，您应该了解太多事情。如果你愿意学习，看看我刚刚写的this Gist
@xxxvodnikxxx 这会浪费我的时间。你应该是谷歌搜索的人，因为即使在 StackOverflow 上也有数以千计的正确示例。学会帮助自己。

【解决方案2】：

为什么这么复杂？

MyThread 课程有什么好处？我所看到的只是启动线程的代码。我不需要为此定义一个新类。我可以用一行代码启动一个新线程：

new Thread(new Producer(...)).start;

然后是你的ThreadContainer 类，它的方法只能由消费者线程调用，而其他方法只能由生产者线程调用。这违反了一个基本的设计原则：每个类应该只负责一件事。

您的许多代码都是将事物绑定到其他事物的粘合剂。（例如，getter 和 setter 方法是胶水，而您的 MyThread 类不过是胶水）。将事物与其他事物绑定称为耦合，当您拥有很多时，称为紧耦合。

程序的各个部分耦合得越紧密，程序就越难理解和改变。紧密耦合的系统更容易被破坏，如果它们被破坏，它们就更难诊断和修复。

松散耦合总是更好：当不是绝对必要时，不要让类和方法依赖。

最后，如果我想演示生产者/消费者的概念，我不会使用 wait()/notify() 来混淆演示。 wait() 和 notify() 方法是 低级原语，用于实现更高级别的同步对象，它们应该对在更高级别运行的代码隐藏.

如果你想演示wait()/notify()，那是一回事。如果你想展示生产者/消费者，那是另一个。我不会试图将两个演示都塞进一个程序中。每个概念只会妨碍展示另一个概念。

Java 标准库提供了许多现成的、更高级别的同步对象供您使用。可能最通用的是BlockingQueue。我的生产者/消费者演示将有一个生产者线程将“产品”塞入ArrayBlockingQueue，以及一个消费者线程将它们拉出并对其进行操作。

【讨论】：