即使条件停止为零，线程池也会将数据减少为负值

Question

处理器class-

public class Processor extends Thread {

    private static int stock = 10;

    private static Object lock1 = new Object();
    private static Object lock2 = new Object();
    private static Object lock3 = new Object();

    private void snooze(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private boolean isStockEmpty() {
        boolean value;
        synchronized (lock1) {
            if (stock == 0) {
                value = true;
            } else {
                value = false;
            }
        }
        return value;
    }

    private void decreaseStock() {
        synchronized (lock2) {
            stock--;
        }
    }

    private int getStockCount() {
        int value;
        synchronized (lock3) {
            value = stock;
        }
        return value;
    }

    private  void doWork() {
        if (!isStockEmpty()) {
            decreaseStock();
            System.out.println(Thread.currentThread().getName() + " takes 1 item from stock\n" +
                    "Items remaining in stock: " + getStockCount());
            snooze(2000);
        } else {
            System.out.println("Stock is empty, " + Thread.currentThread().getName() + "is idle\n" +
                    "Items remaining in stock: " + getStockCount());
            snooze(2000);
        }
    }

    @Override
    public void run() {
        doWork();
    }
}

主要方法-

public class Main {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        for (int i = 1; i <= 12; i++) {
            executorService.submit(new Processor());
        }
        executorService.shutdown();
        System.out.println("All tasks successfully submitted");
        executorService.awaitTermination(1, TimeUnit.DAYS);
        System.out.println("All tasks completed successfully");
    }
}

在这里，我创建了一个包含三个线程的线程池，并为它们分配了一项工作，从 10 件库存中取出 12 件物品，条件是如果库存变空，线程将闲置。

这个程序不是线程安全的，最后的存货数变成负数。

如何使这个程序线程安全？

Answer 1

How to make this program thread safe?

首先确定线程间的共享状态；可以识别四个项目，即：stock、lock1、lock2和lock3。这些字段是 static、hence shared among threads。现在寻找涉及该共享状态的潜在竞争条件。

这些字段是否只被读取？如果是，则不存在竞争条件。这些字段是否以某种方式被修改？是的，那么您需要确保 mutual exclusion 可以访问这些字段。

字段lock1、lock2和lock3的用法如下：

synchronized (lock1) { ... }
... 
synchronized (lock2) { ... }
...
synchronized (lock3) { ... }

因此，那里没有竞争条件。字段 stock 怎么样？！我们在方法 isStockEmpty ( 即 stock == 0 中进行读取，在方法 decreaseStock ( 中进行读取和写入]即stock--;)，最后在方法getStockCount中进行另一次读取（即value = stock;）。因此，多个线程可能会并行发生多个读取和写入，因此，必须确保在该字段上互斥。您添加了同步部分，但是需要使用同一个锁来保证线程不会并发读写。

从oracle tutorial可以读到：

Every object has an intrinsic lock associated with it. By convention, a thread that needs exclusive and consistent access to an object's fields has to acquire the object's intrinsic lock before accessing them, and then release the intrinsic lock when it's done with them. A thread is said to own the intrinsic lock between the time it has acquired the lock and released the lock. As long as a thread owns an intrinsic lock, no other thread can acquire the same lock. The other thread will block when it attempts to acquire the lock.

因此，与其使用三个不同的对象进行同步以确保相同数据的互斥，不如只使用一个，这样您的代码将如下所示：

public class Processor extends Thread {
  
    private static final Object stock_lock = new Object();

    @GuardedBy("lock")
    private static int stock = 10;

    private void snooze(long millis) { ...}

    private boolean isStockEmpty() {
       synchronized (stock_lock) {
           return stock == 0;
       } 
    }

    private void decreaseStock() {
        synchronized (stock_lock) {
            stock--;
        }
    }

    private int getStockCount() {
       synchronized (stock_lock) {
          return stock;
       }
   }

   ...
}

程序现在是线程安全的了吗？！，几乎但是那里仍然存在偷偷摸摸的竞争条件，即：

if (!isStockEmpty()) {
    decreaseStock();

尽管 isStockEmpty 和 decreaseStock 这两个方法分别是线程安全的，但是当它们一起调用时就不是了，为什么？因为检查库存是否为空并减少库存的整个操作需要顺序完成。否则，可能会发生以下竞争条件：

字段 stock 为 1，Thread 1 在 isStockEmpty 中同步检查是否为空并且 !isStockEmpty() 计算为 true、Thread 1 继续调用 decreaseStock()，同时（在 Thread 1 调用 synchronized (stock_lock) 之前）Thread 2 也调用 !isStockEmpty()这也将评估为 true。 Thread 1 执行操作 stock--，使 stock = 0，并且因为 Thread 2 已经在 if (!isStockEmpty()) 的块包装内，Thread 2 也将执行 stock--，使库存=-1。此外，在 doWork 方法内部调用的 getStockCount() 也有类似的 race condition。解决方法是同步整个代码块，即：

private void doWork() {
    synchronized (stock_lock) {
        ....
    }
}

现在，因为 isStockEmpty、decreaseStock 和 getStockCount 都是在 synchronized (stock_lock) 中调用的 private 方法doWork 方法，我们实际上可以从这些方法中删除我们在开始时添加的同步。所以整个代码看起来像这样：

public class Processor extends Thread {
    private static final Object stock_lock = new Object();
    
    @GuardedBy("lock")
    private static int stock = 10;
    
    
    private void snooze(long millis) {...}
    
    private boolean isStockEmpty() { return stock == 0; }
    
    private void decreaseStock() { stock--;}
    
    private int getStockCount() {  return stock;}
    
    private void doWork() {
        synchronized (stock_lock) {
            ....
        }
    }
    
    @Override
    public void run() {
        doWork();
    }
}

现在在现实生活示例中，如果您像那样同步整个程序，您还不如按顺序执行代码。

或者，您可以通过对 stock 字段变量使用 AtomicInteger 使当前程序线程安全：

private static AtomicInteger stock = new AtomicInteger(10);

private void snooze(long millis) {...  }
private void doWork() {
        int value = stock.getAndDecrement();
        if (value != 0) {
            System.out.println(Thread.currentThread().getName() + " takes 1 item from stock\n" +
                    "Items remaining in stock: " + value);
            snooze(2000);
        } else {
            System.out.println("Stock is empty, " + Thread.currentThread().getName() + "is idle\n" +
                    "Items remaining in stock: " + value);
            snooze(2000);
        }
}

没有竞争条件 因为 1) getAndDecrement 是自动完成的； 2) 我们将返回值保存到局部变量（线程私有）中，并使用该值代替 getStockCount。尽管如此，stock 变量在理论上可以获得负值。但是不会显示这些值。

即使条件停止为零，线程池也会将数据减少为负值

Thread Pools decreasing data to negative even when condition is to stop at zero

java

multithreading

parallel-processing

thread-safety

executorservice