並行性

項目 78 共有された可変データへのアクセスを同期する
項目 79 過剰な同期は避ける
項目 80 スレッドよりもエグゼキュータ、タスク、ストリームを選ぶ
項目 81 wait と notify よりも並行処理ユーティリティを選ぶ
項目 82 スレッド安全性を文書化する
項目 83 遅延初期化を注意して使う
項目 84 スレッドスケジューラに依存しない

1. 項目 78 共有された可変データへのアクセスを同期する

1.1. 結論

synchronizedは同期化を行い、アトミック性と可視性を保証する
複数スレッドが可変データを共有する場合には、そのデータを読み書きするスレッドは同期を行わなければならない
- 同期を行わなければ、あるスレッドで行われた変更が他のスレッドから見えることが保証されないから
volatileはアトミック性は保証しないが、可視性を保証する
スレッド間通信だけが必要で、相互排他が必要なければ、volatileを使用できるが、正しく使うのは難しい
- 相互排他が必要かどうかを判断するのが難しいから
可変データの共有はなるべく回避すべき
- 扱いが難しいから

1.2. アトミック性

1つのスレッドだけがある時点で1つのメソッドやブロックを実行する性質
下のようにsynchronizedを使うと、メソッドや処理をアトミックにできる

1
2
3


public synchronized void method() { 
  //アトミックな処理
}

1.3. 可視性

あるスレッドから変数に書き込んだ値が、別スレッドから観測できる性質
下記のクラスのバックグラウンドスレッドは停止するか？

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


public class StopThread1 {
  private static boolean stopRequested;

  public static void main(String[] args) throws InterruptedException {
    Thread backgroundThead = new Thread(() -> {
      int i = 0;
      while (!stopRequested) {
        i++;
      }
    });
    backgroundThead.start();
    TimeUnit.SECONDS.sleep(1);
    stopRequested = true;
  }
}

バックグラウンドスレッドは無限ループする可能性がある
- バックグラウンドスレッドからメインスレッドのbackgroundTheadに書き込んだ値が見えるかが保証されないから
下記のクラスのバックグラウンドスレッドは停止するか？

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23


public class StopThread2 {
  private static boolean stopRequested;

  private static synchronized void requestStop() {
    stopRequested = true;
  }

  private static synchronized boolean stopRequested() {
    return stopRequested;
  }

  public static void main(String[] args) throws InterruptedException {
    Thread backgroundThead = new Thread(() -> {
      int i = 0;
      while (!stopRequested()) {
        i++;
      }
    });
    backgroundThead.start();
    TimeUnit.SECONDS.sleep(1);
    requestStop();
  }
}

バックグラウンドスレッドは停止することが保証される
- 読み込み操作と書き込み操作の両方が同期されていて、バックグラウンドスレッドからメインスレッドのbackgroundTheadに書き込んだ値が見えることが保証されるから

1.4. volatile

volatileはアトミック性は保証しないが、可視性を保証する
先ほどのStopThread2クラスはvolatileを使うことで、下記のようにより簡単なコードにできる
- synchronizedを相互排他のためではなく、スレッド間通信のためだけにしようしているから

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


public class StopThread3 {
  private static volatile boolean stopRequested;

  public static void main(String[] args) throws InterruptedException {
    Thread backgroundThead = new Thread(() -> {
      int i = 0;
      while (!stopRequested) {
        i++;
      }
    });
    backgroundThead.start();
    TimeUnit.SECONDS.sleep(1);
    stopRequested = true;
  }
}

volatileは「アトミックのように見えて実はアトミックではない」という操作に対して使わないようにする注意が必要
- volatileはアトミック性を保証しないから
- 例えば、下記のgenerateSerialNumber()を複数のスレッドから呼び出したら、違う値が帰ってくるか？
1 2 3 4 5

private static volatile int nextSerialNumber = 0; public static int generateSerialNumber() { return nextSerialNumber++; }
- 違う値が帰ってくることは保証されない
  - インクリメントは「アトミックに見えるが実はアトミックでない操作」だから。インクリメントは、値の読み出しと古い値に1を加えた値を書き戻す2つの操作を行う。スレッドが古い値を読み出して新たな値を書き戻す間に、別のスレッドがフィールドを読み出すと、最初のスレッドと同じ値が見えて、同じ値を返してしまう
- nextSerialNumber + 1 はまずどこに書き込まれているか？
  - スタック上の一時領域
  - メモリは参照と書き込みだけしかできない。計算できない。

