如何在不使用同步（无锁序列计数器实现）的情况下修复竞争条件？

Question

有一个场景，其中多个线程在比较代码上有竞争条件。

private int volatile maxValue;
private AtomicInteger currentValue;

public void constructor() {
   this.current = new AtomicInteger(getNewValue());
}

public getNextValue() {
  while(true) {
     int latestValue = this.currentValue.get();
     int nextValue = latestValue + 1;
     if(latestValue == maxValue) {//Race condition 1 
       latestValue = getNewValue();
     }
    if(currentValue.compareAndSet(latestValue, nextValue) {//Race condition 2
      return latestValue;
    }
  }
}

private int getNewValue() {
    int newValue = getFromDb(); //not idempotent
    maxValue = newValue + 10;
    return newValue;
}

问题 ：

解决这个问题的显而易见的方法是在 if 条件周围添加同步块/方法。使用并发 api 而不使用任何类型的锁来解决这个问题的其他高效方法是什么？

如何摆脱 while 循环，以便我们可以在没有或更少线程争用的情况下获得下一个值？

约束：

下一个 db 序列将按递增顺序排列，不一定均匀分布。所以它可能是 1、11、31，其中 21 可能是其他节点询问的。请求的下一个值将始终是唯一的。还需要确保所有序列都被使用，一旦我们达到前一个范围的最大值，则只向 db 请求另一个起始序列，依此类推。

例子 ：

对于增量为 10 的 db next 序列 1,11,31，对于 30 个请求，输出的 next 序列应为 1-10、11-20、31-40。

Answer 1

首先：我建议再考虑一次使用synchronized，因为：

1. 看看这样的代码有多简单：

    private int maxValue;
    private int currentValue;
   
    public constructor() {
      requestNextValue();
    }
   
    public synchronized int getNextValue() {
      currentValue += 1;
      if (currentValue == maxValue) {
        requestNextValue();
      }
      return currentValue;
    }
   
    private void requestNextValue() {
      currentValue = getFromDb(); //not idempotent
      maxValue = currentValue + 10;
    }
   

2. java中的锁实际上非常智能并且具有非常好的性能。
3. 您在代码中与 DB 交谈——仅此一项的性能成本可能比锁的性能成本高出几个数量级。

但总的来说，您的竞争条件是因为您独立更新maxValue而发生的currentValue。
您可以将这 2 个值组合成一个不可变对象，然后以原子方式使用该对象：

private final AtomicReference<State> stateHolder = new AtomicReference<>(newStateFromDb());

public int getNextValue() {
  while (true) {
    State oldState = stateHolder.get();
    State newState = (oldState.currentValue == oldState.maxValue)
        ? newStateFromDb()
        : new State(oldState.currentValue + 1, oldState.maxValue);
    if (stateHolder.compareAndSet(oldState, newState)) {
      return newState.currentValue;
    }
  }
}

private static State newStateFromDb() {
  int newValue = getFromDb(); // not idempotent
  return new State(newValue, newValue + 10);
}


private static class State {

  final int currentValue;
  final int maxValue;

  State(int currentValue, int maxValue) {
    this.currentValue = currentValue;
    this.maxValue = maxValue;
  }
}

修复后，您接下来可能需要解决以下问题：

• 如何防止多个并行getFromDb();（特别是在考虑到该方法是幂等的之后）
• 当一个线程执行时getFromDb();，如何防止其他线程在while(true)循环内忙于旋转并消耗所有可用的cpu时间
• 更多类似问题

解决这些问题中的每一个都可能会使您的代码变得越来越复杂。

所以，恕我直言，这几乎是不值得的——锁可以正常工作并保持代码简单。

Answer 2

解决这个问题的明显方法是在 if 条件周围添加同步块

那是行不通的。让我尝试解释一下。

当您满足条件： 时if(latestValue == maxValue) { ... }，您希望以原子方式更新maxValue和currentValue。像这样的东西：

latestValue = getNewValue();
currentValue.set(latestValue);

getNewValueDB将从和 update获取下一个起始值maxValue，但同时，您currentValue现在想要设置为新的起始值。假设这样的情况：

• 1您首先从数据库中读取。像这样maxValue = 11，currentValue = 1。

• 当你达到条件 时if(latestValue == maxValue)，你想要去数据库获取新的起始位置（比方说21），但同时你希望每个线程现在从 开始21。所以你还必须设置currentValue.

现在的问题是，如果您在同步块下写入currentValue，例如：

if(latestValue == maxValue) {
   synchronized (lock) {
       latestValue = getNewValue();
       currentValue.set(latestValue);
   }
}

你也需要在相同的情况下阅读lock，否则你就会有种族。最初我认为我可以更聪明一点，做一些类似的事情：

if(latestValue == maxValue) {
    synchronized (lock) {
       if(latestValue == maxValue) {
           latestValue = getNewValue();
           currentValue.set(latestValue);
       } else {
          continue;
       }
    }
}

这样，当锁被释放时，等待 a 的所有线程lock都不会覆盖先前写入的值。maxValue但这仍然是一个问题race，并且会在其他地方引起问题，在不同的情况下，相当微不足道。例如：

• ThreadA确实latestValue = getNewValue();如此maxValue == 21。在它发生之前currentValue.set(latestValue);

• ThreadB读取int latestValue = this.currentValue.get();、查看11，当然这将是 false: if(latestValue == maxValue) {，因此它可以写入12( nextValue) 到currentValue. 这破坏了整个算法。

我没有看到任何其他方法来制作getNextValue synchronized互斥锁/自旋锁或以其他方式受到互斥锁/自旋锁的保护。