ConcurrentHashMap：避免使用“putIfAbsent”创建额外的对象？

Question

我正在为多线程环境中的键聚合多个值。密钥是事先不知道的。我以为我会做这样的事情：

class Aggregator {
    protected ConcurrentHashMap<String, List<String>> entries =
                            new ConcurrentHashMap<String, List<String>>();
    public Aggregator() {}

    public void record(String key, String value) {
        List<String> newList =
                    Collections.synchronizedList(new ArrayList<String>());
        List<String> existingList = entries.putIfAbsent(key, newList);
        List<String> values = existingList == null ? newList : existingList;
        values.add(value);
    }
}

我看到的问题是，每次运行此方法时，我都需要创建一个 ArrayList 的新实例，然后将其丢弃（在大多数情况下）。这似乎是对垃圾收集器的不合理滥用。有没有更好的、线程安全的方法来初始化这种结构而不必synchronizerecord 方法？我对让 putIfAbsent 方法不返回新创建的元素的决定感到有些惊讶，而且除非需要（可以这么说），否则缺乏延迟实例化的方法。

Answer 1

Java 8 引入了一个 API 来解决这个确切的问题，提供了一个单行解决方案：

public void record(String key, String value) {
    entries.computeIfAbsent(key, k -> Collections.synchronizedList(new ArrayList<String>())).add(value);
}

对于 Java 7：

public void record(String key, String value) {
    List<String> values = entries.get(key);
    if (values == null) {
        entries.putIfAbsent(key, Collections.synchronizedList(new ArrayList<String>()));
        // At this point, there will definitely be a list for the key.
        // We don't know or care which thread's new object is in there, so:
        values = entries.get(key);
    }
    values.add(value);
}

这是填充 ConcurrentHashMap 时的标准代码模式。

特殊方法putIfAbsent(K, V)) 将把你的值对象放入，或者如果另一个线程在你之前，那么它将忽略你的值对象。无论哪种方式，在调用putIfAbsent(K, V)) 之后，get(key) 保证在线程之间是一致的，因此上面的代码是线程安全的。

唯一浪费的开销是，如果某个其他线程同时为同一个键添加了一个新条目：您可能最终会丢弃新创建的值，但这只有在有还不是一个条目并且你的线程输掉了一场比赛，这通常很少见。

Answer 2

从 Java-8 开始，您可以使用以下模式创建多地图：

public void record(String key, String value) { entries.computeIfAbsent(key, k -> Collections.synchronizedList(new ArrayList<String>())) .add(value); }

ConcurrentHashMap 文档（不是通用合同）指定 ArrayList 只会为每个键创建一次，在为新键创建 ArrayList 时会延迟更新，这会产生轻微的初始成本：

http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ConcurrentHashMap.html#computeIfAbsent-K-java.util.function.Function-

Answer 3

最后，我对@Bohemian 的回答进行了轻微修改。他提出的解决方案用putIfAbsent 调用覆盖了values 变量，这产生了我之前遇到的同样问题。似乎工作的代码如下所示：

    public void record(String key, String value) {
        List<String> values = entries.get(key);
        if (values == null) {
            values = Collections.synchronizedList(new ArrayList<String>());
            List<String> values2 = entries.putIfAbsent(key, values);
            if (values2 != null)
                values = values2;
        }
        values.add(value);
    }

它没有我想要的那么优雅，但它比在每次调用时创建一个新的ArrayList 实例的原始版本要好。

Answer 4

根据 Gene 的回答创建了两个版本

public  static <K,V> void putIfAbsetMultiValue(ConcurrentHashMap<K,List<V>> entries, K key, V value) {
    List<V> values = entries.get(key);
    if (values == null) {
        values = Collections.synchronizedList(new ArrayList<V>());
        List<V> values2 = entries.putIfAbsent(key, values);
        if (values2 != null)
            values = values2;
    }
    values.add(value);
}

public  static <K,V> void putIfAbsetMultiValueSet(ConcurrentMap<K,Set<V>> entries, K key, V value) {
    Set<V> values = entries.get(key);
    if (values == null) {
        values = Collections.synchronizedSet(new HashSet<V>());
        Set<V> values2 = entries.putIfAbsent(key, values);
        if (values2 != null)
            values = values2;
    }
    values.add(value);
}

效果很好

Answer 5

这是一个我也在寻找答案的问题。 putIfAbsent 方法实际上并没有解决额外的对象创建问题，它只是确保这些对象中的一个不会替换另一个。但是线程之间的竞争条件会导致多个对象实例化。我可以为这个问题找到 3 个解决方案（我会按照这个优先顺序）：

1- 如果您使用的是 Java 8，实现此目的的最佳方法可能是 ConcurrentMap 的新 computeIfAbsent 方法。您只需要给它一个将同步执行的计算函数（至少对于ConcurrentHashMap 实现）。示例：

private final ConcurrentMap<String, List<String>> entries =
        new ConcurrentHashMap<String, List<String>>();

public void method1(String key, String value) {
    entries.computeIfAbsent(key, s -> new ArrayList<String>())
            .add(value);
}

这是来自ConcurrentHashMap.computeIfAbsent的javadoc：

如果指定的键尚未与值关联，则尝试 使用给定的映射函数计算其值并输入 进入这张地图，除非为空。执行整个方法调用 原子地，因此每个键最多应用该函数一次。一些 其他线程在此地图上尝试的更新操作可能是 在计算过程中被阻塞，所以计算应该是 简短而简单，并且不得尝试更新任何其他映射 这张地图。

