ゆっちのBlog

2015

Java 8 時代の ConcurrentHashMap で遊んでみた。

またまた Blog の更新をさぼっています。

と言うことで適当なエントリーを書いてみます。

ConcurrentHashMap

J2SE5.0 で java.util.concurrent に ConcurrentHashMap が導入されました。

ConcurrentHashMap は容易に安全に並行処理を実現するために作られました。

Collections.synchronizedMap ってのもあるのですが、それは使い方が微妙で詳しい説明は省略しますが Iterator などの反復処理で ConcurrentModificationException をスローすることがあります。

Collections.synchronizedMap は、Map のロックを取得すると他のスレッドからアクセスを禁じられます。

ConcurrentHashMap だと、Iterator および Enumeration は、ある時点または反復子/列挙の作成以降のハッシュテーブルの状態を反映する要素を返すように設計されているため、

ConcurrentModificationException をスローすることはありません。

しかし、一度に 1 つのスレッドだけしか反復子を使用できません。

あと、HashMap はエントリーのキーや値にnullを使用できますが、ConcurrentHashMapでは使用できません。

NullPointerException が投げられます。

ConcurrentHashMap は、HashMap のようなハッシュを使った Map ですがロックストライピングと言われる粒度の小さなロック方式を採用しています。

つまり、Map 全体をロックして排他アクセスする必要がなければ ConcurrentHashMap は、とても魅力的な存在となります。

しかし、J2SE5.0 の時には、ConcurrentHashMap を便利に使うにはまだまだパーツが揃っていませんでした。

便利に使えそうだったのはアトミックに実行される下記メソッドくらいでした。

public V putIfAbsent(K key, V value)

public boolean remove(Object key, Object value)

public V replace(K key, V value)

public boolean replace(K key, V oldValue, V newValue)

また古い J2SE5.0 ネタかと思わせて、ここから最新の Java SE 8 ネタです。

Java SE 8 で ConcurrentHashMap に追加された便利な機能をいくつか試してみます。

はじめに、ConcurrentHashMap というネーミングから何をやってもスレッドセーフだと思い込んでしまうと泣きます。

例えば、下記プログラムのように複数スレッドから ConcurrentHashMap の値の更新があるとします。

最終的に Key “HOGE”, Value は 80 となることを期待しますが残念なことになります。

package jp.yucchi.concurrenthashmapexample0;

import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *
 * @author Yucchi
 */
class Counter {

    ConcurrentHashMap<String, Integer> map;
    private Random random = new Random();

    public Counter(ConcurrentHashMap<String, Integer> map) {
        this.map = map;
    }

    void incrementCounter() {
        for (int i = 0; i < 10; i++) {
            try {
                TimeUnit.MILLISECONDS.sleep(random.nextInt(50));
            } catch (InterruptedException ex) {
                Logger.getLogger(Counter.class.getName()).log(Level.SEVERE, null, ex);
            }
            int counter = map.get("HOGE");
            map.put("HOGE", ++counter);
        }
        System.out.println(Thread.currentThread().getName() + ": HOGE, " + map.get("HOGE"));
    }

}

package jp.yucchi.concurrenthashmapexample0;

import java.util.Random;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.TimeUnit;

import java.util.logging.Level;

import java.util.logging.Logger;

/**

* @author Yucchi

class Counter {

ConcurrentHashMap<String, Integer> map;

private Random random = new Random();

public Counter(ConcurrentHashMap<String, Integer> map) {

this.map = map;

}

void incrementCounter() {

for (int i = 0; i < 10; i++) {

try {

TimeUnit.MILLISECONDS.sleep(random.nextInt(50));

} catch (InterruptedException ex) {

Logger.getLogger(Counter.class.getName()).log(Level.SEVERE, null, ex);

}

int counter = map.get("HOGE");

map.put("HOGE", ++counter);

}

System.out.println(Thread.currentThread().getName() + ": HOGE, " + map.get("HOGE"));

}

package jp.yucchi.concurrenthashmapexample0;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 *
 * @author Yucchi
 */
public class ConcurrentHashMapExample0 {

    public static void main(String[] args) {

        ConcurrentHashMap<String, Integer> count = new ConcurrentHashMap<>();
        count.put("HOGE", 0);

        Counter counter = new Counter(count);

        CompletableFuture<Void> cnt_1 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_2 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_3 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_4 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_5 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_6 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_7 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_8 = CompletableFuture.runAsync(() -> counter.incrementCounter());

        CompletableFuture<Void> allTask = CompletableFuture.allOf(cnt_1, cnt_2, cnt_3, cnt_4, cnt_5, cnt_6, cnt_7, cnt_8);
        allTask.join();

        if (allTask.isDone()) {
            count.forEach(1, (key, value) -> {
                System.out.println(Thread.currentThread().getName() + " : " + key + ", " + value);
            });
        }
    }

}

package jp.yucchi.concurrenthashmapexample0;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.ConcurrentHashMap;

