Atomic Variables and ConcurrentMap

悠悠 2022-05-14 06:38 536阅读 0赞

### AtomicInteger ###

`java.concurrent.atomic`包中包含了许多可以执行原子操作的类，所谓的原子操作是指在多线程并发的情况下无需使用`synchronized`或者其他锁同步机制的情况下，仍旧可以正确执行的操作，我们就称之为原子操作。  
在JVM内部实现上，这些类通过使用[compare-and-swap][] (CAS)来实现原子操作，CAS是CPU内部的一种原子指令，执行速度远比通过锁同步机制快的多。所以，在一般的多线程编程中，推荐使用原子类。我们看一个简单的例子：`AtomicInteger`

AtomicInteger atomicInt = new AtomicInteger(0);
    
    ExecutorService executor = Executors.newFixedThreadPool(2);
    
    IntStream.range(0, 1000)
        .forEach(i -> executor.submit(atomicInt::incrementAndGet));
    
    stop(executor);
    
    System.out.println(atomicInt.get()); // => 1000

其中，[ stop ][stop]方法定义如下：

public static void stop(ExecutorService executor) {
            try {
                executor.shutdown();
                executor.awaitTermination(60, TimeUnit.SECONDS);
            }
            catch (InterruptedException e) {
                System.err.println("termination interrupted");
            }
            finally {
                if (!executor.isTerminated()) {
                    System.err.println("killing non-finished tasks");
                }
                executor.shutdownNow();
            }
        }

注意：不要直接`shutdown`，更多可参考：[multithreading-with-atomicinteger-not-working][]

> AtomicInteger is thread safe, but you have called AtomicInteger\#get before all tasks finished.  
>   
> ExecutorService\#shutdown is not waiting for tasks to finish.

