Kotlinで学ぶ排他制御の15選の実践方法

はじめに
●Kotlinとは
- ○Kotlinの基本的な特徴
●排他制御の基礎
- ○排他制御とは
- ○排他制御の重要性
●Kotlinでの排他制御の使い方
○サンプルコード1：基本的なsynchronizedの使用
●Kotlinでの排他制御の応用例
●排他制御の注意点と対処法
●Kotlinでの排他制御のカスタマイズ方法
- ○サンプルコード14：カスタムLockの作成
- ○サンプルコード15：条件変数を用いた詳細な同期制御
まとめ

はじめに

あなたがこの記事を手に取ったのは、Kotlinでの排他制御の実践方法を深く知りたいからではないでしょうか。

あるいは、既に何かのアプリケーションを開発中で、正確な排他制御の実装が必要となってきたのかもしれません。

この記事を読めば、Kotlinでの排他制御をしっかりと理解し、効果的に実装することができるようになります。

排他制御は、マルチスレッド環境でのデータの整合性を保つための重要な技術です。

しかし、実際には多くの開発者がその実装に苦労しています。

Kotlinを利用すれば、より簡単かつ安全に排他制御を実装することができます。

この記事では、Kotlinの基本から、排他制御の具体的な実装方法まで、詳しく解説していきます。

●Kotlinとは

Kotlinは、2011年にJetBrains社によって発表された、静的型付けのプログラミング言語です。

Javaに似ているため、Javaの知識がある人には非常に親しみやすい言語と言えます。

しかし、KotlinにはJavaにはない便利な機能や、よりシンプルな文法が多く備わっています。

○Kotlinの基本的な特徴

Kotlinは、特にAndroidアプリ開発において、Javaの後継言語として注目を浴びています。

ここではに、Kotlinの主な特徴をいくつか挙げます。

Null安全：Kotlinは、nullの扱いに関して非常に厳格です。これにより、NullPointerExeptionのようなランタイムエラーを大幅に減少させることが可能です。
拡張関数：既存のクラスに新しいメソッドを追加することなく、そのクラスの振る舞いを拡張することができます。
データクラス：一般的なデータホルダーとしてのクラスを非常にシンプルに定義できます。
ラムダ式と高階関数：関数型プログラミングの要素が多数取り入れられており、シンプルかつ強力なコードを書くことができます。

このように、Kotlinは現代のプログラミングに求められる多くの要素を持っており、開発の生産性や安全性を大幅に向上させることができます。

●排他制御の基礎

排他制御はマルチスレッドプログラミングの中で非常に重要な役割を持っています。

正確に排他制御を行わないと、データの不整合や予期しないバグの原因となり得ます。

ここでは、排他制御の基本について詳しく解説していきます。

○排他制御とは

排他制御、別名ミューテックス(mutex)とは、同時に複数のスレッドがデータやリソースにアクセスすることを防ぐ制御のことを指します。

例えば、銀行の口座のようなものを考えてみましょう。

複数の人が同時に同じ口座にアクセスして預金や引き出しを行うと、正確な金額を計算できなくなる可能性があります。

このような問題を防ぐために、排他制御を使用して一度に一つのスレッドのみがリソースにアクセスできるようにします。

○排他制御の重要性

排他制御はデータの整合性を保つために不可欠です。

マルチスレッド環境では、複数のスレッドが同時にデータにアクセスし、それを変更することが可能です。

しかし、その際に適切な制御を行わないと、データの状態が予期せぬ形になってしまうことがあります。

これは、プログラムの動作が不安定になるだけでなく、重大なデータの損失を引き起こす可能性もあります。

そのため、排他制御はマルチスレッドプログラミングの基礎となる技術の一つと言えるでしょう。

●Kotlinでの排他制御の使い方

Kotlinはその簡潔さとJavaとの互換性から、Android開発を中心に多くの開発者に支持されています。

ここでは、Kotlinを使用して排他制御を実装する方法を取り上げます。

○サンプルコード1：基本的なsynchronizedの使用

KotlinでもJavaと同様にsynchronizedキーワードを使用することで、排他制御を実装することができます。

class Counter {
    private var count = 0

    fun increment() {
        synchronized(this) {
            count++
            println("現在のカウント値：$count")
        }
    }
}

このコードでは、Counterクラス内のcount変数をインクリメントする際に、synchronizedブロックを使用しています。

これにより、複数のスレッドから同時にincrement関数が呼び出された場合でも、count変数へのアクセスが排他的になり、データの不整合を防ぐことができます。

○サンプルコード2：ReentrantLockの基本的な使い方

