Java : ConcurrentLinkedQueue (並列処理用キュー) - API使用例

ConcurrentLinkedQueue (Java SE 20 & JDK 20) の使用例まとめです。
だいたいのメソッドを網羅済みです。
API仕様のおともにどうぞ。

概要

リンク・ノードに基づく、制限なしのスレッドセーフなキューです。このキューはFIFO (先入れ先出し)で要素を順序付けします。このキューの先頭は、キューに入っていた時間がもっとも長い要素です。このキューの末尾は、キューに入っていた時間がもっとも短い要素です。

クラス構成

ConcurrentLinkedQueue クラスは、反復処理 (例えば for-eachループ文) 中に add や remove などによる変更を許容する Queue の実装です。
非チェック例外の ConcurrentModificationException が発生しないことが保証されています。

void test(Queue<Integer> queue) throws InterruptedException {
    for (int i = 0; i < 5; i++) {
        queue.add(i);
    }

    try (final var executor = Executors.newSingleThreadScheduledExecutor()) {

        executor.schedule(() -> {
            System.out.println("-- add value! --");
            queue.add(9999);
        }, 5, TimeUnit.SECONDS);

        try {
            for (final var value : queue) {
                System.out.println("value = " + value);
                TimeUnit.SECONDS.sleep(2);
            }
        } catch (ConcurrentModificationException e) {
            System.out.println("ConcurrentModificationException!");
        }
    }

    System.out.println("-- end --");
    System.out.println("queue = " + queue);
}

test(new LinkedList<>());

// 結果
// ↓
//value = 0
//value = 1
//value = 2
//-- add value! --
//ConcurrentModificationException!
//-- end --
//queue = [0, 1, 2, 3, 4, 9999]

test(new ConcurrentLinkedQueue<>());

// 結果
// ↓
//value = 0
//value = 1
//value = 2
//-- add value! --
//value = 3
//value = 4
//value = 9999
//-- end --
//queue = [0, 1, 2, 3, 4, 9999]

コンストラクタ

ConcurrentLinkedQueue ()

ConcurrentLinkedQueueを作成します。これは最初は空です。

final var queue = new ConcurrentLinkedQueue<String>();

queue.add("aaa");
queue.add("bbb");
queue.add("ccc");

System.out.println(queue); // [aaa, bbb, ccc]
System.out.println(queue.size()); // 3

System.out.println(queue.remove()); // aaa
System.out.println(queue.remove()); // bbb
System.out.println(queue.remove()); // ccc

System.out.println(queue); // []
System.out.println(queue.isEmpty()); // true

ConcurrentLinkedQueue (Collection<? extends E> c)

指定されたコレクションの要素を初期状態で含む(要素はコレクションのイテレータのトラバーサル順に追加) ConcurrentLinkedQueueを作成します。

final var c = List.of("a", "b", "c");

final var queue = new ConcurrentLinkedQueue<>(c);
System.out.println(queue); // [a, b, c]
System.out.println(queue.size()); // 3

メソッド

boolean add (E e)

キューの末尾に、指定された要素を挿入します。

final var queue = new ConcurrentLinkedQueue<String>();
System.out.println(queue); // []

System.out.println(queue.add("aaa")); // true
System.out.println(queue); // [aaa]

System.out.println(queue.add("bbb")); // true
System.out.println(queue); // [aaa, bbb]

System.out.println(queue.add("ccc")); // true
System.out.println(queue); // [aaa, bbb, ccc]

boolean addAll (Collection<? extends E> c)

指定されたコレクション内のすべての要素を、指定されたコレクションのイテレータによって返される順序でこのキューの最後に追加します。

final var queue = new ConcurrentLinkedQueue<String>();
System.out.println(queue); // []

System.out.println(queue.addAll(List.of("a1"))); // true
System.out.println(queue); // [a1]

System.out.println(queue.addAll(List.of("b1", "b2"))); // true
System.out.println(queue); // [a1, b1, b2]

System.out.println(queue.addAll(List.of("c1", "c2", "c3"))); // true
System.out.println(queue); // [a1, b1, b2, c1, c2, c3]

void clear ()

すべての要素をキューから削除します。

final var queue = new ConcurrentLinkedQueue<String>();
System.out.println(queue); // []

queue.add("a");
System.out.println(queue); // [a]

queue.add("b");
System.out.println(queue); // [a, b]

queue.clear();
System.out.println(queue); // []

boolean contains (Object o)

指定された要素がキューに含まれている場合にtrueを返します。

final var queue = new ConcurrentLinkedQueue<String>();
Collections.addAll(queue, "aaa", "bbb", "ccc");
System.out.println(queue); // [aaa, bbb, ccc]

System.out.println(queue.contains("aaa")); // true
System.out.println(queue.contains("bbb")); // true
System.out.println(queue.contains("XXX")); // false

void forEach (Consumer<? super E> action)

Iterableの各要素に対して指定されたアクションを、すべての要素が処理されるか、アクションが例外をスローするまで実行します。

