AtomicInteger クラスの一般的なメソッド#

1
public final int get() //現在の値を取得
2
public final int getAndSet(int newValue)//現在の値を取得し、新しい値を設定
3
public final int getAndIncrement()//現在の値を取得し、インクリメント
4
public final int getAndDecrement() //現在の値を取得し、デクリメント
5
public final int getAndAdd(int delta) //現在の値を取得し、(delta)だけ加算
6
boolean compareAndSet(int expect, int update) //期待値と実際の値が等しい場合、原子的にその値を update に設定
7
public final void lazySet(int newValue)//最終的に newValue を設定。lazySet 後でも他のスレッドがしばらくの間旧値を読む可能性があります。

AtomicInteger クラスの使用例 :

1
import java.util.concurrent.atomic.AtomicInteger;
2

3
public class AtomicIntegerTest {
4

5
    public static void main(String[] args) {
6
        int temvalue = 0;
7
        AtomicInteger i = new AtomicInteger(0);
8
        temvalue = i.getAndSet(3);
9
        System.out.println("temvalue:" + temvalue + ";  i:" + i); //temvalue:0;  i:3
10
        temvalue = i.getAndIncrement();
11
        System.out.println("temvalue:" + temvalue + ";  i:" + i); //temvalue:3;  i:4
12
        temvalue = i.getAndAdd(5);
13
        System.out.println("temvalue:" + temvalue + ";  i:" + i); //temvalue:4;  i:9
14
    }
15
}

基本データ型原子クラスの利点#

簡単な例を通じて、基本データ型原子クラスの利点を見てみましょう。

1. マルチスレッド環境で原子クラスを使用せずにスレッドセーフを保証する（基本データ型）

   1
   class Test {
   2
           private volatile int count = 0;
   3
           // count++ をスレッドセーフに実行するにはロックが必要
   4
           public synchronized void increment() {
   5
                     count++;
   6
           }
   7
   
   8
           public int getCount() {
   9
                     return count;
   10
           }
   11
   }

2. マルチスレッド環境で原子クラスを使用してスレッドセーフを保証する（基本データ型）

   1
   class Test2 {
   2
           private AtomicInteger count = new AtomicInteger();
   3
   
   4
           public void increment() {
   5
                     count.incrementAndGet();
   6
           }
   7
         // AtomicInteger を使用すれば、ロックを取らなくてもスレッドセーフを実現できる。
   8
          public int getCount() {
   9
                   return count.get();
   10
           }
   11
   }

AtomicInteger のスレッドセーフ原理の簡易分析#

AtomicInteger クラスの一部ソースコード：

1
// 更新操作のために Unsafe.compareAndSwapInt を使用する設定
2
    private static final Unsafe unsafe = Unsafe.getUnsafe();
3
    private static final long valueOffset;
4

5
    static {
6
        try {
7
            valueOffset = unsafe.objectFieldOffset
8
                (AtomicInteger.class.getDeclaredField("value"));
9
        } catch (Exception ex) { throw new Error(ex); }
10
    }
11

12
    private volatile int value;

AtomicInteger クラスは主に CAS (compare and swap) + volatile およびネイティブメソッドを用いて原子操作を保証し、synchronized の高オーバーヘッドを回避して実行効率を大幅に向上させます。

CAS の原理は、期待値と現在値を比較し、同じであれば新しい値に更新することです。Unsafe クラスの objectFieldOffset() メソッドはネイティブメソッドで、元の値のメモリアドレスを取得するために使われます。さらに value は volatile 変数で、メモリ上で可視性が保証されるため、JVM は任意の時点で任意のスレッドがこの値の最新値を必ず取得できるようにします。

使用例#