1.5. 可変データの共有の回避

今までの議論のように、可変データへの共有は難しいので、回避できるのであれば回避すべき
回避には2つの方法がある
- 不変データを共有する
- 可変データを共有しない(可変データは単一スレッドで使う)
上記の方針を採用した場合は文書化することが重要
- プログラムが発展する際に方針が維持されるから
AtomicBooleanのようなライブラリを使うと良い。

2. 項目 79 過剰な同期は避ける

2.1. 結論

同期された領域内では、できる限り少ない処理をするべき
- デッドロックやデータ破壊、パフォーマンス低下の原因となるから

2.2. 過剰な同期によるエラーやデッドロック

下記クラスObservableSetは要素が追加されたら任意の処理ができるSetであり、SetObserverインターフェースはObservableSetのObserverインターフェースである

ForwardingSetはSetの転送クラス

項目18：継承よりコンポジションを選ぶで紹介された
Observerパターンを用いている

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45


public class ObservableSet<E> extends ForwardingSet<E> {
	public ObservableSet(Set<E> set) {
		super(set);
	}

	private final List<SetObserver<E>> observers = new ArrayList<>();

	public void addObserver(SetObserver<E> observer) {
		synchronized (observers) {
			observers.add(observer);
		}
	}

	public boolean removeObserver(SetObserver<E> observer) {
		synchronized (observers) {
			return observers.remove(observer);
		}
	}

	private void notifyElementAdded(E element) {
		synchronized (observers) {
			for (SetObserver<E> observer : observers)
				observer.added(this, element);
		}
	}

	@Override
	public boolean add(E element) {
		boolean added = super.add(element);
		if (added)
			notifyElementAdded(element);
		return added;
	}

	@Override
	public boolean addAll(Collection<? extends E> c) {
		boolean result = false;
		for (E element : c)
			result |= add(element); // notifyElementAddedを呼ぶ 
		return result;
		}
	}
  public interface SetObserver<E> {
	  void added(ObservableSet<E> set, E element);
  }

以下のメインメソッドを実行するとどうなるか？

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


public static void main(String[] args) {
  ObservableSet<Integer> set = new ObservableSet<>(new HashSet<>());
  set.addObserver(new SetObserver<>() {
    public void added(ObservableSet<Integer> s, Integer e) {
      System.out.println(e);
      if (e == 23) {
        s.removeObserver(this);
      }
    }
  });
  for (int i = 0; i < 100; i++) {
    set.add(i);
  }
}

i=23のときにConcurrentModificationExceptionが発生する
- notifyElementAdded()でobserversをイテレート処理しているが、notifyElementAdded()中のobserver.added()からobservers.remove()を呼び出していて、observers.remove()がobserversを操作しようとしているので、ConcurrentModificationExceptionとなる

この問題はsyncronizedブロックから未知のメソッドを取り除くことで解決できる

= 同期を最小限にする
下記のようにnotifyElementAdded()からSetObserver#added()(= 未知のメソッド)を取り除くことで、SetObserverの実装との同期を気にしなくてよくなる

1
2
3
4
5
6
7
8
9


    private void notifyElementAdded(E element) {
  	  List<SetObserver<E>> snapshot = null;
  	  synchronized (observers) { // 同期ブロック内ではその時点で observer のコピーを保存
  	    snapshot = new ArrayList<SetObserver<E>>(observers);
  	   }
  	  for (SetObserver<E> observer: snapshot) {
  		  observer.added(this, element);
  	  }
    }

CopyOnWriteArrayListを使っても解決できる
- CopyOnWriteArrayListは前のPJでのちらっと出てた
- 効率が悪いので、使うときは気を付ける
  - synchronizedList

2.3. パフォーマンスの低下

過剰な同期によってパフォーマンスの低下が起こることがある
- ロックの取得にあまり時間はかからないが、メモリの一貫性を保つために遅延が発生するから。また、同期化によってJVM上の最適化がされないことがあるから

2.4. 余談

synchronized(this)
- thisがロックオブジェクト
ロックオブジェクトの話を見ておくといい
- Javaスレッドメモ(Hishidama’s Java thread Memo)

おまけ

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20


class だめな例1 {
   private final List<String> list =
     Collections.synchronizedList(new ArrayList<>());

   public void addFromMultipleThreads(String item) {
       if (!list.contains(item)) {
           list.add(item);
       }
   }
}