2- 如果你不能使用 Java 8，你可以使用Guava 的LoadingCache，它是线程安全的。您为它定义了一个加载函数（就像上面的compute 函数一样），您可以确定它会被同步调用。示例：

private final LoadingCache<String, List<String>> entries = CacheBuilder.newBuilder()
        .build(new CacheLoader<String, List<String>>() {
            @Override
            public List<String> load(String s) throws Exception {
                return new ArrayList<String>();
            }
        });

public void method2(String key, String value) {
    entries.getUnchecked(key).add(value);
}

3- 如果您也不能使用 Guava，您可以随时手动同步并进行双重检查锁定。示例：

private final ConcurrentMap<String, List<String>> entries =
        new ConcurrentHashMap<String, List<String>>();

public void method3(String key, String value) {
    List<String> existing = entries.get(key);
    if (existing != null) {
        existing.add(value);
    } else {
        synchronized (entries) {
            List<String> existingSynchronized = entries.get(key);
            if (existingSynchronized != null) {
                existingSynchronized.add(value);
            } else {
                List<String> newList = new ArrayList<>();
                newList.add(value);
                entries.put(key, newList);
            }
        }
    }
}

我对所有这 3 个方法以及非同步方法做了一个示例实现，这会导致额外的对象创建：http://pastebin.com/qZ4DUjTr

Answer 6

Java 1.7.40 处理空数组列表创建问题的内存浪费（还有 GC 等）。不要担心创建空数组列表。 参考：http://javarevisited.blogspot.com.tr/2014/07/java-optimization-empty-arraylist-and-Hashmap-cost-less-memory-jdk-17040-update.html

Answer 7

putIfAbsent 的方法具有最快的执行时间，在竞争激烈的环境中，它比“lambda”方法快 2 到 50 倍。 Lambda 不是这种“powerloss”背后的原因，问题是在 Java-9 优化之前computeIfAbsent 内部的强制同步。

基准：

import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public class ConcurrentHashMapTest {
    private final static int numberOfRuns = 1000000;
    private final static int numberOfThreads = Runtime.getRuntime().availableProcessors();
    private final static int keysSize = 10;
    private final static String[] strings = new String[keysSize];
    static {
        for (int n = 0; n < keysSize; n++) {
            strings[n] = "" + (char) ('A' + n);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        for (int n = 0; n < 20; n++) {
            testPutIfAbsent();
            testComputeIfAbsentLamda();
        }
    }

    private static void testPutIfAbsent() throws InterruptedException {
        final AtomicLong totalTime = new AtomicLong();
        final ConcurrentHashMap<String, AtomicInteger> map = new ConcurrentHashMap<String, AtomicInteger>();
        final Random random = new Random();
        ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);

        for (int i = 0; i < numberOfThreads; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    long start, end;
                    for (int n = 0; n < numberOfRuns; n++) {
                        String s = strings[random.nextInt(strings.length)];
                        start = System.nanoTime();

                        AtomicInteger count = map.get(s);
                        if (count == null) {
                            count = new AtomicInteger(0);
                            AtomicInteger prevCount = map.putIfAbsent(s, count);
                            if (prevCount != null) {
                                count = prevCount;
                            }
                        }
                        count.incrementAndGet();
                        end = System.nanoTime();
                        totalTime.addAndGet(end - start);
                    }
                }
            });
        }
        executorService.shutdown();
        executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
        System.out.println("Test " + Thread.currentThread().getStackTrace()[1].getMethodName()
                + " average time per run: " + (double) totalTime.get() / numberOfThreads / numberOfRuns + " ns");
    }

    private static void testComputeIfAbsentLamda() throws InterruptedException {
        final AtomicLong totalTime = new AtomicLong();
        final ConcurrentHashMap<String, AtomicInteger> map = new ConcurrentHashMap<String, AtomicInteger>();
        final Random random = new Random();
        ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
        for (int i = 0; i < numberOfThreads; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    long start, end;
                    for (int n = 0; n < numberOfRuns; n++) {
                        String s = strings[random.nextInt(strings.length)];
                        start = System.nanoTime();

                        AtomicInteger count = map.computeIfAbsent(s, (k) -> new AtomicInteger(0));
                        count.incrementAndGet();

                        end = System.nanoTime();
                        totalTime.addAndGet(end - start);
                    }
                }
            });
        }
        executorService.shutdown();
        executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
        System.out.println("Test " + Thread.currentThread().getStackTrace()[1].getMethodName()
                + " average time per run: " + (double) totalTime.get() / numberOfThreads / numberOfRuns + " ns");
    }

}

结果：

Test testPutIfAbsent average time per run: 115.756501 ns
Test testComputeIfAbsentLamda average time per run: 276.9667055 ns
Test testPutIfAbsent average time per run: 134.2332435 ns
Test testComputeIfAbsentLamda average time per run: 223.222063625 ns
Test testPutIfAbsent average time per run: 119.968893625 ns
Test testComputeIfAbsentLamda average time per run: 216.707419875 ns
Test testPutIfAbsent average time per run: 116.173902375 ns
Test testComputeIfAbsentLamda average time per run: 215.632467375 ns
Test testPutIfAbsent average time per run: 112.21422775 ns
Test testComputeIfAbsentLamda average time per run: 210.29563725 ns
Test testPutIfAbsent average time per run: 120.50643475 ns
Test testComputeIfAbsentLamda average time per run: 200.79536475 ns