• AtomicReference クラス:

  1
  import java.util.concurrent.atomic.AtomicReference;
  2
  
  3
  public class AtomicReferenceTest {
  4
  
  5
      public static void main(String[] args) {
  6
          AtomicReference < Person > ar = new AtomicReference < Person > ();
  7
          Person person = new Person("SnailClimb", 22);
  8
          ar.set(person);
  9
          Person updatePerson = new Person("Daisy", 20);
  10
          ar.compareAndSet(person, updatePerson);
  11
  
  12
          System.out.println(ar.get().getName());
  13
          System.out.println(ar.get().getAge());
  14
      }
  15
  }
  16
  
  17
  class Person {
  18
      private String name;
  19
      private int age;
  20
  
  21
      public Person(String name, int age) {
  22
          super();
  23
          this.name = name;
  24
          this.age = age;
  25
      }
  26
  
  27
      public String getName() {
  28
          return name;
  29
      }
  30
      public void setName(String name) {
  31
          this.name = name;
  32
      }
  33
  
  34
      public int getAge() {
  35
          return age;
  36
      }
  37
      public void setAge(int age) {
  38
          this.age = age;
  39
      }
  40
  }

  上記のコードはまず Person オブジェクトを作成し、それを AtomicReference に設定します。その後 compareAndSet メソッドを呼び出して ar の値を更新します。もし ar の値が person であれば、それを updatePerson に設定します。実装原理は AtomicInteger クラスの compareAndSet メソッドと同じです。上記のコードを実行した場合の出力は次のとおりです。

  1
  Daisy
  2
  20

• AtomicStampedReference クラスの使用例 :

  1
  import java.util.concurrent.atomic.AtomicStampedReference;
  2
  
  3
  public class AtomicStampedReferenceDemo {
  4
      public static void main(String[] args) {
  5
          // 初期値とスタンプ値をインスタンス化
  6
          final Integer initialRef = 0, initialStamp = 0;
  7
          final AtomicStampedReference<Integer> asr = new AtomicStampedReference<>(initialRef, initialStamp);
  8
          System.out.println("currentValue=" + asr.getReference() + ", currentStamp=" + asr.getStamp());
  9
  
  10
          // compare and set
  11
          final Integer newReference = 666, newStamp = 999;
  12
          final boolean casResult = asr.compareAndSet(initialRef, newReference, initialStamp, newStamp);
  13
          System.out.println("currentValue=" + asr.getReference()
  14
                  + ", currentStamp=" + asr.getStamp()
  15
                  + ", casResult=" + casResult);
  16
  
  17
          // 現在の値と現在の stamp 値を取得
  18
          int[] arr = new int[1];
  19
          final Integer currentValue = asr.get(arr);
  20
          final int currentStamp = arr[0];
  21
          System.out.println("currentValue=" + currentValue + ", currentStamp=" + currentStamp);
  22
  
  23
          // stamp のみを個別に設定
  24
          final boolean attemptStampResult = asr.attemptStamp(newReference, 88);
  25
          System.out.println("currentValue=" + asr.getReference()
  26
                  + ", currentStamp=" + asr.getStamp()
  27
                  + ", attemptStampResult=" + attemptStampResult);
  28
  
  29
          // 現在の値と stamp を再設定
  30
          asr.set(initialRef, initialStamp);
  31
          System.out.println("currentValue=" + asr.getReference() + ", currentStamp=" + asr.getStamp());
  32
  
  33
          // [推奨されません。コメントの意味を理解した場合のみ] weak compare and set
  34
          // 混乱！weakCompareAndSet は結局 compareAndSet を呼び出します。[バージョン: jdk-8u191]
  35
          // ただしコメントには「May fail spuriously and does not provide ordering guarantees,
  36
          // so is only rarely an appropriate alternative to compareAndSet.」と書かれています。
  37
          // もしかすると JVM がネイティブメソッドで転送しているのかもしれません。
  38
          final boolean wCasResult = asr.weakCompareAndSet(initialRef, newReference, initialStamp, newStamp);
  39
          System.out.println("currentValue=" + asr.getReference()
  40
                  + ", currentStamp=" + asr.getStamp()
  41
                  + ", wCasResult=" + wCasResult);
  42
      }
  43
  }

  出力結果は以下のとおりです：

  1
  currentValue=0, currentStamp=0
  2
  currentValue=666, currentStamp=999, casResult=true
  3
  currentValue=666, currentStamp=999
  4
  currentValue=666, currentStamp=88, attemptStampResult=true
  5
  currentValue=0, currentStamp=0
  6
  currentValue=666, currentStamp=999, wCasResult=true

