在低锁竞争的情况下，用 atomic+while 循环替换同步

Question

我有两个必须在临界区运行的函数：

public synchronized void f1() { ... }
public synchronized void f2() { ... }

假设行为如下：

• f1 几乎从不被调用。实际上，在正常情况下，这个方法永远不会被调用。如果无论如何调用f1，它应该很快返回。
• f2 的调用率非常高。它很快就会返回。
• 这些方法从不相互调用，也没有重入。

换句话说，竞争非常低。所以当f2 被调用时，我们有一些开销来获取锁，在 99.9% 的情况下会立即授予。我想知道是否有办法避免这种开销。

我想出了以下替代方案：

private final AtomicInteger lock = new AtomicInteger(0);

public void f1() {
    while (!lock.compareAndSet(0, 1)) {}

    try {
        ...
    } finally {
        lock.set(0);
    }
}

public void f2() {
    while (!lock.compareAndSet(0, 2)) {}

    try {
        ...
    } finally {
        lock.set(0);
    }
}

还有其他方法吗？ java.util.concurrent 包本身是否提供了一些东西？

更新

虽然我的意图是提出一个笼统的问题，但关于我的情况的一些信息：

f1：如果由于某种原因当前的远程流损坏，例如由于超时，此方法会创建一个新的远程流。远程流可以被认为是一个套接字连接，它消耗从给定位置开始的远程队列：

private Stream stream;

public synchronized void f1() {
     final Stream stream = new Stream(...);

     if (this.stream != null) {
         stream.setPosition(this.stream.getPosition());
     }
     this.stream = stream;
     return stream;
}

f2：此方法推进流位置。这是一个简单的设置器：

public synchronized void f2(Long p) {
    stream.setPosition(p);
}

这里，stream.setPosition(Long) 也被实现为一个普通的 setter：

public class Stream {
    private volatile Long position = 0;

    public void setPosition(Long position) {
        this.position = position;
    }
}

在Stream中，当前位置会定期异步发送到服务器。注意Stream不是我自己实现的。

我的想法是引入如上图所示的比较和交换，并将stream标记为volatile。

Answer 1

你的例子没有做你想做的事。当锁 is 被使用时，您实际上正在执行您的代码。试试这样的：

public void f1() {
    while (!lock.compareAndSet(0, 1)) {
    }

    try {
        ...
    } finally {
        lock.set(0);
    }
}

为了回答你的问题，我认为这不会比使用synchronized 方法更快，而且这种方法更难阅读和理解。

Answer 2

根据描述和您的示例代码，我推断出以下内容：

1. Stream 有自己的内部位置，并且您还在外部跟踪最新的position。您将其用作一种“恢复点”：当您需要重新初始化流时，将其推进到该点。

2. 最后一个已知的position 可能已过时；我假设这是基于您的断言，即流会定期异步通知服务器其当前位置。

3. 在调用f1 时，已知流处于错误状态。

4. 函数f1 和f2 访问相同的数据，并且可以同时运行。但是，f1 和 f2 都不会同时针对自身运行。换句话说，您几乎拥有一个单线程程序，除了f1 和f2 都在执行的极少数情况。

   [旁注：我的解决方案实际上并不关心f1 是否与其自身同时被调用；它只关心f2 不会与其自身同时调用]

如果其中任何一个是错误的，那么下面的解决方案就是错误的。哎呀，无论如何它可能是错误的，要么是因为遗漏了一些细节，要么是因为我犯了一个错误。编写低锁定代码很困难，这正是您应该避免使用它的原因，除非您观察到实际的性能问题。

static class Stream {
    private long position = 0L;

    void setPosition(long position) {
        this.position = position;
    }
}

final static class StreamInfo {
    final Stream stream = new Stream();
    volatile long resumePosition = -1;

    final void setPosition(final long position) {
        stream.setPosition(position);
        resumePosition = position;
    }
}

private final Object updateLock = new Object();
private final AtomicReference<StreamInfo> currentInfo = new AtomicReference<>(new StreamInfo());

void f1() {
    synchronized (updateLock) {
        final StreamInfo oldInfo = currentInfo.getAndSet(null);
        final StreamInfo newInfo = new StreamInfo();

        if (oldInfo != null && oldInfo.resumePosition > 0L) {
            newInfo.setPosition(oldInfo.resumePosition);
        }

        // Only `f2` can modify `currentInfo`, so update it last.
        currentInfo.set(newInfo);

        // The `f2` thread might be waiting for us, so wake them up.
        updateLock.notifyAll();
    }
}

void f2(final long newPosition) {
    while (true) {
        final StreamInfo s = acquireStream();

        s.setPosition(newPosition);
        s.resumePosition = newPosition;

        // Make sure the stream wasn't replaced while we worked.
        // If it was, run again with the new stream.
        if (acquireStream() == s) {
            break;
        }
    }
}

private StreamInfo acquireStream() {
    // Optimistic concurrency: hope we get a stream that's ready to go.
    // If we fail, branch off into a slower code path that waits for it.
    final StreamInfo s = currentInfo.get();
    return s != null ? s : acquireStreamSlow();
}

private StreamInfo acquireStreamSlow() {
    synchronized (updateLock) {
        while (true) {
            final StreamInfo s = currentInfo.get();

            if (s != null) {
                return s;
            }

            try {
                updateLock.wait();
            }
            catch (final InterruptedException ignored) {
            }
        }
    }
}

如果流发生故障并被f1 替换，则之前对f2 的调用可能仍在（现已失效的）流上执行某些操作。我假设这没问题，并且它不会引入不良的副作用（除了那些已经存在于基于锁的版本中的副作用）。我做出这个假设是因为我们已经在上面的列表中确定了您的恢复点可能已过时，并且我们还确定了 f1 仅在已知流处于错误状态时才被调用。

根据我的 JMH 基准，这种方法比 CAS 或同步版本（它们本身非常接近）快大约 3 倍。

Answer 3

另一种方法是使用时间戳锁，其作用类似于修改计数。如果您具有较高的读写比率，这将非常有效。

另一种方法是拥有一个通过 AtomicReference 存储状态的不可变对象。如果您有非常高的读写比率，这很有效。