class だめな例2 {
   private final List<String> list = new CopyOnWriteArrayList<>();

   public void addFromMultipleThreads(String item) {
       if (!list.contains(item)) {
           list.add(item);
       }
   }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


class いい例1 {
   private final CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();

   public void addFromMultipleThreads(String item) {
       list.addIfAbsent(item);
   }
}

class いい例2 {
   private final List<String> list = new ArrayList<>();

   public synchronized void addFromMultipleThreads(String item) {
       if (!list.contains(item)) {
           list.add(item);
       }
   }
}

余談：ロックストライピングについては以下のサイトを見ておくと良いかも。
・Javaの理論と実践: 優れたHashMapの構築
・ストライピングロックを試す - にょきにょきブログ

3. 項目 80 スレッドよりもエグゼキュータ、タスク、ストリームを選ぶ

3.1. 結論

並列処理では、 Java SE 5 から導入されたエグゼキュータ、タスクを使う。
この機構は、 Java 7 で ForkJoin 機構により強化され、Java 8 で Lambda/Stream でさらに使いやすくなった。 Java 1.4 まででは Java.util.Thread を使うしかなかったが、低レイヤなAPIのため扱いがデリケートであり、もはや直接使う理由はない。

3.2. 並列処理の軌跡

Executorを使うことで、処理の開始、待ち合わせ、周期的実行、といったよくある並列処理のユースケースがカバーされる。
並列処理はデリケートでありバグっていても試験で見つけることが難しいので、この領域で言語標準ライブラリでサポートしてもらえるのは重要。

以下は、並列処理の機能追加によって、どのように記法が変わっていったかを一巡りするソース。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56


@Slf4j
public class Sample {
	public static void main(String[] args) {
		// Java1.4
		Runnable runnable = new Runnable() {
			@Override
			public void run() {
				log.info("run!");
			}
		};
		Thread worker = new Thread(runnable);
		worker.start();
		try {
			log.info("wait...");
			Thread.sleep(1000L);
			worker.join();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		// Java 5
		ExecutorService es = Executors.newSingleThreadExecutor();
		Runnable task = new Runnable() {
			@Override
			public void run() {
				log.info("run!");
			}
		};
		es.submit(task);
		es.shutdown();
		// Java 7
		ForkJoinPool forkJoinPool = new ForkJoinPool();
		long result = forkJoinPool.invoke(new Sigma(100));
		log.info("result={}", result);
		// Java 8 
		long lambdaResult = LongStream.range(1, 10000).parallel().sum();
		log.info("lambdaResult={}", lambdaResult);
	}

	static class Sigma extends RecursiveTask<Long> {
		private final long value;

		public Sigma(long value) {
			this.value = value;
		}

		@Override
		protected Long compute() {
			if (value <= 1L) {
				return value;
			}
			Sigma sigma1 = new Sigma(value - 1);
			sigma1.fork();
			return value + sigma1.join();
		}
	}
}

なお、Java7のForkJoinは、Java5のExecutorをベースにしている。同様に、Java8のparallel-streamはForkJoinをベースにしている。

3.2.1. Executor の種類

いくつか種類があるので知っておくと良い。

使うときは、 https://docs.oracle.com/javase/jp/11/docs/api/java.base/java/util/concurrent/Executors.html を利用すると簡単に使える。
一般的なスレッドプールがだいたい網羅されているユーティリティクラス。
これを通じて、 ExecutorService インタフェースを通じて並列処理を実行する。

Executorsは以下のメソッドを持っているが、いずれもExecutorServiceの生成方法の違いしかない。

newCachedThreadPool : min/maxを指定するスレッドプール
newFixedThreadPool : 固定数のスレッドプール
newScheduledThreadPool : タイマー付きスレッドプール
newSingleThreadExecutor : 1本だけのスレッドプール

3.2.2. Spring Framework のサポート

Springは、Java標準に対しさらに機能を追加したり、インタフェースを統一したクラスを提供している。

https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/core/task/TaskExecutor.html を起点に拡張している。よく使うのは以下の2つ。

ThreadPoolTaskExecutor
ThreadPoolTaskScheduler

リファレンスはこちら

SpringのWebやIntegration、Batchなどを支える並列処理ライブラリ。
ExecutorServiceをベースにしているので、できることは大きくは変わらないが、設定項目のわかりやすさ、DI時の扱いやすさから、SpringベースのAPならばこちらを使うことを考えるとよい。

4. 項目 81 wait と notify よりも並行処理ユーティリティを選ぶ

4.1. 結論

Object.wait() , Object.notify() といったスレッド制御の機構は、非常に扱いづらい。
Java 5 からの java.util.concurrent は並列実行でインスタンスを編集するケースや、同期化についてもサポートしているので、wait/notify は使わないこと。

4.1.1. 何を使うのか

コンカレントコレクション
- ConcurrentHashMap
- ArrayBlockingQueue
シンクロナイザー
- CountDonwLatch
- Semaphore

以下は、CountDownLatchの例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20


