Java 编程问题：十一、并发-深入探索4-阿里云开发者社区

Java 编程问题：十一、并发-深入探索3https://developer.aliyun.com/article/1426167

221 原子变量

通过Runnable计算从 1 到 1000000 的所有数字的简单方法如下所示：

public class Incrementator implements Runnable {
  public [static] int count = 0;
  @Override
  public void run() {
    count++;
  }
  public int getCount() {
    return count;
  }
}

让我们旋转五个线程，同时递增count变量：

Incrementator nonAtomicInc = new Incrementator();
ExecutorService executor = Executors.newFixedThreadPool(5);
for (int i = 0; i < 1 _000_000; i++) {
  executor.execute(nonAtomicInc);
}

但是，如果我们多次运行此代码，会得到不同的结果，如下所示：

997776, 997122, 997681 ...

所以，为什么我们不能得到预期的结果，1000000？原因是count++不是原子操作/动作。它由三个原子字节码指令组成：

iload_1
iinc 1, 1
istore_1

在一个线程中，读取count值并逐个递增，另一个线程读取较旧的值，导致错误的结果。在多线程应用中，调度器可以停止在这些字节码指令之间执行当前线程，并启动一个新线程，该线程在同一个变量上工作。我们可以通过同步来修复问题，或者通过原子变量来更好地解决问题。

原子变量类在java.util.concurrent.atomic中可用。它们是将争用范围限制为单个变量的包装类；它们比 Java 同步轻量级得多，基于 CAS（简称比较交换）：现代 CPU 支持这种技术，它将给定内存位置的内容与给定值进行比较，如果当前值等于预期值，则更新为新值。主要是以类似于volatile的无锁方式影响单个值的原子复合作用。最常用的原子变量是标量：

AtomicInteger
AtomicLong
AtomicBoolean
AtomicReference

并且，以下是针对数组的：

AtomicIntegerArray
AtomicLongArray
AtomicReferenceArray

让我们通过AtomicInteger重写我们的示例：

public class AtomicIncrementator implements Runnable {
  public static AtomicInteger count = new AtomicInteger();
  @Override
  public void run() {
    count.incrementAndGet();
  }
  public int getCount() {
    return count.get();
  }
}

注意，我们写的不是count++，而是count.incrementAndGet()。这只是AtomicInteger提供的方法之一。此方法以原子方式递增变量并返回新值。这一次，count将是 1000000。

下表列出了几种常用的AtomicInteger方法。左栏包含方法，右栏包含非原子含义：

AtomicInteger ai = new AtomicInteger(0); // atomic
int i = 0; // non-atomic
// and
int q = 5;
int r;
// and
int e = 0;
boolean b;

原子操作	非原子对应物
`r = ai.get();`	`r = i;`
`ai.set(q);`	`i = q;`
`r = ai.incrementAndGet();`	`r = ++i;`
`r = ai.getAndIncrement();`	`r = i++;`
`r = ai.decrementAndGet();`	`r = --i;`
`r = ai.getAndDecrement();`	`r = i--;`
`r = ai.addAndGet(q);`	`i = i + q; r = i;`
`r = ai.getAndAdd(q);`	`r = i; i = i + q;`
`r = ai.getAndSet(q);`	`r = i; i = q;`
`b = ai.compareAndSet(e, q);`	`if (i == e) { i = q; return true; } else { return false; }`

让我们通过原子操作解决几个问题：

通过updateAndGet(IntUnaryOperator updateFunction)更新数组元素：

// [9, 16, 4, 25]
AtomicIntegerArray atomicArray
  = new AtomicIntegerArray(new int[] {3, 4, 2, 5});
for (int i = 0; i < atomicArray.length(); i++) {
  atomicArray.updateAndGet(i, elem -> elem * elem);
}

通过updateAndGet(IntUnaryOperator updateFunction)更新单个整数：

// 15
AtomicInteger nr = new AtomicInteger(3);
int result = nr.updateAndGet(x -> 5 * x);

通过accumulateAndGet(int x, IntBinaryOperator accumulatorFunction)更新单个整数：