Kotlinでは、Javaのjava.util.concurrent.locksパッケージにあるReentrantLockを利用することもできます。

ここでは、ReentrantLockを用いた排他制御の基本的な例を紹介します。

import java.util.concurrent.locks.ReentrantLock

class SafeCounter {
    private var count = 0
    private val lock = ReentrantLock()

    fun increment() {
        lock.lock()
        try {
            count++
            println("現在のカウント値：$count")
        } finally {
            lock.unlock()
        }
    }
}

このコードでは、ReentrantLockのインスタンスをlockという名前で保持し、increment関数内でlockとunlockを使用して排他制御を行っています。

この方法を使用することで、より高度な排他制御や、より柔軟なロックの制御が可能となります。

○サンプルコード3：ReadWriteLockの実装例

多くの同時処理のアプリケーションでは、データへの読み取りが書き込みよりも頻繁に行われることが多いです。

このような場合、読み取りと書き込みのロックを区別することで、効率的な排他制御を実現することができます。

ReadWriteLockはこの目的で使用されるツールの一つです。

Kotlinでは、ReadWriteLockを使用して、読み取り専用のロックと書き込み専用のロックを別々に管理することができます。

import java.util.concurrent.locks.ReentrantReadWriteLock

class DataStore {
    private var data: Int = 0
    private val rwLock = ReentrantReadWriteLock()
    private val readLock = rwLock.readLock()
    private val writeLock = rwLock.writeLock()

    fun readData(): Int {
        readLock.lock()
        try {
            println("データを読み取りました: $data")
            return data
        } finally {
            readLock.unlock()
        }
    }

    fun writeData(newValue: Int) {
        writeLock.lock()
        try {
            data = newValue
            println("データを更新しました: $data")
        } finally {
            writeLock.unlock()
        }
    }
}

このコードでは、ReentrantReadWriteLockのインスタンスを生成し、そのインスタンスから読み取り専用のロックと書き込み専用のロックを取得しています。

readData関数は、読み取り専用のロックを取得してデータを読み取り、writeData関数は書き込み専用のロックを取得してデータを更新しています。

このようにして、複数のスレッドが同時にデータを読み取ることは可能でありながら、書き込み時には排他制御が適用されるため、データの不整合を防ぐことができます。

○サンプルコード4：Semaphoreの使用例

Semaphoreは、指定された数のスレッドのみが同時にリソースにアクセスできるように制限する排他制御のツールです。

これにより、リソースへの同時アクセスを制限して、リソースの過度な使用や競合を防ぐことができます。

import java.util.concurrent.Semaphore

class LimitedResource {
    private val semaphore = Semaphore(3)  // 同時に3つのスレッドのみがアクセス可能

    fun use() {
        semaphore.acquire()
        try {
            println("リソースを使用しています。")
            // ここでリソースを使用する処理を行う
        } finally {
            println("リソースの使用を終了しました。")
            semaphore.release()
        }
    }
}

このコードは、同時に3つのスレッドのみがリソースを使用できるように、Semaphore(3)を指定してSemaphoreのインスタンスを作成しています。

use関数内で、acquireメソッドを呼び出すことで、利用可能な許可がある場合はリソースを使用し、利用可能な許可がない場合は待機します。

リソースの使用が終了したら、releaseメソッドを呼び出して許可を返します。

これにより、指定された数以上のスレッドが同時にリソースを使用することが防がれます。

○サンプルコード5：CountDownLatchの活用方法

CountDownLatchは、一つまたはそれ以上のスレッドが、他の一つまたはそれ以上のスレッドが指定されたタスクを完了するのを待つための同期補助ツールです。

このツールは、特定の数のイベントが発生するのを待って、それらのイベントがすべて発生した後に、待機していたスレッドが進行を再開するというパターンで使われます。

KotlinでCountDownLatchを活用する一例を見てみましょう。

import java.util.concurrent.CountDownLatch

fun main() {
    val latch = CountDownLatch(3)

    for (i in 1..3) {
        Thread {
            println("タスク$i 開始")
            Thread.sleep((i * 1000).toLong())
            println("タスク$i 完了")
            latch.countDown()
        }.start()
    }

    println("すべてのタスクの完了を待っています…")
    latch.await()
    println("すべてのタスクが完了しました。")
}

このコードでは、3つの異なるタスクを実行するスレッドを作成しています。

それぞれのタスクは、そのタスクが完了するとlatch.countDown()を呼び出して、CountDownLatchのカウントをデクリメントします。

メインスレッドは、latch.await()を使って、3つのタスクがすべて完了するのを待ちます。