• AtomicMarkableReference クラスの使用例 :

  1
  import java.util.concurrent.atomic.AtomicMarkableReference;
  2
  
  3
  public class AtomicMarkableReferenceDemo {
  4
      public static void main(String[] args) {
  5
          // 初期値とマーク値を取得
  6
          final Boolean initialRef = null, initialMark = false;
  7
          final AtomicMarkableReference<Boolean> amr = new AtomicMarkableReference<>(initialRef, initialMark);
  8
          System.out.println("currentValue=" + amr.getReference() + ", currentMark=" + amr.isMarked());
  9
  
  10
          // compare and set
  11
          final Boolean newReference1 = true, newMark1 = true;
  12
          final boolean casResult = amr.compareAndSet(initialRef, newReference1, initialMark, newMark1);
  13
          System.out.println("currentValue=" + amr.getReference()
  14
                  + ", currentMark=" + amr.isMarked()
  15
                  + ", casResult=" + casResult);
  16
  
  17
          // 現在の値と現在の mark 値を取得
  18
          boolean[] arr = new boolean[1];
  19
          final Boolean currentValue = amr.get(arr);
  20
          final boolean currentMark = arr[0];
  21
          System.out.println("currentValue=" + currentValue + ", currentMark=" + currentMark);
  22
  
  23
          // 単独で mark 値を設定
  24
          final boolean attemptMarkResult = amr.attemptMark(newReference1, false);
  25
          System.out.println("currentValue=" + amr.getReference()
  26
                  + ", currentMark=" + amr.isMarked()
  27
                  + ", attemptMarkResult=" + attemptMarkResult);
  28
  
  29
          // 現在の値と mark を再設定
  30
          amr.set(initialRef, initialMark);
  31
          System.out.println("currentValue=" + amr.getReference() + ", currentMark=" + amr.isMarked());
  32
  
  33
          // [推奨されません。コメントの意味を理解した場合のみ] weak compare and set
  34
          // 混乱！weakCompareAndSet は結局 compareAndSet を呼び出します。
  35
          // ただしコメントには「May fail spuriously and does not provide ordering guarantees,
  36
          // so is only rarely an appropriate alternative to compareAndSet.」と書かれています。
  37
          // もしかすると JVM がネイティブメソッドで転送しているのかもしれません
  38
          final boolean wCasResult = amr.weakCompareAndSet(initialRef, newReference1, initialMark, newMark1);
  39
          System.out.println("currentValue=" + amr.getReference()
  40
                  + ", currentMark=" + amr.isMarked()
  41
                  + ", wCasResult=" + wCasResult);
  42
      }
  43
  }

  出力結果は以下のとおりです：

  1
  currentValue=null, currentMark=false
  2
  currentValue=true, currentMark=true, casResult=true
  3
  currentValue=true, currentMark=true
  4
  currentValue=true, currentMark=false, attemptMarkResult=true
  5
  currentValue=null, currentMark=false
  6
  currentValue=true, currentMark=true, wCasResult=true

BlockingQueue の概要#

前述のとおり、ConcurrentLinkedQueue は高性能な非ブロッキングキューです。ここからはブロッキングキューである BlockingQueue について説明します。BlockingQueue は広く「生産者-消費者」問題で用いられ、ブロック可能な挿入と削除のメソッドを提供します。キューが満杯のとき、生産者スレッドはブロックされ、キューに空きができるまで待機します。キューが空のとき、消費者スレッドはブロックされ、キューが非空になるまで待機します。

