java - ConcurrentHashMap 与 AtomicReference>

Question

选择什么更好？为什么？该 map 用作临时存储。它将项目保留一段时间，然后刷新到数据库。

这是我使用原子引用实现的虚拟代码:

public class Service {
    private final AtomicReference<Map<String, Entity>> storedEntities = new AtomicReference<>(new HashMap<>());
    private final AtomicReference<Map<String, Entity>> newEntities = new AtomicReference<>(new HashMap<>());

    private final Dao dao;

    public Service(Dao dao) {
        this.dao = dao;
    }

    @Transactional
    @Async
    public CompletableFuture<Void> save() {
        Map<String, Entity> map = newEntities.getAndSet(new HashMap<>());
        return dao.saveAsync(map.values());
    }

    @Transactional(readOnly = true)
    @Async
    public CompletableFuture<Map<String, Entity>> readAll() {
        return dao.getAllAsync().thenApply(map -> {
            storedEntities.set(map);
            return map;
        });
    }

    @Scheduled(cron = "${cron}")
    public void refreshNow() {
        save();
        readAll();
    }

    public void addNewentity(Entity entity) {
        newEntities.getAndUpdate(map -> {
            map.put(entity.getHash(), entity);
            return map;
        });
    }

    public AtomicReference<List<Entity>> getStoredEntities() {
        return storedEntities.get().values();
    }

    public AtomicReference<List<Entity>> getNewEntities() {
        return newEntities.get().values();
    }
}

正如我所说，我只需要保留数据一段时间，然后通过 cron 将其刷新到数据库。我对什么是更好的方法感兴趣 - AR 与 CHM？

Answer 1

我首先假设您需要跨多个线程共享访问某些资源(临时存储映射)。

简短回答:

使用ConcurrentHashMap。

长(呃)答案:

如果您的期望是通过使用 AtomicReference，您将收到一致或线程安全的 map View ，或者您的操作将以原子方式执行，那么您很可能是不正确的 - 但是，因为您没有提供您的用法示例，我不能完全肯定地说。

虽然性能不容忽视，但它不应该是您关心的主要问题 - 相反，您应该确保程序能够正确且按预期执行，然后寻求改进其性能。

如果您有多个线程读取和/或写入临时存储映射，则映射的内部状态保持一致和正确非常重要；您可以通过以atomic的方式实现您的操作来实现这一点。和线程安全，从这个意义上说，两者都是 ConcurrentHashMap或由 SynchronizedMap 包裹的 map 将实现这一目标。然而，上述每种方法都以不同的粒度实现了确保这一点的方法，并且由于 ConcurrentHashMap 采取了专门的(乐观)方法，与包装映射的朴素(悲观)方法相反，ConcurrentHashMap 较少容易受到资源争用的影响，并且可能是两者中性能更高的一个。

继续前进

我提供了下面这篇文章中讨论的三种机制的实现；除了 Java API 文档之外，还可以探索下面的代码。