// 15
AtomicInteger nr = new AtomicInteger(3);
// x = 3, y = 5
int result = nr.accumulateAndGet(5, (x, y) -> x * y);

通过addAndGet(int delta)更新单个整数：

// 7
AtomicInteger nr = new AtomicInteger(3);
int result = nr.addAndGet(4);

通过compareAndSet(int expectedValue, int newValue)更新单个整数：

// 5, true
AtomicInteger nr = new AtomicInteger(3);
boolean wasSet = nr.compareAndSet(3, 5);

从 JDK9 开始，原子变量类被多种方法所丰富，如get/setPlain()、get/setOpaque()、getAcquire()以及它们的同伴。要了解这些方法，请看一下 Doug Lea 的《使用 JDK9 内存顺序模式》。

加法器和累加器

在 JavaAPI 文档之后，如果多线程应用更新频繁，但读取频率较低，建议使用LongAdder、DoubleAdder、LongAccumulator、DoubleAccumulator，而不是AtomicFoo类。对于这种情况，这些类的设计是为了优化线程的使用。

这意味着，我们不需要使用AtomicInteger来计算从 1 到 1000000 的整数，而可以使用LongAdder如下：

public class AtomicAdder implements Runnable {
  public static LongAdder count = new LongAdder();
  @Override
  public void run() {
    count.add(1);
  }
  public long getCount() {
    return count.sum();
  }
}

或者，我们可以使用LongAccumulator如下：

public class AtomicAccumulator implements Runnable {
  public static LongAccumulator count
    = new LongAccumulator(Long::sum, 0);
  @Override
  public void run() {
    count.accumulate(1);
  }
  public long getCount() {
    return count.get();
  }
}

LongAdder和DoubleAdder适用于暗示加法的场景（特定于加法的操作），而LongAccumulator和DoubleAccumulator适用于依赖给定函数组合值的场景。

222 重入锁

Lock接口包含一组锁定操作，可以显式地用于微调锁定过程（它提供比内在锁定更多的控制）。其中，我们有轮询、无条件、定时和可中断的锁获取。基本上，Lock用附加功能公开了synchronized关键字的Future。Lock接口如下图所示：

public interface Lock {
  void lock();
  void lockInterruptibly() throws InterruptedException;
  boolean tryLock();
  boolean tryLock(long timeout, TimeUnit unit)
  throws InterruptedException;
  void unlock();
  Condition newCondition();
}

Lock的实现之一是ReentrantLock。可重入锁的作用如下：当线程第一次进入锁时，保持计数设置为 1。在解锁之前，线程可以重新进入锁，从而使每个条目的保持计数增加一。每个解锁请求将保留计数减一，当保留计数为零时，将打开锁定的资源。

与synchronized关键字具有相同的坐标，ReentrantLock遵循以下实现习惯用法：

Lock / ReentrantLock lock = new ReentrantLock();
...
lock.lock();
try {
  ...
} finally {
  lock.unlock();
}

对于非公平锁，未指定线程被授予访问权限的顺序。如果锁应该是公平的（优先于等待时间最长的线程），那么使用ReentrantLock(boolean fair)构造器。

通过ReentrantLock将 1 到 1000000 之间的整数求和可以如下完成：

public class CounterWithLock {
  private static final Lock lock = new ReentrantLock();
  private static int count;
  public void counter() {
    lock.lock();
    try {
      count++;
    } finally {
      lock.unlock();
    }
  }
}

让我们通过几个线程来使用它：

CounterWithLock counterWithLock = new CounterWithLock();
Runnable task = () -> {
  counterWithLock.counter();
};
ExecutorService executor = Executors.newFixedThreadPool(8);
for (int i = 0; i < 1 _000_000; i++) {
  executor.execute(task);
}

完成！

另外，下面的代码表示一种基于ReentrantLock.lockInterruptibly()解决问题的习惯用法。绑定到本书的代码附带了一个使用lockInterruptibly()的示例：

Lock / ReentrantLock lock = new ReentrantLock();
public void execute() throws InterruptedException {
  lock.lockInterruptibly();
  try {
    // do something
  } finally {
    lock.unlock();
  }
}

