生产者消费者使用可重入锁不起作用答案

【问题标题】：Producer Consumer using Reentrant lock not working生产者消费者使用可重入锁不起作用
【发布时间】：2018-12-14 07:26:07
【问题描述】：

我正在尝试使用ReentrantLock 来实现消费者-生产者，如下所示：

class Producer implements Runnable {

    private List<String> data;
    private ReentrantLock lock;

    Producer(List<String> data,ReentrantLock lock)
    {
        this.data = data;
        this.lock = lock;
    }

    @Override
    public void run() {
        int counter = 0;
        synchronized (lock)
        {
            while (true)
            {

                if ( data.size() < 5)
                {
                    counter++;
                    data.add("writing:: "+counter);
                }
                else
                {
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

    }
}

class Consumer implements Runnable{

    private List<String> data;
    private ReentrantLock lock;

    Consumer(List<String> data,ReentrantLock lock)
    {
        this.data = data;
        this.lock = lock;
    }

    @Override
    public void run() {
        int counter = 0;
        synchronized (lock)
        {
            while (true)
            {

                if ( data.size() > 0)
                {
                    System.out.println("reading:: "+data.get(data.size()-1));
                    data.remove(data.size()-1);
                }
                else
                {
                    System.out.println("Notifying..");
                    lock.notify();
                }
            }
        }

    }
}

public class ProducerConsumer  {

    public static void main(String[] args) {
        List<String> str = new LinkedList<>();
        ReentrantLock lock= new ReentrantLock();
        Thread t1 = new Thread(new Producer(str,lock));
        Thread t2 = new Thread(new Consumer(str,lock));
        t1.start();
        t2.start();
    }
}

因此，它只向列表写入一次，然后消费者无限期地等待。为什么会这样？为什么 Producer 不获取锁？

【问题讨论】：

你读过这方面的教程吗？例如，docs.oracle.com/javase/tutorial/essential/concurrency/…
阅读链接后，我想我应该尝试使用 try {lock.lock()} 最后 {lock.unlock()} 相同吗？但是，我对使用它有点困惑。

标签： java multithreading locking producer-consumer reentrantlock

【解决方案1】：

您使用lock() 和unlock() 获取和释放ReentrantLock，而不是synchronized。

正如 Antoniosss 所指出的，这是一个死锁，其中一个线程正在等待另一个永远不会放弃的锁（当两个线程试图协调该锁时）。这是一种解决方案：

import static java.util.Objects.requireNonNull;

import java.util.LinkedList;
import java.util.List;

class Producer implements Runnable {
    private final List<String> data;

    Producer(List<String> data) {
        this.data = requireNonNull(data);
    }

    @Override
    public void run() {
        int counter = 0;
        while (true) {
            synchronized (data) {
                if (data.size() < 5) {
                    counter++;
                    data.add("writing:: " + counter);
                } else {
                    try {
                        data.wait();
                    } catch (InterruptedException e) {
                        return;
                    }
                }
            }
        }
    }
}

class Consumer implements Runnable {
    private final List<String> data;

    Consumer(List<String> data) {
        this.data = requireNonNull(data);
    }

    @Override
    public void run() {
        while (true) {
            synchronized (data) {
                if (data.size() > 0) {
                    System.out.println("reading:: " + data.get(data.size() - 1));
                    data.remove(data.size() - 1);
                }
                data.notify();
            }
        }
    }
}

public class ProducerConsumer {
    public static void main(String[] args) {
        List<String> data = new LinkedList<>();
        Thread t1 = new Thread(new Producer(data));
        Thread t2 = new Thread(new Consumer(data));
        t1.start();
        t2.start();
    }
}

我们已经取消了锁定对象并在列表本身上进行同步。生产者在 while 循环内同步，其效果是一旦列表中至少有 5 个项目，生产者将等待，放弃列表上的监视器。

消费者也在循环内部同步，这很关键，因为在您的代码中，一旦获得锁，它就不会放弃监视器。事实上，如果你先启动了消费者（或者很不幸），那么根本就不会产生任何东西。当离开同步块或方法时，或者当线程持有监视器 wait()s 时释放监视器，但 notify() 和 notifyAll() 不释放监视器。

消费者读取最后一项，如果有的话，立即通知生产者并释放锁。两个注意事项：

首先，不清楚您期望的物品顺序。您是希望生产者生产 5 个，消费者消费 5 个，等等，还是希望 5 只是一个限制，以至于无法形成积压（这很好，这称为背压），但是消费者只要有货就急切地消费？这个实现是后者。

其次，消费者一旦释放它就会尝试获取列表中的监视器。这是忙等待的一种形式，消费者和生产者竞相获取锁，可能消费者经常赢得这场比赛，一旦列表为空，这将变得毫无意义。在 Java 9 或更高版本中，在同步块之外但在使用者的 while 循环内调用 onSpinWait 可能是明智的。在早期版本中，yield 可能是合适的。但是在我的测试中，没有任何一个代码都可以正常工作。

Antoniossss 提出了另一个建议，即使用 LinkedBlockingQueue，但目前的代码总是采用最后一项，使用队列会改变这种行为。相反，我们可以使用双端队列（双端队列），将项目放在末尾并从末尾取出。看起来是这样的：

import static java.util.Objects.requireNonNull;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;

class Producer implements Runnable {
    private final BlockingDeque<String> data;

    Producer(BlockingDeque<String> data) {
        this.data = requireNonNull(data);
    }

    @Override
    public void run() {
        int counter = 0;
        while (true) {
            counter++;
            try {
                data.put("writing:: " + counter);
            } catch (InterruptedException e) {
                break;
            }
        }
    }
}

class Consumer implements Runnable {
    private final BlockingDeque<String> data;

    Consumer(BlockingDeque<String> data) {
        this.data = requireNonNull(data);
    }

    @Override
    public void run() {
        while (true) {
            try {
                System.out.println("reading:: " + data.takeLast());
            } catch (InterruptedException e) {
                break;
            }
        }
    }
}

public class ProducerConsumer {
    public static void main(String[] args) {
        BlockingDeque<String> data = new LinkedBlockingDeque<>(5);
        Thread t1 = new Thread(new Producer(data));
        Thread t2 = new Thread(new Consumer(data));
        t1.start();
        t2.start();
    }
}

因为LinkedBlockingDeque 是一个并发数据结构，我们不需要任何同步块或在此处等待或通知。我们可以简单地从双端队列尝试put 和takeLast，如果双端队列已满或空，它将分别阻塞。双端队列的容量为 5，因此如果生产者领先那么远，它会向生产者施加背压，就像原来的一样。

没有什么可以阻止生产者尽可能快地生产元素，消费者可以消费它们，这意味着第一个元素可能需要等待任意长的时间才能被消费。我不清楚这是否是您的代码的意图。有一些方法可以实现这一点，或者通过再次引入wait() 和notify()，通过使用Semaphores，或其他方式，但我会保留它，因为不清楚你是否想要它。

关于InterruptedException 的最后一点说明。如果有人在线程上调用interrupt()，就会发生这种情况，但唯一持有对生产者和消费者线程的引用的是main() 方法，并且它永远不会中断它们。所以这里不应该发生异常，但如果它以某种方式发生，我只是让生产者或消费者退出。在更复杂的场景中，如果线程处于睡眠状态或处于阻塞方法（或者甚至在一个方法之外，如果它显式检查它），中断线程可以用作向它发出信号的一种方式，但我们在这里没有使用它。

【讨论】：

这并没有错，但它本身是没有用的，也没有解决他们代码中的问题。他们如何与ReentrantLock 进行跨线程通信，相当于他们对wait 和notify 的调用？他们如何正确地做到这一点？
@SotiriosDelimanolis 我认为如果一个答案必须提供一个有效的替代品，那么这个问题应该被简单地关闭为太宽泛;否则，提问者应该解决问题并提出另一个问题。但我会扩展我的答案。

【解决方案2】：

正如你想要的那样，Producer 不停地产生值，Consumer 不停地消耗这些值并最终等待值被生成，因为没有值，在同步和锁定时跳过并使用@987654323 @

简单地生产你：

queue.put(value)

在消费者中你会这样做

value=queue.take();

瞧，消费者将获取任何值，并等待值队列为空。

至于你的代码：

您根本没有使用ReentrantLock。您可以将其替换为 new Object 并且输出将是相同的。 wait 和 notify 是 Object 的方法。

您只能设法产生单一价值的原因在于您的消费者。实际上你有这样的东西：

synchronized(lock){ // aquire lock's monitor

while(true){

  lock.notify();
 }

} // release lock's monitor - never doing that, thread never leaves this block

这里的问题是你执行永远不会离开同步块。所以你正在调用notify，但你并没有通过退出synchronized 块来释放lock 监视器。生产者被“通知”，但为了继续执行，它必须重新获取lock 的监视器——但它可以，因为消费者从不释放它。这里几乎是典型的僵局。

你可以想象房间里有几个人，只有一个拿着棍子的人可以说话。所以第一个得到了syncronized(stick) 的支持。做它必须做的，并决定它必须为其他人wait，所以他打电话给wait并把棍子递回去。现在第二个人可以说话，做他的工作，并决定给他棍子的那个人现在可以继续。他打电话给notify - 现在他必须通过离开synchronized(stick) 块来传递棒。如果他不这样做，第一人称将无法继续 - 这就是您的情况。

【讨论】：

这不是死锁，我不知道几乎是什么意思。您使用 LinkedBlockingQueue 的建议违背了他们代码的目的。您对行为的解释很好，但没有说明如何纠正它。我认为答案没有用。为什么总是要破坏或滥用？

【解决方案3】：

@Antoniossss 是对的。您没有正确使用 ReentrantLock，而是可以将其替换为 Object。如果您想改用 ReentrantLock（这是最新的），那么我会建议类似：

package Multithreading;


import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

class Producer implements Runnable{

  private List<String> data;
  private ReentrantLock lock;

  Producer(List<String> data,ReentrantLock lock)
  {
    this.data = data;
    this.lock = lock;
  }

  @Override
  public void run() {
    int counter = 0;
    while (true)
    {


        try {
          lock.lock();
          if ( data.size() < 5)
          {
            counter++;
            data.add("writing:: " + counter);
          }
        }finally {
          lock.unlock();
        }

      try {
        TimeUnit.SECONDS.sleep(1);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }

}

class Consumer implements Runnable{

  private List<String> data;
  private ReentrantLock lock;

  Consumer(List<String> data,ReentrantLock lock)
  {
    this.data = data;
    this.lock = lock;
  }

  @Override
  public void run() {
    int counter = 0;
    while (true)
    { 
      try {
        lock.lock();
          if ( data.size() > 0) 
          {
            System.out.println("reading:: "+data.get(data.size()-1));
            data.remove(data.size()-1);
          }
       }finally {
         lock.unlock();
       }
       try 
       {
          TimeUnit.SECONDS.sleep(1);
       } catch (InterruptedException e) {
         e.printStackTrace();
       }
      }
    }
  }
}

public class ProducerConsumer  {

  public static void main(String[] args) {
    List<String> str = new LinkedList<>();
    ReentrantLock lock= new ReentrantLock();
    Thread t1 = new Thread(new Producer(str,lock));
    Thread t2 = new Thread(new Consumer(str,lock));
    t1.start();
    t2.start();
  }
}

我删除了通知调用，但如果你真的需要一次允许一个线程，只需使用 lock.notify()

【讨论】：

你应该在检查data.size()之前锁定

【解决方案4】：

我想指出你犯的两个错误：

ReentrantLock 的错误用法。

每个对象都可以用于内部锁，因此无需查找特定的Lock 类。由于每个synchronized 块都由一个方法限定，并且您不需要any enhanced means，因此ReentrantLock 在这里是多余的。

synchronized 块的位置不正确。

一旦你进入一个同步区块，在你离开之前没有人可以进入那里。显然，你永远不会因为while(true)而退出它。

我建议您删除 ReentrantLocks 并在 data 上进行同步。

@Override
public void run() {
    int counter = 0;
    while (true) {
        synchronized (data) {
            if (data.size() < 5) {
                data.add("writing:: " + ++counter);
            } else {
                try {
                    data.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        // we are out of the synchornized block here
        // to let others use data as a monitor somewhere else
    }
}

【讨论】：