See [ExecutorService\#shutdown docs][ExecutorService_shutdown docs]:

> This method does not wait for previously submitted tasks to complete execution. Use awaitTermination to do that.

Use

service.awaitTermination(10, TimeUnit.SECONDS)

to wait for all tasks finished

`AtomicInteger`有很多原子操作，`updateAndGet()`可以接收一个lambda表达式来执行整形数值上的任意表达式运算。

AtomicInteger atomicInt = new AtomicInteger(0);
    
    ExecutorService executor = Executors.newFixedThreadPool(2);
    
    IntStream.range(0, 1000)
        .forEach(i -> {
            Runnable task = () ->
                atomicInt.updateAndGet(n -> n + 2);
            executor.submit(task);
        });
    
    stop(executor);
    
    System.out.println(atomicInt.get()); // => 2000

`accumulateAndGet()`可以接收`IntBinaryOperator`类型的lambda表达式。

AtomicInteger atomicInt = new AtomicInteger(0);
    
    ExecutorService executor = Executors.newFixedThreadPool(2);
    
    IntStream.range(0, 1000)
        .forEach(i -> {
            Runnable task = () ->
                atomicInt.accumulateAndGet(i, (n, m) -> n + m); executor.submit(task); }); stop(executor); System.out.println(atomicInt.get()); // => 499500

### LongAdder ###

与`AtomicLong`一样，`LongAdder`可以用来求和

ExecutorService executor = Executors.newFixedThreadPool(2);
    
    IntStream.range(0, 1000)
        .forEach(i -> executor.submit(adder::increment));
    
    stop(executor);
    
    System.out.println(adder.sumThenReset()); // => 1000

### LongAccumulator ###

`LongAccumulator`是一个更加通用的`LongAdder`,相比于`LongAdder`仅支持简单的长整型加法操作，`LongAccumulator`可以支持`LongBinaryOperator`类型的lambda表达式。比如下面的示例：

LongBinaryOperator op = (x, y) -> 2 * x + y;
    LongAccumulator accumulator = new LongAccumulator(op, 1L);
    
    ExecutorService executor = Executors.newFixedThreadPool(2);
    
    IntStream.range(0, 10)
        .forEach(i -> executor.submit(() -> accumulator.accumulate(i))); stop(executor); System.out.println(accumulator.getThenReset()); // => 2539

### ConcurrentMap and ConcurrentHashMap ###

`ConcurrentMap`继承实现了`Map`接口，定义了常用并发集合类型。在Java 8中，接口新增了许多方法以支持函数式编程。

ConcurrentMap<String, String> map = new ConcurrentHashMap<>();
    map.put("foo", "bar");
    map.put("han", "solo");
    map.put("r2", "d2");
    map.put("c3", "p0");

`forEach`方法接收`BiConsumer`类型的lambda表达式，其将map中的键值对作为参数传入给`BiConsumer`。`forEach()`可以作为迭代遍历`map`的方法替代。

map.forEach((key, value) -> System.out.printf("%s = %s\n", key, value));

*  putIfAbsent()

String value = map.putIfAbsent("c3", "p1");
    System.out.println(value);    // p0

*  getOrDefault()

String value = map.getOrDefault("hi", "there");
    System.out.println(value);    // there

*  replaceAll()

map.replaceAll((key, value) -> "r2".equals(key) ? "d3" : value); System.out.println(map.get("r2")); // d3

*  compute

map.compute("foo", (key, value) -> value + value);
    System.out.println(map.get("foo"));   // barbar

*  merge()

map.merge("foo", "boo", (oldVal, newVal) -> newVal + " was " + oldVal); System.out.println(map.get("foo")); // boo was foo

*  search

String result = map.search(1, (key, value) -> { System.out.println(Thread.currentThread().getName()); if ("foo".equals(key)) { return value; } return null; }); System.out.println("Result: " + result); // ForkJoinPool.commonPool-worker-2 // main // ForkJoinPool.commonPool-worker-3 // Result: bar

String result = map.searchValues(1, value -> {
        System.out.println(Thread.currentThread().getName());
        if (value.length() > 3) {
            return value;
        }
        return null;
    });
    
    System.out.println("Result: " + result);
    
    // ForkJoinPool.commonPool-worker-2
    // main
    // main
    // ForkJoinPool.commonPool-worker-1
    // Result: solo

*  reduce

`reduce`接收两个`BiFunction`类型的lambda表达式作为参数，第一个参数将键值对转换为任意类型的单值，第二个参数将转换后的值组合到一起组成一个结果值，忽略`null`场景。

String result = map.reduce(1,
        (key, value) -> {
            System.out.println("Transform: " + Thread.currentThread().getName());
            return key + "=" + value;
        },
        (s1, s2) -> {
            System.out.println("Reduce: " + Thread.currentThread().getName());
            return s1 + ", " + s2;
        });
    
    System.out.println("Result: " + result);
    
    // Transform: ForkJoinPool.commonPool-worker-2
    // Transform: main
    // Transform: ForkJoinPool.commonPool-worker-3
    // Reduce: ForkJoinPool.commonPool-worker-3
    // Transform: main
    // Reduce: main
    // Reduce: main
    // Result: r2=d2, c3=p0, han=solo, foo=bar

[https://winterbe.com/posts/2015/05/22/java8-concurrency-tutorial-atomic-concurrent-map-examples/][https_winterbe.com_posts_2015_05_22_java8-concurrency-tutorial-atomic-concurrent-map-examples]

[compare-and-swap]: http://en.wikipedia.org/wiki/Compare-and-swap
[stop]: https://github.com/winterbe/java8-tutorial/blob/master/src/com/winterbe/java8/samples/concurrent/ConcurrentUtils.java
[multithreading-with-atomicinteger-not-working]: https://stackoverflow.com/questions/51451705/multithreading-with-atomicinteger-not-working
[ExecutorService_shutdown docs]: https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ExecutorService.html#shutdown%28%29
[https_winterbe.com_posts_2015_05_22_java8-concurrency-tutorial-atomic-concurrent-map-examples]: https://winterbe.com/posts/2015/05/22/java8-concurrency-tutorial-atomic-concurrent-map-examples/