使用Java 8并发性对变量进行高吞吐量写入？

Question

如果我在Java 8程序中有一个简单的Integer，可以被多个线程读取和写。

如果我被告知这个应用程序需要支持高吞吐量的读取和很少的写-答案是非常简单的，我只是使用读写锁。然后，多个线程可以同时执行读取而没有阻塞-而且只有在不频繁写入时才会发生阻塞。

但在这种情况下，如果我被告知应用程序需要支持高吞吐量写入(即共享变量被不同线程频繁更新)。无论我在这里使用什么样的锁，据我所见，它总是会导致线程阻塞--当线程获得变量上的锁并更新它时，其他试图更新变量的线程将不得不等待直到它们获得锁--这是正确的，还是Java 8中遗漏了什么？

我可以在共享变量上编写某种异步更新方法，其中线程调用它立即返回的update方法，并且我在幕后使用某种数据结构来对对共享变量的写入进行排队。至少这样，在尝试更新共享变量时，我可以防止线程阻塞。当然，这种方法还会引发其他问题，例如，如果线程假定有其保证的写def。成功或我应该提供一个回传通知更新成功等。除了这样的东西，我没有办法绕过阻塞时，使用任何锁在Java 8的高通量写入？(或者我是否应该接受阻塞，并且无论如何只使用Lock，即使是在高吞吐量的情况下也是如此)。谢谢

Answer 1

严格地说，Integer -您可以使用LongAdder，它的实现完全适合您的情况。如果你关心的话，这里有一些额外的细节。

它在引擎盖下使用CAS (比较和交换)，很像AtomicLong，但有一些不同之处。首先，它持有的实际long value被包装在一个所谓的Cell中--基本上是一个类--允许cas (比较和交换) value到一个新值，如果您想要的话，就像一个setter。为了防止错误共享，还对此Cell进行了@sun.misc.Contended注释；以下是对它的解释(来自代码注释)：

但是，驻留在数组中的原子对象往往被放置在相邻的位置，因此在没有这种预防措施的情况下，大多数情况下都会共享缓存行(具有巨大的负面性能影响)。

从这里开始，实现非常有趣。让我们看看当您调用add(long x)方法时会发生什么：

 public void add(long x) {
    Cell[] cs; long b, v; int m; Cell c;
    if ((cs = cells) != null || !casBase(b = base, b + x)) {
        boolean uncontended = true;
        if (cs == null || (m = cs.length - 1) < 0 ||
            (c = cs[getProbe() & m]) == null ||
            !(uncontended = c.cas(v = c.value, v + x)))
            longAccumulate(x, null, uncontended);
    }
}

这样做的想法是，如果Cell [] cs为null，则以前没有争用，这意味着long value要么未初始化，要么所有以前的CAS操作都已由所有线程成功完成。在本例中，尝试将新值CAS为long value --如果效果良好，我们就完成了。但是，如果失败，就会创建一个Cell []数组，以便每个单独的线程都能在自己的空间中工作，从而最小化争用。

下一句是如果我正确理解了你的问题，你真正关心的是什么(这是我的问题，根本不来自代码注释)：

简单地说:如果线程之间没有争用，那么工作就会像使用AtomicLong一样完成(某种程度上)，否则就会为每个线程创建一个单独的空间。

如果你关心一些我觉得有趣的其他细节：

Cell[]总是由两个(很像HashMap内部数组)组成；然后每个线程都使用ThreadLocalRandom来创建一些hashCode，试图在数组Cell [] cs中找到要写入的条目，甚至重新散列，使用Marsaglia XorShif来尝试在这个数组中找到一个空闲的插槽；数组的大小被限制为您拥有的核数(实际上是两个内核的最接近的功率)，可以调整数组的大小，这样它就可以增长，所有这些操作都是使用volatile int cellsBusy自旋锁完成的。这个代码非常好，但是正如前面所说的，我并不完全理解它。