如果持有此锁的线程被中断，则抛出InterruptedException。用lock()代替lockInterruptibly()不接受中断。

另外，下面的代码表示使用ReentrantLock.tryLock(long timeout, TimeUnit unit) throws InterruptedException的习惯用法。本书附带的代码还有一个示例：

Lock / ReentrantLock lock = new ReentrantLock();
public boolean execute() throws InterruptedException {
  if (!lock.tryLock(n, TimeUnit.SECONDS)) {
    return false;
  }
  try {
    // do something
  } finally {
    lock.unlock();
  }
  return true;
}

注意，tryLock()尝试获取指定时间的锁。如果这段时间过去了，那么线程将不会获得锁。它不会自动重试。如果线程在获取锁的过程中被中断，则抛出InterruptedException。

最后，绑定到本书的代码附带了一个使用ReentrantLock.newCondition()的示例。下一个屏幕截图显示了这个成语：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-QcGAKhWJ-1657346000216)(https://github.com/apachecn/apachecn-java-zh/raw/master/docs/java-coding-prob/img/1bb55dfb-35d8-4092-9924-66c157844a72.png)]

223 重入读写锁

通常，读写连接（例如，读写文件）应基于两个语句完成：

只要没有编写器（共享悲观锁），读者就可以同时阅读。
单个写入程序一次可以写入（独占/悲观锁定）。

下图显示了左侧的读取器和右侧的写入器：

主要由ReentrantReadWriteLock实现以下行为：

为两个锁（读锁和写锁）提供悲观锁语义。
如果某些读取器持有读锁，而某个写入器需要写锁，则在写入器释放写锁之前，不允许更多的读取器获取读锁。
写入程序可以获得读锁，但读取器不能获得写锁。

对于非公平锁，未指定线程被授予访问权限的顺序。如果锁应该是公平的（优先于等待时间最长的线程），那么使用ReentrantReadWriteLock(boolean fair)构造器。

ReentrantReadWriteLock的用法如下：

ReadWriteLock / ReentrantReadWriteLock lock 
  = new ReentrantReadWriteLock();
...
lock.readLock() / writeLock().lock();
try {
  ...
} finally {
  lock.readLock() / writeLock().unlock();
}

下面的代码表示一个ReentrantReadWriteLock用例，它读取和写入一个整数量变量：

public class ReadWriteWithLock {
  private static final Logger logger
    = Logger.getLogger(ReadWriteWithLock.class.getName());
  private static final Random rnd = new Random();
  private static final ReentrantReadWriteLock lock
    = new ReentrantReadWriteLock(true);
  private static final Reader reader = new Reader();
  private static final Writer writer = new Writer();
  private static int amount;
  private static class Reader implements Runnable {
    @Override
    public void run() {
      if (lock.isWriteLocked()) {
        logger.warning(() -> Thread.currentThread().getName() 
          + " reports that the lock is hold by a writer ...");
      }
      lock.readLock().lock();
      try {
        logger.info(() -> "Read amount: " + amount 
          + " by " + Thread.currentThread().getName());
      } finally {
        lock.readLock().unlock();
      }
    }
  }
  private static class Writer implements Runnable {
    @Override
    public void run() {
        lock.writeLock().lock();
        try {
          Thread.sleep(rnd.nextInt(2000));
          logger.info(() -> "Increase amount with 10 by " 
            + Thread.currentThread().getName());
          amount += 10;
        } catch (InterruptedException ex) {
          Thread.currentThread().interrupt();
          logger.severe(() -> "Exception: " + ex);
        } finally {
          lock.writeLock().unlock();
        }
      }
      ...
  }

让我们用两个读取器和四个写入器执行 10 次读卡和 10 次写卡：

ExecutorService readerService = Executors.newFixedThreadPool(2);
ExecutorService writerService = Executors.newFixedThreadPool(4);
for (int i = 0; i < 10; i++) {
  readerService.execute(reader);
  writerService.execute(writer);
}

可能的输出如下：

[09:09:25] [INFO] Read amount: 0 by pool-1-thread-1
[09:09:25] [INFO] Read amount: 0 by pool-1-thread-2
[09:09:26] [INFO] Increase amount with 10 by pool-2-thread-1
[09:09:27] [INFO] Increase amount with 10 by pool-2-thread-2
[09:09:28] [INFO] Increase amount with 10 by pool-2-thread-4
[09:09:29] [INFO] Increase amount with 10 by pool-2-thread-3
[09:09:29] [INFO] Read amount: 40 by pool-1-thread-2
[09:09:29] [INFO] Read amount: 40 by pool-1-thread-1
[09:09:31] [INFO] Increase amount with 10 by pool-2-thread-1
...

在决定依赖ReentrantReadWriteLock之前，请考虑它可能会挨饿（例如，当作家被优先考虑时，读者可能会挨饿）。此外，我们无法将读锁升级为写锁（可以从写入器降级为读取器），并且不支持乐观读。如果其中任何一个问题对您来说很重要，那么请考虑StampedLock，我们将在下一个问题中研究它。

224 冲压锁

简言之，StampedLock的性能优于ReentrantReadWriteLock，支持乐观读取。它不像可重入的；因此容易死锁。主要地，锁获取返回一个戳记（一个long值），它在finally块中用于解锁。每次尝试获取一个锁都会产生一个新的戳，如果没有可用的锁，那么它可能会阻塞，直到可用为止。换句话说，如果当前线程持有锁，并且再次尝试获取锁，则可能导致死锁。

StampedLock读/写编排过程通过以下几种方法实现：

readLock()：非独占获取锁，必要时阻塞，直到可用。对于获取读锁的非阻塞尝试，我们必须tryReadLock()。对于超时阻塞，我们有tryReadLock(long time, TimeUnit unit)。退回的印章用于unlockRead()。
writeLock()：独占获取锁，必要时阻塞直到可用。对于获取写锁的非阻塞尝试，我们有tryWriteLock()。对于超时阻塞，我们有tryWriteLock(long time, TimeUnit unit)。退回的印章用于unlockWrite()。
tryOptimisticRead()：这是给StampedLock增加一个大加号的方法。此方法返回一个应通过validate()标志方法验证的戳记。如果锁当前未处于写入模式，则返回的戳记仅为非零。

readLock()和writeLock()的成语非常简单：

StampedLock lock = new StampedLock();
...
long stamp = lock.readLock() / writeLock();
try {
  ...
} finally {
  lock.unlockRead(stamp) / unlockWrite(stamp);
}

试图给tryOptimisticRead()一个成语可能会导致以下结果：

StampedLock lock = new StampedLock();
int x; // a writer-thread can modify x
...
long stamp = lock.tryOptimisticRead();
int thex = x;
if (!lock.validate(stamp)) {
  stamp = lock.readLock();
  try {
    thex = x;
  } finally {
    lock.unlockRead(stamp);
  }
}
return thex;

在这个习惯用法中，注意初始值（x）是在获得乐观读锁之后分配给thex变量的。然后利用validate()标志法验证了自给定戳的发射度以来，戳锁没有被独占获取。如果validate()返回false（相当于在获得乐观锁之后由线程获取写锁），则通过阻塞readLock()获取读锁，并再次赋值（x。请记住，如果有任何写锁，读锁可能会阻塞。获取乐观锁允许我们读取值，然后验证这些值是否有任何更改。只有在存在的情况下，我们才能通过阻塞读锁。

下面的代码表示一个StampedLock用例，它读取和写入一个整数量变量。基本上，我们通过乐观的方式重申了前一个问题的解决方案：

public class ReadWriteWithStampedLock {
  private static final Logger logger
    = Logger.getLogger(ReadWriteWithStampedLock.class.getName());
  private static final Random rnd = new Random();
  private static final StampedLock lock = new StampedLock();
  private static final OptimisticReader optimisticReader
    = new OptimisticReader();
  private static final Writer writer = new Writer();
  private static int amount;
  private static class OptimisticReader implements Runnable {
    @Override
    public void run() {
      long stamp = lock.tryOptimisticRead();
      // if the stamp for tryOptimisticRead() is not valid
      // then the thread attempts to acquire a read lock
      if (!lock.validate(stamp)) {
        stamp = lock.readLock();
        try {
          logger.info(() -> "Read amount (read lock): " + amount 
            + " by " + Thread.currentThread().getName());
        } finally {
          lock.unlockRead(stamp);
        }
      } else {
        logger.info(() -> "Read amount (optimistic read): " + amount 
          + " by " + Thread.currentThread().getName());
      }
    }
  }
  private static class Writer implements Runnable {
    @Override
    public void run() {
      long stamp = lock.writeLock();
      try {
        Thread.sleep(rnd.nextInt(2000));
        logger.info(() -> "Increase amount with 10 by " 
          + Thread.currentThread().getName());
        amount += 10;
      } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
        logger.severe(() -> "Exception: " + ex);
      } finally {
        lock.unlockWrite(stamp);
      }
    }
  }
  ...
}

让我们用两个读取器和四个写入器执行 10 次读卡和 10 次写卡：

ExecutorService readerService = Executors.newFixedThreadPool(2);
ExecutorService writerService = Executors.newFixedThreadPool(4);
for (int i = 0; i < 10; i++) {
  readerService.execute(optimisticReader);
  writerService.execute(writer);
}

可能的输出如下：

...
[12:12:07] [INFO] Increase amount with 10 by pool-2-thread-4
[12:12:07] [INFO] Read amount (read lock): 90 by pool-1-thread-2
[12:12:07] [INFO] Read amount (optimistic read): 90 by pool-1-thread-2
[12:12:07] [INFO] Increase amount with 10 by pool-2-thread-1
...

从 JDK10 开始，我们可以使用isWriteLockStamp()、isReadLockStamp()、isLockStamp()和isOptimisticReadStamp()查询戳记的类型。根据类型，我们可以决定合适的解锁方法，例如：

if (StampedLock.isReadLockStamp(stamp))
  lock.unlockRead(stamp);
}