このコードを実行すると、次のような出力結果が得られることが予想されます。

タスク1 開始
タスク2 開始
タスク3 開始
すべてのタスクの完了を待っています…
(約1秒後) タスク1 完了
(約2秒後) タスク2 完了
(約3秒後) タスク3 完了
すべてのタスクが完了しました。

CountDownLatchは、異なるスレッドが同時に複数のタスクを実行している場合に、特定のポイントで同期をとるための強力なツールです。

例えば、初期データのロードや外部サービスへの複数のリクエストを並列して実行し、すべてのリクエストが完了するのを待つ場面などで利用することができます。

●Kotlinでの排他制御の応用例

排他制御の基本的な概念を把握したところで、Kotlinにおける実践的な排他制御の応用方法を学んでいきましょう。

ここでは、複数のスレッドがデータを安全に更新する方法や、並列処理の同期化に関する具体的なサンプルコードを取り上げます。

○サンプルコード6：複数スレッドでのデータの安全な更新

複数のスレッドが同時にアクセスするデータを安全に更新するには、排他制御が必要です。

ここでは、ReentrantLockを使用して複数のスレッドからデータの更新を行う際の例を紹介します。

import java.util.concurrent.locks.ReentrantLock

val lock = ReentrantLock()
var sharedData = 0

fun updateData() {
    lock.lock()
    try {
        // データの更新処理
        sharedData++
        println("データ更新: $sharedData")
    } finally {
        lock.unlock()
    }
}

fun main() {
    val threads = List(10) {
        Thread {
            updateData()
        }
    }

    threads.forEach { it.start() }
    threads.forEach { it.join() }
}

このコードは、ReentrantLockを使用して共有データsharedDataの更新処理を排他制御しています。

このようにして、複数のスレッドが同時にデータを更新する場合でも、データの整合性を保つことができます。

○サンプルコード7：並列処理の同期化

並列処理の中で、特定の処理の完了を待ちたい場合も排他制御が役立ちます。

ここでは、CountDownLatchを使用して、複数のスレッドでの処理が完了するのを待つ例を紹介します。

import java.util.concurrent.CountDownLatch

fun task(latch: CountDownLatch, taskId: Int) {
    Thread.sleep((taskId * 1000).toLong())
    println("タスク$taskId 完了")
    latch.countDown()
}

fun main() {
    val latch = CountDownLatch(3)

    for (i in 1..3) {
        Thread {
            task(latch, i)
        }.start()
    }

    println("すべてのタスクの完了を待機中…")
    latch.await()
    println("全タスク完了")
}

このコードは、3つのタスクが完了するのをメインスレッドで待っています。

CountDownLatchを使うことで、指定した数のタスクが完了するまでメインスレッドの処理をブロックすることができます。

○サンプルコード8：リソースの制限付きアクセス

リソースへのアクセスを制限する必要が生じる場面は多々あります。

たとえば、一度に多数のユーザーが同時にアクセスすることで、システムに過度な負荷がかかり、性能が低下する可能性があります。

このようなシチュエーションを回避するためには、一度にアクセス可能なユーザー数を制限する必要があります。

Kotlinでは、このようなアクセス制限を実現するためにSemaphoreクラスを使用することができます。

Semaphoreは、指定された数の許可を表し、アクセスを行いたい場合には許可を取得する必要があります。

ここでは、一度に3つのリソースのみへのアクセスを許可するSemaphoreの使用例を紹介します。

import java.util.concurrent.Semaphore

val semaphore = Semaphore(3)  // 3つの許可を持つセマフォ

fun accessResource(id: Int) {
    semaphore.acquire()  // 許可を取得
    try {
        println("リソースにアクセス中: $id")
        Thread.sleep(2000)  // 2秒間リソースを使用する想定
    } finally {
        println("リソースの使用完了: $id")
        semaphore.release()  // 許可を返却
    }
}

fun main() {
    for (i in 1..10) {
        Thread {
            accessResource(i)
        }.start()
    }
}

このコードでは、semaphore.acquire()メソッドを使用してリソースへのアクセス許可を取得しています。

許可がない場合、アクセスを待機する必要があります。

一方、リソースの使用が完了したら、semaphore.release()メソッドを使用して許可を返却します。

このコードを実行すると、3つのスレッドが同時にリソースにアクセスします。

それ以外のスレッドは、リソースが使用可能になるまで待機します。

○サンプルコード9：複雑なロジックの同期

リソースへのアクセス制限だけでなく、複雑なロジックを持つアプリケーションでは、特定のタスクが特定の順番で実行されるように同期を取る必要があります。