import java.io.PrintStream;
import java.text.DecimalFormat;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
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.AtomicReference;
import java.util.function.BiFunction;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class ConcurrencyExample {

    interface TemporaryStorage<K, V> {

        V compute(K key, BiFunction<? super K, ? super V, ? extends V> remapper);

        V put(K key, V value);

        V get(K key);

        void clear();

        @FunctionalInterface
        interface UnitTest<K, V> {

            void test(TemporaryStorage<K, V> store, K key);

        }

    }

    static class ConcurrentHashMapTS<K, V> implements TemporaryStorage<K, V> {

        private final Map<K, V> map = new ConcurrentHashMap<>();

        @Override
        public V compute(K key, BiFunction<? super K, ? super V, ? extends V> remapper) {
            return map.compute(key, remapper);
        }

        @Override
        public V put(K key, V value) {
            return map.put(key, value);
        }

        @Override
        public V get(K key) {
            return map.get(key);
        }

        @Override
        public void clear() {
            map.clear();
        }
    }

    static class AtomicReferenceHashMapTS<K, V> implements TemporaryStorage<K, V> {

        private final AtomicReference<Map<K, V>> map = new AtomicReference<>(new HashMap<>());

        @Override
        public V compute(K key, BiFunction<? super K, ? super V, ? extends V> remapper) {
            return map.get().compute(key, remapper);
        }

        @Override
        public V put(K key, V value) {
            return map.get().put(key, value);
        }

        @Override
        public V get(K key) {
            return map.get().get(key);
        }

        @Override
        public void clear() {
            map.get().clear();
        }
    }

    static class MonitorLockedHashMapTS<K, V> implements TemporaryStorage<K, V> {

        private final Map<K, V> map = new HashMap<>();
        private final Object mutex = new Object(); // could use the map as the mutex

        @Override
        public V compute(K key, BiFunction<? super K, ? super V, ? extends V> remapper) {
            synchronized (mutex) {
                return map.compute(key, remapper);
            }
        }

        @Override
        public V put(K key, V value) {
            synchronized (mutex) {
                return map.put(key, value);
            }
        }

        @Override
        public V get(K key) {
            synchronized (mutex) {
                return map.get(key);
            }
        }

        @Override
        public void clear() {
            synchronized (mutex) {
                map.clear();
            }
        }
    }

    static class WrappedHashMapTS<K, V> implements TemporaryStorage<K, V> {

        private final Map<K, V> map = Collections.synchronizedMap(new HashMap<>());

        @Override
        public V compute(K key, BiFunction<? super K, ? super V, ? extends V> remapper) {
            return map.compute(key, remapper);
        }

        @Override
        public V put(K key, V value) {
            return map.put(key, value);
        }

        @Override
        public V get(K key) {
            return map.get(key);
        }

        @Override
        public void clear() {
            map.clear();
        }
    }

    static class AtomicUnitTest implements TemporaryStorage.UnitTest<Integer, Integer> {

        @Override
        public void test(TemporaryStorage<Integer, Integer> store, Integer key) {
            store.compute(key, (k, v) -> (v == null ? 0 : v) + 1);
        }
    }

    static class UnsafeUnitTest implements TemporaryStorage.UnitTest<Integer, Integer> {

        @Override
        public void test(TemporaryStorage<Integer, Integer> store, Integer key) {
            Integer value = store.get(key);
            store.put(key, (value == null ? 0 : value) + 1);
        }
    }

    public static class TestRunner {

        public static void main(String... args) throws InterruptedException {
            final int iterations = 1_000;
            final List<Integer> keys = IntStream.rangeClosed(1, iterations).boxed().collect(Collectors.toList());

            final int expected = iterations;

            for (int batch = 1; batch <= 5; batch++) {

                System.out.println(String.format("--- START BATCH %d ---", batch));

                test(System.out, new ConcurrentHashMapTS<>(), new AtomicUnitTest(), keys, expected, iterations);
                test(System.out, new ConcurrentHashMapTS<>(), new UnsafeUnitTest(), keys, expected, iterations);

                test(System.out, new AtomicReferenceHashMapTS<>(), new AtomicUnitTest(), keys, expected, iterations);
                test(System.out, new AtomicReferenceHashMapTS<>(), new UnsafeUnitTest(), keys, expected, iterations);

                test(System.out, new MonitorLockedHashMapTS<>(), new AtomicUnitTest(), keys, expected, iterations);
                test(System.out, new MonitorLockedHashMapTS<>(), new UnsafeUnitTest(), keys, expected, iterations);

                test(System.out, new WrappedHashMapTS<>(), new AtomicUnitTest(), keys, expected, iterations);
                test(System.out, new WrappedHashMapTS<>(), new UnsafeUnitTest(), keys, expected, iterations);

                System.out.println(String.format("--- END   BATCH %d ---", batch));

                System.out.println();

            }
        }

        private static <K, V> void test(PrintStream printer, TemporaryStorage<K, V> store, TemporaryStorage.UnitTest<K, V> work, List<K> keys, V expected, int iterations) throws InterruptedException {
            test(printer, store, work, keys, expected, iterations, Runtime.getRuntime().availableProcessors() * 4);
        }

        private static <K, V> void test(PrintStream printer, TemporaryStorage<K, V> store, TemporaryStorage.UnitTest<K, V> work, List<K> keys, V expected, int iterations, int parallelism) throws InterruptedException {
            final ExecutorService workers = Executors.newFixedThreadPool(parallelism);
            final long start = System.currentTimeMillis();

            for (K key : keys) {
                for (int iteration = 1; iteration <= iterations; iteration++) {

                    workers.execute(() -> {
                        try {
                            work.test(store, key);
                        } catch (Exception e) {
                            //e.printStackTrace(); //thrown by the AtomicReference<Map<K, V>> implementation
                        }
                    });

                }
            }

            workers.shutdown();
            workers.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);

            final long finish = System.currentTimeMillis();

            final DecimalFormat formatter = new DecimalFormat("###,###");

            final long correct = keys.stream().filter(key -> expected.equals(store.get(key))).count();

            printer.println(String.format("Store '%s' performed %s iterations of %s across %s threads in %sms. Accuracy: %d / %d (%4.2f percent)", store.getClass().getSimpleName(), formatter.format(iterations), work.getClass().getSimpleName(), formatter.format(parallelism), formatter.format(finish - start), correct, keys.size(), ((double) correct / keys.size()) * 100));
        }

    }

}