在捆绑到本书的代码中，还有一个应用，用于举例说明tryConvertToWriteLock()方法。此外，您可能对开发使用tryConvertToReadLock()和tryConvertToOptimisticRead()的应用感兴趣。

225 死锁（哲学家聚餐）

什么是僵局？网上一个著名的笑话解释如下：

面试官：向我们解释一下，我们会雇佣你的！

我：雇佣我，我会向你解释…

简单的死锁可以解释为一个持有L锁并试图获取P 锁的A线程，同时，还有一个持有P锁并试图获取L锁的B线程。这种死锁称为循环等待。Java 没有死锁检测和解决机制（就像数据库一样），因此死锁对于应用来说非常尴尬。死锁可能会完全或部分阻塞应用，会导致严重的性能损失、奇怪的行为等等。通常情况下，死锁很难调试，解决死锁的唯一方法是重新启动应用，并希望取得最好的结果。

哲学家吃饭是一个著名的问题，用来说明僵局。这个问题说五位哲学家围坐在一张桌子旁。他们每个人轮流思考和吃饭。为了吃饭，哲学家需要双手叉子左手叉子右手叉子。困难是因为只有五个叉子。吃完后，哲学家把两个叉子放回桌上，然后由另一个重复同样循环的哲学家拿起。当一个哲学家不吃饭时，他/她在思考。下图说明了这种情况：