ここでは、複数のステージを持つタスクを順番に実行するための例を紹介します。

import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock

val lock = ReentrantLock()
var currentStage = 1

fun executeTask(taskId: Int, requiredStage: Int) {
    lock.withLock {
        while (currentStage != requiredStage) {
            lock.wait()
        }
        println("タスク $taskId: ステージ $requiredStage 実行中")
        currentStage++
        lock.notifyAll()
    }
}

fun main() {
    val tasks = listOf(
        Pair(1, 3),
        Pair(2, 1),
        Pair(3, 2)
    )

    tasks.forEach { (taskId, stage) ->
        Thread {
            executeTask(taskId, stage)
        }.start()
    }
}

このコードでは、ReentrantLockとそのwait/notifyAllメソッドを使用してタスクを同期しています。

指定されたステージのタスクが完了すると、次のステージのタスクが実行されます。

このコードを実行すると、タスク2、タスク3、タスク1の順に実行されることが確認できます。

○サンプルコード10：非同期タスクの順番保証

非同期タスクは、多くのシステムやアプリケーションにおいて、効率的にタスクを実行するための手法として用いられます。

しかし、特定の順番でタスクを実行する必要がある場合、単純に非同期で実行するだけでは順番の保証が難しくなります。

Kotlinでは、このような非同期タスクの順番を保証するための方法として、Deferredというクラスを提供しています。

ここでは、Deferredを使用して非同期タスクの実行順を保証するサンプルコードを紹介します。

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

suspend fun taskA(): Int {
    delay(1000)
    println("タスクA完了")
    return 1
}

suspend fun taskB(resultFromA: Int): Int {
    delay(500)
    println("タスクB完了")
    return resultFromA * 10
}

suspend fun taskC(resultFromB: Int) {
    delay(100)
    println("タスクC完了, 最終結果: $resultFromB")
}

fun main() = runBlocking {
    val time = measureTimeMillis {
        val resultA = async { taskA() }
        val resultB = async { taskB(resultA.await()) }
        taskC(resultB.await())
    }
    println("全タスク完了, 所要時間: $time ms")
}

このコードでは、taskA、taskB、taskCの3つの非同期タスクがあり、taskBはtaskAの結果を、taskCはtaskBの結果を使用しています。

async関数を使って非同期タスクを開始し、await関数を使って結果を取得します。

この方法により、非同期であるにも関わらず、タスクの実行順を保証することができます。

このコードを実行すると、まず「タスクA完了」、次に「タスクB完了」、最後に「タスクC完了」と表示されることが確認できます。

全てのタスクが正確な順番で完了した後、全体の所要時間も表示されます。

●排他制御の注意点と対処法

排他制御は多くのシステムやアプリケーションにおいて重要な役割を果たしています。

しかし、正しく実装されない場合、さまざまな問題や不具合を引き起こす可能性があります。

ここでは、排他制御の実装に関する主な注意点とそれらの問題を回避または対処するための方法を取り上げます。

○サンプルコード11：デッドロックの回避

デッドロックは、複数のスレッドがリソースを待っている間にお互いの進行が停止してしまう現象です。

ここではは、デッドロックが発生しやすい状況を示すサンプルコードと、その回避策を採用したコードを紹介します。

import java.util.concurrent.locks.ReentrantLock

val lock1 = ReentrantLock()
val lock2 = ReentrantLock()

fun threadA() {
    lock1.lock()
    Thread.sleep(100)
    lock2.lock()
    println("スレッドA完了")
    lock2.unlock()
    lock1.unlock()
}

fun threadB() {
    lock2.lock()
    Thread.sleep(100)
    lock1.lock()
    println("スレッドB完了")
    lock1.unlock()
    lock2.unlock()
}

このコードのthreadAとthreadBは、異なる順番でロックを取得しようとするため、デッドロックが発生する可能性があります。

デッドロックを回避する一つの方法は、ロックを取得する順序を全てのスレッドで同じにすることです。

そのため、上記のコードの問題を回避するためには、threadAとthreadBのロック取得の順序を同じにすれば良いでしょう。

○サンプルコード12：スターベーション問題への対応

スターベーションは、あるスレッドが必要なリソースのアクセスを得られないまま、他のスレッドによって繰り返しリソースが利用される状況を指します。

下記のコードは、スターベーション問題を表すものです。

import java.util.concurrent.locks.ReentrantLock

val sharedLock = ReentrantLock()

fun highPriorityTask() {
    while (true) {
        sharedLock.lock()
        // 重要な処理
        println("高優先度タスク実行中")
        sharedLock.unlock()
        Thread.sleep(10)
    }
}

