Atomic Variables and ConcurrentMap-蒲公英云

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 方法定义如下：

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:

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
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