  private static final int multiplicity = 3;
	private static final CountDownLatch LATCH = new CountDownLatch(multiplicity);
	// 省略
	SecureRandom random = SecureRandom.getInstanceStrong();
	var es = Executors.newFixedThreadPool(multiplicity);
	for(int i = 0; i<multiplicity; i++){
		es.submit(() -> {
			try {
				var sleepTime = random.nextInt(10);
				log.info("{} sec sleep...", sleepTime);
				TimeUnit.SECONDS.sleep(sleepTime);
				LATCH.countDown();
				LATCH.await();
				log.info("start!!!");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		});
	}
	es.shutdown();

4.2. wait/notifyの難しさ

wait()、notify()、notifyAll() を呼び出すためにはロックが必要。

https://docs.oracle.com/javase/jp/11/docs/api/java.base/java/lang/Object.html#wait(long,int)

に書いてある。

Java のすべてのオブジェクトは条件キューを持っていて、Object クラスの wait() と notify() メソッドによってこのキューを操作することができる。
Object.wait() を呼び出すと、呼び出したスレッドはそのオブジェクトの条件キューに入る。そして、そのスレッドはそこでブロックします。
Object.notify()、または Object.notifyAll() を呼び出すと、そのオブジェクトの条件キューに入っているどれかのスレッドのブロックを解放する。どのスレッドが解放されるかは保証されない。
wait()、notify()、notifyAll() を呼び出すためにはロックが必要。

1
2
3
4


	synchronized(obj) {// obj のロックを取得する必要がある
		obj.wait();
    // スレッドは条件キューに入り、スレッドの動作は止まる。そして、wait() はロックを解放する
	}

一方、notify() や notifyAll() は条件キューに入ったスレッドを解放し、wait() の待ちを解除する。

1
2
3
4


synchronized(obj) { // obj のロックを取得する必要がある 
// 条件キューに含まれているすべてのスレッドのブロックを解放する
  obj.notifyAll();
}

また、注意としては以下が知られている。

while ループで条件が満たされてない場合はループさせます
- while ループ外で wait() を呼び出してはいけません
なるべく notifyAll() を使う

これをコードにするとこうなる。

1
2
3
4
5
6
7


// wait を利用するための標準イディオム
synchronized (obj) {
  while (<条件が満たされていない>) { 
    obj.wait();
  }
  // 満たされた条件に対して適切な処理を行う
}

このようなデリケートな処理をミスなく実装しきるのは困難なため、 java.util.concurrent のようなライブラリを使うべし。

5. 項目 82 スレッド安全性を文書化する

5.1. 結論

スレッド安全性について「クラスのJavaDocに書く」もしくはアノテーションを付けましょう。
- 利用者が、スレッド安全性に関しての何らかの仮定をせざるを得なくなってしまうため。
  - 仮定が誤っていれば、「不十分な同期」や「過度な同期」になってしまい重大なエラーとなるかもしれないから。

備考：スレッド安全性のアノテーションについて
・Immutable
・ThreadSafe
・NotThreadSafe

5.1.1. スレッド安全性のレベル

スレッド安全性のレベルとして、一般的なものとして以下があります。

不変(Immutable)
- 外部から*定数のように見えるもの。
- 例：String, Integer, BigIntegerなど。
無条件スレッドセーフ(unconfitionally thread-safe)
- クラスのインスタンスは可変だが、全てのメソッドが十分な内部同期を含んでいる。(ユーザ側での同期は不要。)
- 例：RandomやConcurrentHashMapなど。
条件付きスレッドセーフ(conditionally thread-safe)
- いくつかのメソッドが、ユーザ側での同期を必要とする。
- 例：Collections.synchronizedラッパーが返すコレクションのイテレータは外部同期が必要。
スレッドセーフでない(not thread-safe)
- 全てのメソッドは、ユーザ側での同期を必要とする。
- 例：ArrayListやHashMapなどの汎用コレクションの実装。
スレッド敵対(thread-hostile)
- 全てのメソッドを、ユーザ側で同期して利用したとしても、安全ではない。
- スレッド敵対は、一般に、staticのデータを同期なしで変更することに起因している。
- 例：Javaプラットフォームライブラリにほとんどないが、System#runFinalizersOnExitメソッドは、スレッド敵対で非推奨。

備考：<System#runFinalizersOnExit が非推奨の理由>
・System#runFinalizersOnExitで、Runtime.runFinalizersOnExit(flag);を呼んでいる。
・Runtime#runFinalizersOnExitで、Shutdown.setRunFinalizersOnExit(value);を呼んでいる。
・上記メソッドにより、終了時に各オブジェクトの finalize() メソッドが必ず呼び出されるようになる。
・その際、もし並行して実行されているオブジェクトがあると危険。ということなのかな。

5.2. サンプル

無条件スレッドセーフ

1
2
3
4
5
6
7
8


// ユーザ側からlockオブジェクトにアクセス不可であるため、ユーザはオブジェクトの同期に干渉することは不可能。
private final Object lock = new Object();

public void foo() {
    synchronized(lock) {
        ....
    }
}

備考：lockオブジェクトをprivate static finalにしていれば、複数のインスタンスで1つのlockオブジェクトとなり、JVM上で1つしか実行できないようにする等はできる。

同期化する箇所を、局所化する際にロックオブジェクトを使う例。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16


class DataAccessor {
  private final List<> item;

  get(){}
  remove(){}

  get() {
    List<> itemClone;
    synchronized(lock){
      item = list.get();
      var itemClone = item.clone();
      // itemClone の加工
    }
    return itemClone;
  }
}

条件付きスレッドセーフ

1
2
3
4
5
6
7
8


Map<K, V> map = Collections.synchronizedMap(new HashMap<K, V>()));
Set<K> set = m.keySet();
// イテレータを利用する場合、map に対するロックが必要
synchronized (map) {
    for (K key: set) { 
        ....
    }
}

イテレータによる走査中にmapに変更が加わると、ConcurrentModificationExceptionが発生するため。

5.3. ポイント

ユーザにとって、スレッドセーフなのかどうかを示すために、以下をしましょう。
- スレッド安全性についてJavaDocに記載する。
- スレッド安全性についてのアノテーションを付ける。

6. 項目 83 遅延初期化を注意して使う

6.1. 結論

ほとんどの場合は、普通に初期化しましょう。

遅延初期化はもろ刃の剣であるため。
- フィールドへのアクセスコストを増加させる犠牲を払う。
- クラスの初期化コストやインスタンスの生成コストを抑えれる。

遅延初期化を利用するシーン

パフォーマンス目標を達成するため。
問題となる初期化循環を断つため。

以下、遅延初期化を利用する場合。

インスタンスフィールドの場合
- 基本は二重チェックイデオム。
- ただし、複数回の初期化が許容される場合は、単一チェックイデオムも可能。
staticフィールドの場合
- 遅延初期化ホルダークラスイデオム

6.2. サンプル

6.2.1. インスタンスフィールドの遅延初期化

二重チェックイデオム

初回アクセス時、ロックする。
２回目以降、ロックせずに済む。
- ロックしないので、フィールドをvolatile宣言して、直接メモリ書き込みを行わせることで、複数スレッドからのアクセスにも対応できるようにしている。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


private volatile FieldType field;
 private FieldType getField() {
    FieldType result = field;
    if (result == null) {  // First check (no locking)
        synchronized(this) {
            if (field == null)  // Second check (with locking)
                field = result = computeFieldValue();
        }
    }
    return result;
}

備考：ローカル変数resultは厳密には不要。
しかし、約25%のパフォーマンス改善のために使用している。2回目以降のアクセス時に、field値の読み込みが1回しか行われないよう保証している。

「Java 5からダブルチェックイデオムが使えるようになった。Java 1.4まで機能しそうに見えて、機能しなかった。」ということを知っておきましょう。

単一チェックイデオム

何度も初期化する可能性がある場合に利用する。
getField()が初めて呼ばれた時に、初めて初期化される。

1
2
3
4
5
6
7
8


private volatile FieldType field;

private FieldType getField() {
    FieldType result = field;
    if (result == null)
        field = result = computeFieldValue();
    return result;
}

普段使用するのに対しては、非標準な技法だけれども、アーキテクチャによっては、フィールドへのアクセスが速くなります。
理由は、同期しないので別スレッドの処理を待たなくなるから。

6.2.2. staticフィールドの遅延初期化

クラスが初めて見られたときに、初期化するというクラスローダの仕組みを利用している。

getField()が初めて呼ばれた時に、初めてFieldHolder.fieldが読み出されて、初期化される。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


public class SampleClass{