fun lowPriorityTask() {
    while (true) {
        sharedLock.lock()
        // 通常の処理
        println("低優先度タスク実行中")
        sharedLock.unlock()
        Thread.sleep(100)
    }
}

このコードでは、highPriorityTaskが頻繁にロックを取得するため、lowPriorityTaskがロックを取得する機会がほとんどなくなり、スターベーションが発生します。

スターベーション問題への対応としては、公平なロックの取得を考慮することが重要です。

ReentrantLockのコンストラクタにtrueを渡すことで、公平性を保証するロックを作成することができます。

○サンプルコード13：競合状態の回避テクニック

競合状態は、複数のスレッドが同時に同じデータにアクセスし、不具合や予期しない結果を引き起こす状態を指します。

下記のサンプルコードは、競合状態が発生する可能性のあるシナリオを表しています。

var counter = 0

fun increment() {
    for (i in 1..1000) {
        counter++
    }
}

fun main() {
    val thread1 = Thread { increment() }
    val thread2 = Thread { increment() }
    thread1.start()
    thread2.start()
    thread1.join()
    thread2.join()
    println(counter) 
}

このコードを実行すると、期待される値である2000よりも小さい値が出力される可能性が高いです。

なぜなら、counter++の操作はアトミックではないため、競合状態が発生するからです。

競合状態を回避するための一つの方法は、synchronizedブロックを使用して、同時に一つのスレッドだけが特定のコードを実行できるようにすることです。

●Kotlinでの排他制御のカスタマイズ方法

Kotlinでの排他制御の実装は、多様なニーズに応えるためのカスタマイズが可能です。

ここでは、標準の排他制御手法をカスタマイズし、より高度な制御を行うための方法を2つの実践的なサンプルコードを交えて解説します。

○サンプルコード14：カスタムLockの作成

KotlinではReentrantLockのような既存のロックをカスタマイズして、独自のロック機構を作成することができます。

下記のサンプルコードでは、タイムアウト付きのカスタムロックを実装しています。

import java.util.concurrent.TimeUnit
import java.util.concurrent.locks.ReentrantLock

class TimeoutLock {
    private val internalLock = ReentrantLock()

    fun tryLockWithTimeout(timeout: Long, unit: TimeUnit): Boolean {
        return internalLock.tryLock(timeout, unit)
    }

    fun unlock() {
        if (internalLock.isHeldByCurrentThread) {
            internalLock.unlock()
        }
    }
}

fun main() {
    val lock = TimeoutLock()

    val task = Thread {
        if (lock.tryLockWithTimeout(2, TimeUnit.SECONDS)) {
            println("ロックを取得しました。")
            Thread.sleep(5000)
            lock.unlock()
        } else {
            println("指定した時間内にロックを取得できませんでした。")
        }
    }

    task.start()
}

このコードでは、TimeoutLockクラスを使って2秒間ロックの取得を試みます。

指定時間内にロックを取得できなければ、メッセージが表示されます。

○サンプルコード15：条件変数を用いた詳細な同期制御

条件変数は、特定の条件が満たされるまでスレッドの実行を一時停止し、条件が満たされたときに再開するための仕組みです。

下記のサンプルコードでは、条件変数を用いて生産者と消費者の問題を解決しています。

import java.util.LinkedList
import java.util.concurrent.locks.ReentrantLock
import java.util.concurrent.locks.Condition

class SharedQueue<T> {
    private val lock = ReentrantLock()
    private val notEmpty: Condition = lock.newCondition()
    private val queue = LinkedList<T>()

    fun put(item: T) {
        lock.lock()
        queue.add(item)
        notEmpty.signal()
        lock.unlock()
    }

    fun take(): T {
        lock.lock()
        while (queue.isEmpty()) {
            notEmpty.await()
        }
        val item = queue.removeFirst()
        lock.unlock()
        return item
    }
}

fun main() {
    val sharedQueue = SharedQueue<Int>()

    val producer = Thread {
        for (i in 1..5) {
            sharedQueue.put(i)
            println("生産: $i")
            Thread.sleep(1000)
        }
    }

    val consumer = Thread {
        for (i in 1..5) {
            val value = sharedQueue.take()
            println("消費: $value")
            Thread.sleep(1500)
        }
    }

    producer.start()
    consumer.start()
}

このコードでは、生産者は1秒ごとに数字を生成し、SharedQueueに追加します。

一方、消費者は1.5秒ごとにSharedQueueから数字を取り出します。

Queueが空の場合、消費者はnotEmpty.await()で待機し、新しいアイテムが追加されると再開されます。