BlockingQueue はインタフェースで、Queue を継承しています。従って、その実装クラスも Queue の実装として使用でき、Queue は Collection インタフェースを継承します。以下は BlockingQueue の関連実装クラスです。

以下では、3つの一般的な BlockingQueue の実装クラス：ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue について主に説明します。

ArrayBlockingQueue#

ArrayBlockingQueue は BlockingQueue インタフェースの有界キュー実装クラスで、底層は配列を用いて実装されています。

1
public class ArrayBlockingQueue<E>
2
extends AbstractQueue<E>
3
implements BlockingQueue<E>, Serializable{}

ArrayBlockingQueue は作成後、容量を変更できません。その並行制御は再入可能ロック ReentrantLock を用い、挿入操作でも読み取り操作でもロックを取得してから操作を行います。キューの容量が満杯の場合、要素を挿入しようとすると操作がブロックされ、空のキューから要素を取り出そうとしても同様にブロックされます。

ArrayBlockingQueue はデフォルトではスレッドがキューへアクセスする公平性を保証しません。公平性とは、スレッドの待機時間の絶対的な順序に厳密に従うこと、つまり最初に待っていたスレッドが最初に ArrayBlockingQueue にアクセスできることを指します。一方、公平性を保証しない場合は、アクセスの順序が厳密な時間順にはなりません。長時間待機しているスレッドがアクセスできないこともあります。公平性を保証する ArrayBlockingQueue を得たい場合は、以下のコードを使用します：

1
private static ArrayBlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(10,true);

LinkedBlockingQueue#

LinkedBlockingQueue は一方向リンクリストを基盤とするブロッキングキューです。無限大に近いキューとして使うことも、有界キューとして使うこともできます。FIFO を満たし、ArrayBlockingQueue に比べてスループットが高くなります。ただし、容量が急速に増えると大量のメモリを消費します。通常、LinkedBlockingQueue オブジェクトを作成する際にはサイズを指定します。未指定の場合、容量は Integer.MAX_VALUE となります。

関連コンストラクタ:

1
/**
2
     *ある意味での無界キュー
3
     * Creates a {@code LinkedBlockingQueue} with a capacity of
4
     * {@link Integer#MAX_VALUE}.
5
     */
6
    public LinkedBlockingQueue() {
7
        this(Integer.MAX_VALUE);
8
    }
9

10
    /**
11
     *有界キュー
12
     * Creates a {@code LinkedBlockingQueue} with the given (fixed) capacity.
13
     *
14
     * @param capacity the capacity of this queue
15
     * @throws IllegalArgumentException if {@code capacity} is not greater
16
     *         than zero
17
     */
18
    public LinkedBlockingQueue(int capacity) {
19
        if (capacity <= 0) throw new IllegalArgumentException();
20
        this.capacity = capacity;
21
        last = head = new Node<E>(null);
22
    }

PriorityBlockingQueue#

PriorityBlockingQueue は優先度をサポートする無限ブロッキングキューです。デフォルトでは要素は自然順序でソートされ、あるいは Comparable を実装したクラスを用いてソート規則を指定することができます。また、初期化時にコンストラクタ引数 Comparator を用いてソート規則を指定することもできます。

PriorityBlockingQueue の同期制御は再入可能ロック ReentrantLock を用い、キューは無界です（ArrayBlockingQueue は有界、LinkedBlockingQueue も構築時に容量を指定して最大容量を設定可能ですが、PriorityBlockingQueue は初期のキューサイズのみを指定でき、後から要素を挿入する際には容量が足りない場合自動的に拡張されます）。

簡単に言えば、それは PriorityQueue のスレッドセーフ版です。null 値の挿入はできず、挿入されるオブジェクトは比較可能でなければならず、そうでない場合は ClassCastException が発生します。挿入操作の put はブロックされません。無界キューなので（take はキューが空のときにブロックします）。