主要任务是找到解决这个问题的办法，让哲学家们思考和进食，以避免饿死。

在《法典》中，我们可以把每个哲学家看作一个实例。作为Runnable实例，我们可以在不同的线程中执行它们。每个哲学家都能拿起两个叉子放在他左右两侧。如果我们将叉表示为String，则可以使用以下代码：

public class Philosopher implements Runnable {
  private final String leftFork;
  private final String rightFork;
  public Philosopher(String leftFork, String rightFork) {
    this.leftFork = leftFork;
    this.rightFork = rightFork;
  }
  @Override
  public void run() {
    // implemented below
  }
}

所以，哲学家可以拿起leftFork和rightFork。但是，由于哲学家们共用这些叉子，哲学家必须在这两个叉子上获得唯一的锁。leftFork上有专用锁，且rightFork上有专用锁，等于手中有两个叉子。leftFork和rightFork上有专属锁，相当于哲学家的饮食。释放两个专属锁就等于哲学家不吃不思。

通过synchronized关键字可以实现锁定，如下run()方法：

@Override
public void run() {
  while (true) {
    logger.info(() -> Thread.currentThread().getName() 
      + ": thinking");
    doIt();
    synchronized(leftFork) {
      logger.info(() -> Thread.currentThread().getName() 
        + ": took the left fork (" + leftFork + ")");
      doIt();
      synchronized(rightFork) {
        logger.info(() -> Thread.currentThread().getName() 
          + ": took the right fork (" + rightFork + ") and eating");
        doIt();
        logger.info(() -> Thread.currentThread().getName() 
          + ": put the right fork ( " + rightFork 
          + ") on the table");
        doIt();
      }
      logger.info(() -> Thread.currentThread().getName() 
        + ": put the left fork (" + leftFork 
        + ") on the table and thinking");
      doIt();
    }
  }
}