/**

* @author Yucchi

public class ConcurrentHashMapExample0 {

public static void main(String[] args) {

ConcurrentHashMap<String, Integer> count = new ConcurrentHashMap<>();

count.put("HOGE", 0);

Counter counter = new Counter(count);

CompletableFuture<Void> cnt_1 = CompletableFuture.runAsync(() -> counter.incrementCounter());

CompletableFuture<Void> cnt_2 = CompletableFuture.runAsync(() -> counter.incrementCounter());

CompletableFuture<Void> cnt_3 = CompletableFuture.runAsync(() -> counter.incrementCounter());

CompletableFuture<Void> cnt_4 = CompletableFuture.runAsync(() -> counter.incrementCounter());

CompletableFuture<Void> cnt_5 = CompletableFuture.runAsync(() -> counter.incrementCounter());

CompletableFuture<Void> cnt_6 = CompletableFuture.runAsync(() -> counter.incrementCounter());

CompletableFuture<Void> cnt_7 = CompletableFuture.runAsync(() -> counter.incrementCounter());

CompletableFuture<Void> cnt_8 = CompletableFuture.runAsync(() -> counter.incrementCounter());

CompletableFuture<Void> allTask = CompletableFuture.allOf(cnt_1, cnt_2, cnt_3, cnt_4, cnt_5, cnt_6, cnt_7, cnt_8);

allTask.join();

if (allTask.isDone()) {

count.forEach(1, (key, value) -> {

System.out.println(Thread.currentThread().getName() + " : " + key + ", " + value);

});

}

このプログラムの実行結果は次のようになります。

ForkJoinPool.commonPool-worker-18: HOGE, 48
ForkJoinPool.commonPool-worker-29: HOGE, 58
ForkJoinPool.commonPool-worker-8: HOGE, 65
ForkJoinPool.commonPool-worker-4: HOGE, 67
ForkJoinPool.commonPool-worker-11: HOGE, 68
ForkJoinPool.commonPool-worker-15: HOGE, 70
ForkJoinPool.commonPool-worker-25: HOGE, 73
ForkJoinPool.commonPool-worker-22: HOGE, 75
main : HOGE, 75

更新処理まではスレッドセーフでないということですね。

それではどのように対処すればいいでしょうか？

java.util.concurrent.atomic.LongAdder を使ってみます。

LongAdder クラスは、プリミティブ型に対するアトミック操作を行うため Java SE 8 から追加されたクラスです。

LongAdder クラスには、add(1)と等価な increment() メソッドと、add(-1)と等価な decrement() メソッドがあります。

LongAdder クラスを使って修正したプログラムは下記のようになります。

package jp.yucchi.concurrenthashmapexample1;

import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *
 * @author Yucchi
 */
class Counter {

    ConcurrentHashMap<String, LongAdder> map;
    private Random random = new Random();

    public Counter(ConcurrentHashMap<String, LongAdder> map) {
        this.map = map;
    }

    void incrementCounter() {
        for (int i = 0; i < 10; i++) {
            try {
                TimeUnit.MILLISECONDS.sleep(random.nextInt(50));
            } catch (InterruptedException ex) {
                Logger.getLogger(Counter.class.getName()).log(Level.SEVERE, null, ex);
            }
            map.get("HOGE").increment();
        }
        System.out.println(Thread.currentThread().getName() + ": HOGE, " + map.get("HOGE"));
    }

}

package jp.yucchi.concurrenthashmapexample1;

import java.util.Random;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.atomic.LongAdder;

import java.util.logging.Level;

import java.util.logging.Logger;

/**

* @author Yucchi

class Counter {

ConcurrentHashMap<String, LongAdder> map;

private Random random = new Random();

public Counter(ConcurrentHashMap<String, LongAdder> map) {

this.map = map;

}

void incrementCounter() {

for (int i = 0; i < 10; i++) {

try {

TimeUnit.MILLISECONDS.sleep(random.nextInt(50));

} catch (InterruptedException ex) {

Logger.getLogger(Counter.class.getName()).log(Level.SEVERE, null, ex);

}

map.get("HOGE").increment();

}

System.out.println(Thread.currentThread().getName() + ": HOGE, " + map.get("HOGE"));

}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;

/**
 *
 * @author Yucchi
 */
public class ConcurrentHashMapExample1 {

    public static void main(String[] args) {

        ConcurrentHashMap<String, LongAdder> count = new ConcurrentHashMap<>();
        count.put("HOGE", new LongAdder());

        Counter counter = new Counter(count);

        CompletableFuture<Void> cnt_1 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_2 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_3 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_4 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_5 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_6 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_7 = CompletableFuture.runAsync(() -> counter.incrementCounter());
        CompletableFuture<Void> cnt_8 = CompletableFuture.runAsync(() -> counter.incrementCounter());

        CompletableFuture<Void> allTask = CompletableFuture.allOf(cnt_1, cnt_2, cnt_3, cnt_4, cnt_5, cnt_6, cnt_7, cnt_8);
        allTask.join();

        if (allTask.isDone()) {
            count.forEach(1, (key, value) -> {
                System.out.println(Thread.currentThread().getName() + " : " + key + ", " + value);
            });
        }
    }

}

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.atomic.LongAdder;

/**

* @author Yucchi

public class ConcurrentHashMapExample1 {

public static void main(String[] args) {

ConcurrentHashMap<String, LongAdder> count = new ConcurrentHashMap<>();

count.put("HOGE", new LongAdder());