final var queue = new ConcurrentLinkedQueue<String>();
Collections.addAll(queue, "aaa", "bbb", "ccc");
System.out.println(queue); // [aaa, bbb, ccc]

System.out.println("-- forEach --");
queue.forEach(value -> {
    System.out.println("value = " + value);
});

// 結果
// ↓
//-- forEach --
//value = aaa
//value = bbb
//value = ccc

boolean isEmpty ()

キューに要素がない場合にtrueを返します。

final var queue = new ConcurrentLinkedQueue<String>();
queue.add("a");
queue.add("b");
queue.add("c");

System.out.println(queue); // [a, b, c]
System.out.println(queue.isEmpty()); // false

queue.clear();
System.out.println(queue); // []
System.out.println(queue.isEmpty()); // true

Iterator<E> iterator ()

このキュー内の要素を適切な順序で反復処理するイテレータを返します。

final var queue = new ConcurrentLinkedQueue<String>();
Collections.addAll(queue, "aaa", "bbb", "ccc");
System.out.println(queue); // [aaa, bbb, ccc]

final var iterator = queue.iterator();

System.out.println("-- forEachRemaining --");
iterator.forEachRemaining(value -> {
    System.out.println("value = " + value);
});

// 結果
// ↓
//-- forEachRemaining --
//value = aaa
//value = bbb
//value = ccc

boolean offer (E e)

キューの末尾に、指定された要素を挿入します。

final var queue = new ConcurrentLinkedQueue<String>();
System.out.println(queue); // []

System.out.println(queue.offer("aaa")); // true
System.out.println(queue); // [aaa]

System.out.println(queue.offer("bbb")); // true
System.out.println(queue); // [aaa, bbb]

System.out.println(queue.offer("ccc")); // true
System.out.println(queue); // [aaa, bbb, ccc]

E peek ()

キューの先頭を取得しますが、削除しません。キューが空の場合はnullを返します。

final var queue = new ConcurrentLinkedQueue<String>();
System.out.println(queue); // []

queue.add("aaa");
queue.add("bbb");
queue.add("ccc");

System.out.println(queue.peek()); // aaa
System.out.println(queue); // [aaa, bbb, ccc]

System.out.println(queue.poll()); // aaa
System.out.println(queue); // [bbb, ccc]

System.out.println(queue.peek()); // bbb
System.out.println(queue); // [bbb, ccc]

System.out.println(queue.poll()); // bbb
System.out.println(queue); // [ccc]

System.out.println(queue.peek()); // ccc
System.out.println(queue); // [ccc]

System.out.println(queue.poll()); // ccc
System.out.println(queue); // []

System.out.println(queue.peek()); // null
System.out.println(queue); // []

E poll ()

キューの先頭を取得および削除します。キューが空の場合はnullを返します。

final var queue = new ConcurrentLinkedQueue<String>();
System.out.println(queue); // []

queue.add("aaa");
queue.add("bbb");
queue.add("ccc");

System.out.println(queue); // [aaa, bbb, ccc]

System.out.println(queue.poll()); // aaa
System.out.println(queue); // [bbb, ccc]

System.out.println(queue.poll()); // bbb
System.out.println(queue); // [ccc]

System.out.println(queue.poll()); // ccc
System.out.println(queue); // []

System.out.println(queue.poll()); // null
System.out.println(queue); // []

boolean remove (Object o)

指定された要素の単一のインスタンスがこのキューに存在する場合は、キューから削除します。

final var queue = new ConcurrentLinkedQueue<String>();
Collections.addAll(queue, "aaa", "BBB", "aaa", "BBB", "ccc");
System.out.println(queue); // [aaa, BBB, aaa, BBB, ccc]

System.out.println(queue.remove("aaa")); // true
System.out.println(queue); // [BBB, aaa, BBB, ccc]

System.out.println(queue.remove("BBB")); // true
System.out.println(queue); // [aaa, BBB, ccc]

System.out.println(queue.remove("BBB")); // true
System.out.println(queue); // [aaa, ccc]

System.out.println(queue.remove("XXX")); // false
System.out.println(queue); // [aaa, ccc]

boolean removeAll (Collection<?> c)

指定されたコレクションにも格納されているこのコレクションのすべての要素を削除します(オプションの操作)。

final var src = List.of("a", "b", "a", "b", "A", "B");
System.out.println(src); // [a, b, a, b, A, B]

{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.removeAll(List.of())); // false
    System.out.println(queue); // [a, b, a, b, A, B]
}
{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.removeAll(List.of("a"))); // true
    System.out.println(queue); // [b, b, A, B]
}
{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.removeAll(List.of("a", "b"))); // true
    System.out.println(queue); // [A, B]
}
{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.removeAll(List.of("b", "a"))); // true
    System.out.println(queue); // [A, B]
}
{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.removeAll(List.of("A"))); // true
    System.out.println(queue); // [a, b, a, b, B]
}
{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.removeAll(List.of("X", "Y", "Z"))); // false
    System.out.println(queue); // [a, b, a, b, A, B]
}
{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.removeAll(List.of("A", "X", "Y", "Z"))); // true
    System.out.println(queue); // [a, b, a, b, B]
}