哲学家从思考开始。过了一会儿他饿了，所以他试着拿起左叉子和右叉子。如果成功，他会吃一会儿。后来，他把叉子放在桌子上，继续思考，直到他又饿了。同时，另一个哲学家会吃东西。

doIt()方法通过随机睡眠模拟所涉及的动作（思考、进食、采摘和放叉）。代码中可以看到以下内容：

private static void doIt() {
  try {
    Thread.sleep(rnd.nextInt(2000));
  } catch (InterruptedException ex) {
    Thread.currentThread().interrupt();
    logger.severe(() -> "Exception: " + ex);
  }
}

最后，我们需要福克斯和哲学家，请参见以下代码：

String[] forks = {
  "Fork-1", "Fork-2", "Fork-3", "Fork-4", "Fork-5"
};
Philosopher[] philosophers = {
  new Philosopher(forks[0], forks[1]),
  new Philosopher(forks[1], forks[2]),
  new Philosopher(forks[2], forks[3]),
  new Philosopher(forks[3], forks[4]),
  new Philosopher(forks[4], forks[0])
};

每个哲学家都将在一个线程中运行，如下所示：

Thread threadPhilosopher1 
  = new Thread(philosophers[0], "Philosopher-1");
...
Thread threadPhilosopher5 
  = new Thread(philosophers[4], "Philosopher-5");
threadPhilosopher1.start();
...
threadPhilosopher5.start();

这个实现似乎还可以，甚至可以正常工作一段时间。但是，此实现迟早会以如下方式阻止输出：

[17:29:21] [INFO] Philosopher-5: took the left fork (Fork-5)
...
// nothing happens

这是僵局！每个哲学家都有左手叉（锁在上面），等待右手叉放在桌子上（锁要放了）。显然，这种期望是不能满足的，因为只有五个叉子，每个哲学家手里都有一个叉子。

为了避免这种死锁，有一个非常简单的解决方案。我们只是强迫其中一个哲学家先拿起正确的叉子。在成功地选择了右叉子之后，他可以试着选择左叉子。在代码中，这是对以下行的快速修改：

// the original line
new Philosopher(forks[4], forks[0])
// the modified line that eliminates the deadlock
new Philosopher(forks[0], forks[4])

这一次我们可以在没有死锁的情况下运行应用。

总结

好吧，就这些！本章讨论了 Fork/Join 框架、CompletableFuture、ReentrantLock、ReentrantReadWriteLock、StampedLock、原子变量、任务取消、可中断方法、线程局部和死锁等问题

从本章下载应用以查看结果和其他详细信息。**

Java 编程问题：十一、并发-深入探索4

221 原子变量

加法器和累加器

222 重入锁

223 重入读写锁

224 冲压锁

225 死锁（哲学家聚餐）

总结

热门文章

最新文章

相关课程

相关电子书

相关实验场景