  private static class FieldHolder {
    static final FieldType field = computeFieldValue();
  }
  public static FieldType getField() {
    return FieldHolder.field;
  }

}

Springを使っている場合は、アノテーション@Lazy(true)を使うことが多いでしょう。
テストのしやすさも、JavaだとClassLoaderを触らないとダメだが、Spring管理だとUTの範囲ではなくなる。SIとかに回せる。

6.3. ポイント

遅延初期化は普通は使わないけれど、利用するシーンとしては以下。

パフォーマンス目標を達成するため。
問題となる初期化循環を断つため。

とりあえずは、以下と覚えておきましょう。

インスタンスフィールドの場合
- 基本は二重チェックイデオム。
staticフィールドの場合
- 遅延初期化ホルダークラスイデオムの一択。

7. 項目 84 スレッドスケジューラに依存しない

7.1. 結論

アプリケーションの動作の正しさに関して、スレッドスケジューラに依存してはいけない。

スレッドスケジューラは、「どのスレッドをどのくらいの時間実行するか」を決める。その決定は、OSのポリシーによって異なる可能性があるから。

実行可能なスレッド数はプロセッサの数と比べて非常に大きくならないようにするべき。

スレッドスケジューラーがどのようなアルゴリズムを採用していてもプログラムは正しく動作するから。
エグゼキューターフレームワークでは、スレッド数を制限したスレッドプールを作成できる。
- 参考：ThreadPoolExecutor (Java Platform SE 8 )

極端な話、もしスレッド数が1万スレッドになった場合、何が起きるんでしょうか？関連キーワード：context switch, c10k問題

1スレッド=1クライアント(1万スレッド=1万クライアント)：c10k問題

1スレッドあたり2MB

1万スレッド=20GB

もしメモリが足りてもCPUが対応できなくなっていく。

context switchが多くなる。（コンピュータの処理装置(CPU)が現在実行している処理の流れ(プロセス、スレッド)を一時停止し、別のものに切り替えて実行を再開すること。）

結論として、スレッド数には限界値を設けましょう。

7.2. サンプル

スレッドはbusy waitを行うべきではない。
busy waitは、プロセッサへの負荷をかなり増大させて、他のスレッドができる処理の量を減らしてしまう可能性がある。

busy waitとは、プロセスが条件に成り立つかどうかを定期的にチェックする手法の一種。
今回は、何かが起きるのを待つ(カウントが0かどうかを確認する)ために共有されたオブジェクトを繰り返し検査している。

CountDownLatch (Java Platform SE 8 )を遅くしたサンプルプログラム。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27


class SlowCountDownLatch {
    private int count;