boolean removeIf (Predicate<? super E> filter)

指定された述語を満たすこのコレクションの要素をすべて削除します。

final var queue = new ConcurrentLinkedQueue<String>();
queue.add("aaa");
queue.add("BBB");
queue.add("ccc");
queue.add("DDD");

System.out.println(queue); // [aaa, BBB, ccc, DDD]

final var ret = queue.removeIf(s -> {
    // 大文字であるかチェックします。
    return s.equals(s.toUpperCase());
});

System.out.println(ret); // true
System.out.println(queue); // [aaa, ccc]

final var queue = new ConcurrentLinkedQueue<String>();
queue.add("aaa");
queue.add("bbb");

System.out.println(queue); // [aaa, bbb]

final var ret = queue.removeIf(s -> s.equals(s.toUpperCase()));

System.out.println(ret); // false
System.out.println(queue); // [aaa, bbb]

boolean retainAll (Collection<?> c)

このコレクションにおいて、指定されたコレクションに格納されている要素だけを保持します(オプションの操作)。

final var src = List.of("a", "b", "a", "b", "A", "B");
System.out.println(src); // [a, b, a, b, A, B]

{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.retainAll(List.of())); // true
    System.out.println(queue); // []
}
{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.retainAll(List.of("a", "b"))); // true
    System.out.println(queue); // [a, b, a, b]
}
{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.retainAll(List.of("b", "a"))); // true
    System.out.println(queue); // [a, b, a, b]
}
{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.retainAll(List.of("A"))); // true
    System.out.println(queue); // [A]
}
{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.retainAll(List.of("X", "Y", "Z"))); // true
    System.out.println(queue); // []
}
{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.retainAll(List.of("A", "X", "Y", "Z"))); // true
    System.out.println(queue); // [A]
}
{
    final var queue = new ConcurrentLinkedQueue<>(src);
    System.out.println(queue); // [a, b, a, b, A, B]

    System.out.println(queue.retainAll(List.of("a", "b", "A", "B"))); // false
    System.out.println(queue); // [a, b, a, b, A, B]
}

int size ()

キュー内にある要素の数を返します。

final var queue = new ConcurrentLinkedQueue<String>();
System.out.println(queue); // []
System.out.println(queue.size()); // 0

queue.add("aaa");
System.out.println(queue); // [aaa]
System.out.println(queue.size()); // 1

queue.add("bbb");
System.out.println(queue); // [aaa, bbb]
System.out.println(queue.size()); // 2

queue.add("ccc");
System.out.println(queue); // [aaa, bbb, ccc]
System.out.println(queue.size()); // 3

Spliterator<E> spliterator ()

このキュー内の要素に対するSpliteratorを返します。

final var queue = new ConcurrentLinkedQueue<String>();
Collections.addAll(queue, "aaa", "bbb", "ccc");
System.out.println(queue); // [aaa, bbb, ccc]

final var spliterator = queue.spliterator();

System.out.println("-- forEachRemaining --");
spliterator.forEachRemaining(value -> {
    System.out.println("value = " + value);
});

// 結果
// ↓
//-- forEachRemaining --
//value = aaa
//value = bbb
//value = ccc

Object[] toArray ()

キュー内のすべての要素を適切な順序で含む配列を返します。

final var queue = new ConcurrentLinkedQueue<String>();
Collections.addAll(queue, "a", "b", "c");
System.out.println(queue); // [a, b, c]

final Object[] array = queue.toArray();
System.out.println(Arrays.toString(array)); // [a, b, c]

<T> T[] toArray (T[] a)

キュー内のすべての要素を適切な順序で含む配列を返します。返される配列の実行時の型は、指定された配列の型になります。

final var queue = new ConcurrentLinkedQueue<String>();
Collections.addAll(queue, "a", "b", "c");
System.out.println(queue); // [a, b, c]

final String[] array = queue.toArray(new String[0]);
System.out.println(Arrays.toString(array)); // [a, b, c]

final var queue = new ConcurrentLinkedQueue<String>();
Collections.addAll(queue, "a", "b", "c");
System.out.println(queue); // [a, b, c]

{
    final String[] array = new String[3];
    System.out.println(Arrays.toString(array)); // [null, null, null]

    final var ret = queue.toArray(array);
    System.out.println(Arrays.toString(array)); // [a, b, c]
    System.out.println(Arrays.toString(ret)); // [a, b, c]
}
{
    final String[] array = new String[5];
    System.out.println(Arrays.toString(array)); // [null, null, null, null, null]

    final var ret = queue.toArray(array);
    System.out.println(Arrays.toString(array)); // [a, b, c, null, null]
    System.out.println(Arrays.toString(ret)); // [a, b, c, null, null]
}
{
    final String[] array = new String[1];
    System.out.println(Arrays.toString(array)); // [null]

    final var ret = queue.toArray(array);
    System.out.println(Arrays.toString(array)); // [a]
    System.out.println(Arrays.toString(ret)); // [a, b, c]
}