    public SlowCountDownLatch(int count) {
        if (count < 0)
            throw new IllegalArgumentException(count + " < 0");
        this.count = count;
    }

    // ここの実装がめっちゃbusy-wait
    // スピンロック（Spinlock）ともいう。
    // ソフトウェア工学におけるロックの一種で、
    // スレッドがロックを獲得できるまで単純にループ（スピン）して定期的にロックをチェックしながら待つ方式。
    public void await() {
        while (true) {
            synchronized (this) {
                if (count == 0)
                    return;
            }
        }
    }

    public synchronized void countDown() {
        if (count != 0)
            count--;
    }
}

CountDownLatch#await()の実コード

「スレッド割り込みが発生するまで」としている。

1
2
3
4
5
6


/**
  スレッドで割り込みが発生しないかぎり、ラッチのカウント・ダウンがゼロになるまで現在のスレッドを待機させます。
*/
public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

7.3. ポイント

アプリケーションの動作の正しさに関して、スレッドスケジューラに依存してはいけない。

スレッドスケジューラはOSのポリシー依存であるため。

実行可能なスレッド数はプロセッサの数と比べて非常に大きくならないようにするべき。
そのために、エグゼキューターフレームワーク等で、スレッド数を制限したスレッドプールを作成して、対応するようにしましょう。