MENU
  1. ホーム
  2. Python
  3. スレッド
  4. 【Python】マルチスレッドで複数のスレッドを同期させる方法

【Python】マルチスレッドで複数のスレッドを同期させる方法

スレッド

Pythonでマルチスレッドプログラミングを行うと、複数の作業を同時に実行できるようになります。

しかし、スレッドが同時に動くことでデータの競合やデッドロックといった問題が発生することもあります。

この記事では、Pythonのthreadingモジュールを使ってスレッドを作成し、同期させる方法をわかりやすく解説します。

具体的なコード例を交えながら、ロックやセマフォ、イベントなどの同期手法を学び、スレッド間の通信やリソース管理のベストプラクティスを紹介します。

目次から探す

Pythonでのマルチスレッドの実装

Pythonでマルチスレッドを実装するためには、標準ライブラリのthreadingモジュールを使用します。

このモジュールを使うことで、複数のスレッドを簡単に作成し、並行して実行することができます。

以下では、threadingモジュールの基本的な使い方と、スレッドの作成および実行方法について詳しく解説します。

threadingモジュールの紹介

threadingモジュールは、Pythonでマルチスレッドプログラミングを行うための標準ライブラリです。

このモジュールを使用することで、スレッドの作成、管理、同期を簡単に行うことができます。

threadingモジュールには、以下のような主要なクラスと関数が含まれています。

クラス名説明
Threadクラス新しいスレッドを作成するためのクラス
Lockクラススレッド間の排他制御を行うためのクラス
RLockクラス再帰的なロックを提供するクラス
Conditionクラススレッド間の条件変数を提供するクラス
Semaphoreクラススレッド間のリソース制限を行うためのクラス
Eventクラススレッド間のイベント通知を行うためのクラス

基本的なスレッドの作成と実行

スレッドの作成

スレッドを作成するためには、threading.Threadクラスを使用します。

Threadクラスのインスタンスを作成する際には、ターゲットとなる関数を引数として渡します。

以下に、基本的なスレッドの作成方法を示します。

import threading
def print_numbers():
    for i in range(5):
        print(i)
# スレッドの作成
thread = threading.Thread(target=print_numbers)

上記の例では、print_numbersという関数をターゲットとして、新しいスレッドを作成しています。

このスレッドは、print_numbers関数を実行するために使用されます。

スレッドの開始と終了

スレッドを開始するためには、Threadクラスのstartメソッドを呼び出します。

スレッドが開始されると、ターゲット関数が並行して実行されます。

スレッドの終了を待つためには、joinメソッドを使用します。

以下に、スレッドの開始と終了の方法を示します。

import threading
def print_numbers():
    for i in range(5):
        print(i)
# スレッドの作成
thread = threading.Thread(target=print_numbers)
# スレッドの開始
thread.start()
# スレッドの終了を待つ
thread.join()
print("スレッドが終了しました")

上記の例では、thread.start()を呼び出すことでスレッドが開始され、print_numbers関数が並行して実行されます。

thread.join()を呼び出すことで、メインスレッドは新しいスレッドの終了を待ちます。

新しいスレッドが終了すると、スレッドが終了しましたというメッセージが表示されます。

このようにして、threadingモジュールを使用することで、Pythonで簡単にマルチスレッドプログラミングを行うことができます。

次のセクションでは、スレッドの同期方法について詳しく解説します。

スレッドの同期とは

マルチスレッドプログラミングでは、複数のスレッドが同時に実行されるため、データの整合性や一貫性を保つためにスレッド間の同期が必要です。

同期を適切に行わないと、データ競合やデッドロックなどの問題が発生する可能性があります。

同期の必要性

スレッドの同期が必要な理由は主に以下の通りです。

データ競合の防止

複数のスレッドが同じデータに同時にアクセスし、変更を加えると、データの整合性が失われる可能性があります。

例えば、銀行口座の残高を更新する際に、複数のスレッドが同時に残高を変更すると、最終的な残高が正しくない値になることがあります。

これを防ぐために、スレッド間でデータのアクセスを同期する必要があります。

デッドロックの回避

デッドロックとは、複数のスレッドが互いにリソースを待ち続ける状態のことです。

これにより、プログラムが停止してしまうことがあります。

適切な同期を行うことで、デッドロックの発生を防ぐことができます。

スレッド間の通信

スレッド間でデータをやり取りする際に、データの受け渡しが正しく行われるようにするために同期が必要です。

例えば、あるスレッドがデータを生成し、別のスレッドがそのデータを消費する場合、データの生成と消費のタイミングを同期する必要があります。

同期の基本概念

スレッドの同期を実現するためには、いくつかの基本的な概念と技術があります。

以下に代表的なものを紹介します。

ロック (Lock)

ロックは、スレッドが特定のリソースにアクセスする際に、そのリソースを他のスレッドから保護するための仕組みです。

ロックを取得したスレッドだけがリソースにアクセスでき、他のスレッドはロックが解放されるまで待機します。

Pythonでは、threading.Lockクラスを使用してロックを実現できます。

再帰ロック (RLock)

再帰ロックは、同じスレッドが複数回ロックを取得できるようにするための仕組みです。

通常のロックでは、同じスレッドが再度ロックを取得しようとするとデッドロックが発生しますが、再帰ロックを使用することでこれを防ぐことができます。

Pythonでは、threading.RLockクラスを使用して再帰ロックを実現できます。

条件変数 (Condition)

条件変数は、スレッド間で特定の条件が満たされるまで待機するための仕組みです。

あるスレッドが条件を満たしたときに、他のスレッドに通知を送ることができます。

Pythonでは、threading.Conditionクラスを使用して条件変数を実現できます。

セマフォ (Semaphore)

セマフォは、特定のリソースにアクセスできるスレッドの数を制限するための仕組みです。

例えば、データベース接続の数を制限する場合に使用されます。

Pythonでは、threading.Semaphoreクラスを使用してセマフォを実現できます。

イベント (Event)

イベントは、スレッド間で特定の状態を通知するための仕組みです。

あるスレッドがイベントを設定すると、他のスレッドがそのイベントを検知して動作を開始します。

Pythonでは、threading.Eventクラスを使用してイベントを実現できます。

これらの同期手法を適切に組み合わせることで、マルチスレッドプログラミングにおけるデータの整合性や一貫性を保ち、効率的なスレッド間の通信を実現することができます。

Pythonでのスレッド同期の方法

Pythonでスレッドを同期させるためには、いくつかの方法があります。

ここでは、代表的な同期手法であるロック、再帰ロック、条件変数、セマフォ、イベントについて詳しく解説します。

ロック(Lock)オブジェクト

ロックの基本的な使い方

ロックは、スレッドが共有リソースにアクセスする際に競合を防ぐための基本的な同期手法です。

PythonのthreadingモジュールにはLockクラスが用意されています。

以下は、ロックを使った基本的な例です。

import threading
# 共有リソース
shared_resource = 0
# ロックオブジェクトの作成
lock = threading.Lock()
def increment():
    global shared_resource
    for _ in range(100000):
        # ロックを取得
        lock.acquire()
        shared_resource += 1
        # ロックを解放
        lock.release()
# スレッドの作成
thread1 = threading.Thread(target=increment)
thread2 = threading.Thread(target=increment)
# スレッドの開始
thread1.start()
thread2.start()
# スレッドの終了を待つ
thread1.join()
thread2.join()
print(f"最終的な共有リソースの値: {shared_resource}")

この例では、2つのスレッドが同時にshared_resourceをインクリメントしています。

ロックを使用することで、同時にアクセスすることによる競合を防いでいます。

ロックのデッドロック問題とその回避方法

ロックを使用する際に注意しなければならないのがデッドロックです。

デッドロックは、複数のスレッドが互いにロックを待ち続ける状態を指します。

以下はデッドロックの例です。

import threading

# ロックオブジェクトを作成
lock1 = threading.Lock()
lock2 = threading.Lock()

# スレッド1のタスクを定義
def thread1_task():
    lock1.acquire()  # lock1を取得
    lock2.acquire()  # lock2を取得
    print("Thread 1 is running")
    lock2.release()  # lock2を解放
    lock1.release()  # lock1を解放

# スレッド2のタスクを定義
def thread2_task():
    lock2.acquire()  # lock2を取得
    lock1.acquire()  # lock1を取得
    print("Thread 2 is running")
    lock1.release()  # lock1を解放
    lock2.release()  # lock2を解放

# スレッド1を作成
thread1 = threading.Thread(target=thread1_task)
# スレッド2を作成
thread2 = threading.Thread(target=thread2_task)

# スレッド1を開始
thread1.start()
# スレッド2を開始
thread2.start()

# スレッド1の完了を待機
thread1.join()
# スレッド2の完了を待機
thread2.join()

この例では、thread1_taskがlock1を取得し、thread2_taskがlock2を取得した後、互いのロックを待ち続けるためデッドロックが発生します。

デッドロックを回避するためには、以下のような方法があります。

  1. ロックの取得順序を統一する。
  2. タイムアウトを設定する。 
import threading

# ロックオブジェクトを作成
lock1 = threading.Lock()
lock2 = threading.Lock()

# スレッド1のタスクを定義
def thread1_task():
    while True:
        # lock1を取得しようと試みる
        if lock1.acquire(timeout=1):
            # lock2を取得しようと試みる
            if lock2.acquire(timeout=1):
                print("Thread 1 is running")
                lock2.release()  # lock2を解放
                lock1.release()  # lock1を解放
                break  # ループを抜ける
            lock1.release()  # lock2が取得できなかった場合、lock1を解放

# スレッド2のタスクを定義
def thread2_task():
    while True:
        # lock2を取得しようと試みる
        if lock2.acquire(timeout=1):
            # lock1を取得しようと試みる
            if lock1.acquire(timeout=1):
                print("Thread 2 is running")
                lock1.release()  # lock1を解放
                lock2.release()  # lock2を解放
                break  # ループを抜ける
            lock2.release()  # lock1が取得できなかった場合、lock2を解放

# スレッド1を作成
thread1 = threading.Thread(target=thread1_task)
# スレッド2を作成
thread2 = threading.Thread(target=thread2_task)

# スレッド1を開始
thread1.start()
# スレッド2を開始
thread2.start()

# スレッド1の完了を待機
thread1.join()
# スレッド2の完了を待機
thread2.join()

RLock(再帰ロック)

RLockの基本的な使い方

RLock(再帰ロック)は、同じスレッドが複数回ロックを取得できるようにするためのロックです。

Lockオブジェクトとは異なり、RLockは同じスレッドが再度ロックを取得することができます。

以下は、RLockを使った基本的な例です。

import threading

# 再帰的ロックオブジェクトを作成
rlock = threading.RLock()

# 再帰的タスクを定義
def recursive_task(n):
    rlock.acquire()  # ロックを取得
    print(f"Entering level {n}")
    if n > 0:
        recursive_task(n - 1)  # 再帰呼び出し
    print(f"Exiting level {n}")
    rlock.release()  # ロックを解放

# スレッドを作成
thread = threading.Thread(target=recursive_task, args=(3,))

# スレッドを開始
thread.start()

# スレッドの完了を待機
thread.join()

この例では、再帰的に関数を呼び出しながらロックを取得しています。

RLockを使用することで、同じスレッドが再度ロックを取得できるようになっています。

RLockの利点と注意点

RLockの利点は、再帰的な関数呼び出しや複数のメソッドが同じロックを共有する場合に便利です。

しかし、RLockを多用するとデバッグが難しくなるため、必要な場合にのみ使用することが推奨されます。

Conditionオブジェクト

Conditionの基本的な使い方

Conditionオブジェクトは、スレッド間の通信を行うための同期プリミティブです。

Conditionオブジェクトを使用すると、スレッドは特定の条件が満たされるまで待機し、条件が満たされたときに通知を受け取ることができます。

以下は、Conditionオブジェクトを使った基本的な例です。

import threading

# 条件変数オブジェクトを作成
condition = threading.Condition()
shared_resource = []

# 生産者タスクを定義
def producer():
    global shared_resource
    with condition:
        for i in range(5):
            shared_resource.append(i)
            print(f"Produced: {i}")
            condition.notify()  # 消費者に通知
            condition.wait()  # 消費者がアイテムを消費するのを待つ

# 消費者タスクを定義
def consumer():
    global shared_resource
    with condition:
        while True:
            condition.wait()  # 生産者からの通知を待つ
            if shared_resource:
                item = shared_resource.pop(0)
                print(f"Consumed: {item}")
                condition.notify()  # 生産者に通知
            else:
                break  # リソースが空の場合、ループを抜ける

# 生産者スレッドを作成
producer_thread = threading.Thread(target=producer)
# 消費者スレッドを作成
consumer_thread = threading.Thread(target=consumer)

# スレッドを開始
producer_thread.start()
consumer_thread.start()

# スレッドの完了を待機
producer_thread.join()
consumer_thread.join()

この例では、producerスレッドがアイテムを生成し、consumerスレッドがアイテムを消費します。

Conditionオブジェクトを使用することで、スレッド間の通信がスムーズに行われます。

Conditionを使ったスレッド間の通信

Conditionオブジェクトを使うことで、スレッド間の通信が容易になります。

例えば、生産者-消費者問題を解決する際に役立ちます。

Conditionオブジェクトを使用することで、スレッドは特定の条件が満たされるまで待機し、条件が満たされたときに通知を受け取ることができます。

Semaphoreオブジェクト

Semaphoreの基本的な使い方

Semaphoreオブジェクトは、特定の数のスレッドが同時にリソースにアクセスできるようにするための同期プリミティブです。

Semaphoreは、リソースの数をカウントし、リソースが利用可能な場合にスレッドがアクセスできるようにします。

以下は、Semaphoreを使った基本的な例です。

import threading
import time

# セマフォオブジェクトを作成、最大2つのスレッドが同時に実行できる
semaphore = threading.Semaphore(2)

# タスクを定義
def task(name):
    semaphore.acquire()  # セマフォを取得
    print(f"{name} is running")
    time.sleep(2)  # タスクの実行（2秒間スリープ）
    print(f"{name} is done")
    semaphore.release()  # セマフォを解放

threads = []

# 5つのスレッドを作成して開始
for i in range(5):
    thread = threading.Thread(target=task, args=(f"Thread-{i}",))
    threads.append(thread)
    thread.start()

# すべてのスレッドの完了を待機
for thread in threads:
    thread.join()

この例では、最大2つのスレッドが同時に実行されるようにSemaphoreを設定しています。

スレッドがリソースを取得し、処理が完了するとリソースを解放します。

Semaphoreの応用例

Semaphoreは、リソースの数を制限する場合に非常に便利です。

例えば、データベース接続の数を制限したり、同時に実行できるタスクの数を制限する場合に使用されます。

Eventオブジェクト

Eventの基本的な使い方

Eventオブジェクトは、スレッド間のシグナリングを行うための同期プリミティブです。

Eventオブジェクトを使用すると、スレッドは特定のイベントが発生するまで待機し、イベントが発生したときに通知を受け取ることができます。

以下は、Eventを使った基本的な例です。

import threading
import time

# イベントオブジェクトを作成
event = threading.Event()

# タスクを定義
def task():
    print("Waiting for event to be set")
    event.wait()  # イベントがセットされるのを待つ
    print("Event is set, continuing execution")

# スレッドを作成して開始
thread = threading.Thread(target=task)
thread.start()

# 2秒間スリープ
time.sleep(2)

# イベントをセット
print("Setting event")
event.set()

# スレッドの完了を待機
thread.join()

この例では、taskスレッドがイベントがセットされるまで待機し、イベントがセットされた後に処理を続行します。

Eventを使ったスレッドの制御

Eventオブジェクトを使用することで、スレッドの一時停止と再開が容易になります。

例えば、特定の条件が満たされるまでスレッドを待機させたり、複数のスレッドを同時に再開させる場合に使用されます。

以上が、Pythonでのスレッド同期の方法です。

各同期プリミティブの特性を理解し、適切に使用することで、スレッド間の競合やデッドロックを防ぎ、効率的なマルチスレッドプログラムを作成することができます。

実践例

ここでは、実際にPythonでスレッドを同期させる方法を具体的な例を通じて解説します。

各同期オブジェクトの使い方を理解するために、サンプルコードとその実行結果を示します。

ロックを使った同期の例

ロックを使って複数のスレッドが同時に同じリソースにアクセスしないようにする方法を見てみましょう。

import threading
# 共有リソース
counter = 0
# ロックオブジェクトの作成
lock = threading.Lock()
def increment():
    global counter
    for _ in range(100000):
        # ロックを取得
        lock.acquire()
        counter += 1
        # ロックを解放
        lock.release()
# スレッドの作成
threads = []
for _ in range(10):
    thread = threading.Thread(target=increment)
    threads.append(thread)
    thread.start()
# 全てのスレッドが終了するのを待つ
for thread in threads:
    thread.join()
print(f"最終的なカウンターの値: {counter}")

このコードでは、10個のスレッドが同時にcounterをインクリメントします。

ロックを使うことで、counterの値が正しく更新されることを保証しています。

Conditionを使った生産者-消費者問題の解決

Conditionオブジェクトを使って、生産者-消費者問題を解決する方法を見てみましょう。

import threading
import time
import random

# バッファとそのサイズを定義
buffer = []
buffer_size = 10
condition = threading.Condition()

# 生産者タスクを定義
def producer():
    global buffer
    while True:
        item = random.randint(1, 100)  # ランダムなアイテムを生成
        condition.acquire()
        while len(buffer) >= buffer_size:  # バッファが満杯の場合待機
            condition.wait()
        buffer.append(item)  # バッファにアイテムを追加
        print(f"生産者がアイテム {item} を追加しました。")
        condition.notify()  # 消費者に通知
        condition.release()
        time.sleep(random.random())  # ランダムな時間スリープ

# 消費者タスクを定義
def consumer():
    global buffer
    while True:
        condition.acquire()
        while not buffer:  # バッファが空の場合待機
            condition.wait()
        item = buffer.pop(0)  # バッファからアイテムを取り出す
        print(f"消費者がアイテム {item} を消費しました。")
        condition.notify()  # 生産者に通知
        condition.release()
        time.sleep(random.random())  # ランダムな時間スリープ

# 生産者スレッドを作成して開始
producer_thread = threading.Thread(target=producer)
# 消費者スレッドを作成して開始
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

# スレッドの完了を待機
producer_thread.join()
consumer_thread.join()

このコードでは、生産者スレッドがアイテムを生成し、消費者スレッドがアイテムを消費します。

Conditionオブジェクトを使って、バッファが満杯または空のときにスレッドを待機させることができます。

Semaphoreを使ったリソース制限の例

Semaphoreを使って、同時にアクセスできるリソースの数を制限する方法を見てみましょう。

import threading
import time
# セマフォオブジェクトの作成(最大3つのスレッドが同時にアクセス可能)
semaphore = threading.Semaphore(3)
def access_resource(thread_id):
    print(f"スレッド {thread_id} がリソースを待っています...")
    semaphore.acquire()
    print(f"スレッド {thread_id} がリソースにアクセスしています...")
    time.sleep(2)
    print(f"スレッド {thread_id} がリソースを解放しました。")
    semaphore.release()
threads = []
for i in range(10):
    thread = threading.Thread(target=access_resource, args=(i,))
    threads.append(thread)
    thread.start()
for thread in threads:
    thread.join()

このコードでは、最大3つのスレッドが同時にリソースにアクセスできるようにセマフォを使用しています。

セマフォを使うことで、リソースの競合を防ぐことができます。

Eventを使ったスレッドの一時停止と再開

Eventオブジェクトを使って、スレッドを一時停止させたり再開させたりする方法を見てみましょう。

import threading
import time
# イベントオブジェクトの作成
event = threading.Event()
def worker():
    while True:
        print("スレッドが動作中...")
        time.sleep(1)
        event.wait()  # イベントがセットされるまで待機
        print("スレッドが再開しました。")
def controller():
    time.sleep(5)
    print("スレッドを一時停止します。")
    event.clear()  # イベントをクリアしてスレッドを停止
    time.sleep(5)
    print("スレッドを再開します。")
    event.set()  # イベントをセットしてスレッドを再開
worker_thread = threading.Thread(target=worker)
controller_thread = threading.Thread(target=controller)
worker_thread.start()
controller_thread.start()
worker_thread.join()
controller_thread.join()

このコードでは、workerスレッドが定期的に動作し、controllerスレッドがworkerスレッドを一時停止させたり再開させたりします。

Eventオブジェクトを使うことで、スレッドの動作を制御することができます。

これらの実践例を通じて、Pythonでのスレッド同期の基本的な使い方を理解できたでしょう。

次に、スレッド同期のベストプラクティスについて見ていきましょう。

スレッド同期のベストプラクティス

スレッド同期は、マルチスレッドプログラムの安定性と効率性を確保するために非常に重要です。

ここでは、スレッド同期のベストプラクティスについて解説します。

適切な同期オブジェクトの選択

Pythonには、スレッド同期のためのさまざまなオブジェクトが用意されています。

適切な同期オブジェクトを選択することは、プログラムのパフォーマンスと可読性を向上させるために重要です。

ロック(Lock)

ロックは、最も基本的な同期オブジェクトです。

単一のスレッドがリソースにアクセスすることを保証します。

シンプルな排他制御が必要な場合に適しています。

import threading
lock = threading.Lock()
def critical_section():
    with lock:
        # ここで排他制御が行われる
        print("クリティカルセクションに入っています")
thread1 = threading.Thread(target=critical_section)
thread2 = threading.Thread(target=critical_section)
thread1.start()
thread2.start()
thread1.join()
thread2.join()

RLock(再帰ロック)

RLockは、同じスレッドが複数回ロックを取得できるようにする再帰的なロックです。

再帰的なロックが必要な場合に使用します。

import threading

# 再帰的ロックオブジェクトを作成
rlock = threading.RLock()

# 再帰的な関数を定義
def recursive_function(n):
    with rlock:  # ロックを取得
        if n > 0:
            print(f"再帰レベル: {n}")
            recursive_function(n - 1)  # 再帰呼び出し

# スレッドを作成して開始
thread = threading.Thread(target=recursive_function, args=(3,))
thread.start()

# スレッドの完了を待機
thread.join()

Condition

Conditionは、スレッド間の通信を行うためのオブジェクトです。

特定の条件が満たされるまでスレッドを待機させることができます。

import threading

# 条件変数オブジェクトを作成
condition = threading.Condition()
shared_data = []

# 生産者タスクを定義
def producer():
    with condition:
        shared_data.append(1)  # データを追加
        condition.notify()  # 消費者に通知

# 消費者タスクを定義
def consumer():
    with condition:
        condition.wait()  # 生産者からの通知を待つ
        print("データを消費しました:", shared_data.pop())  # データを消費

# スレッドを作成
thread1 = threading.Thread(target=producer)
thread2 = threading.Thread(target=consumer)

# 消費者スレッドを開始
thread2.start()
# 生産者スレッドを開始
thread1.start()

# スレッドの完了を待機
thread1.join()
thread2.join()

Semaphore

Semaphoreは、特定の数のスレッドがリソースにアクセスできるようにするためのオブジェクトです。

リソースの制限が必要な場合に使用します。

import threading

# セマフォオブジェクトを作成、最大2つのスレッドが同時にリソースにアクセスできる
semaphore = threading.Semaphore(2)

# リソースにアクセスするタスクを定義
def access_resource():
    with semaphore:  # セマフォを使用してリソースにアクセス
        print("リソースにアクセスしています")
        threading.current_thread().name
        # リソースにアクセスする処理をここに追加することができます

# スレッドを作成
threads = [threading.Thread(target=access_resource) for _ in range(4)]

# スレッドを開始
for thread in threads:
    thread.start()

# スレッドの完了を待機
for thread in threads:
    thread.join()

Event

Eventは、スレッドの一時停止と再開を制御するためのオブジェクトです。

特定のイベントが発生するまでスレッドを待機させることができます。

import threading

# イベントオブジェクトを作成
event = threading.Event()

# イベントを待機するタスクを定義
def wait_for_event():
    print("イベントを待っています")
    event.wait()  # イベントがセットされるのを待つ
    print("イベントが発生しました")

# スレッドを作成して開始
thread = threading.Thread(target=wait_for_event)
thread.start()

# ここで何かの処理を行うことができます
# 例えば、2秒間のスリープ
# time.sleep(2)

# イベントをセット
event.set()

# スレッドの完了を待機
thread.join()

デッドロックの回避方法

デッドロックは、複数のスレッドが互いにロックを待ち続ける状態です。

これを回避するための方法をいくつか紹介します。

ロックの順序を統一する

複数のロックを取得する場合、すべてのスレッドでロックの取得順序を統一することでデッドロックを回避できます。

import threading

# ロックオブジェクトを作成
lock1 = threading.Lock()
lock2 = threading.Lock()

# スレッド1のタスクを定義
def thread1():
    with lock1:  # lock1を取得
        with lock2:  # lock2を取得
            print("Thread 1")

# スレッド2のタスクを定義
def thread2():
    with lock1:  # lock1を取得
        with lock2:  # lock2を取得
            print("Thread 2")

# スレッドを作成
t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)

# スレッドを開始
t1.start()
t2.start()

# スレッドの完了を待機
t1.join()
t2.join()

タイムアウトを設定する

ロックの取得にタイムアウトを設定することで、デッドロックを回避できます。

import threading

# ロックオブジェクトを作成
lock = threading.Lock()

# ロックの取得を試みるタスクを定義
def try_lock():
    if lock.acquire(timeout=1):  # 1秒以内にロックを取得しようとする
        try:
            print("ロックを取得しました")
        finally:
            lock.release()  # ロックを解放
    else:
        print("ロックの取得に失敗しました")

# スレッドを作成
thread1 = threading.Thread(target=try_lock)
thread2 = threading.Thread(target=try_lock)

# スレッドを開始
thread1.start()
thread2.start()

# スレッドの完了を待機
thread1.join()
thread2.join()

パフォーマンスの最適化

スレッド同期は、プログラムのパフォーマンスに影響を与える可能性があります。

以下の方法でパフォーマンスを最適化できます。

ロックの粒度を小さくする

ロックの範囲を最小限にすることで、スレッドの競合を減らし、パフォーマンスを向上させることができます。

import threading

# ロックオブジェクトを作成
lock = threading.Lock()
shared_data = 0

# 共有データをインクリメントするタスクを定義
def increment():
    global shared_data
    for _ in range(100000):
        with lock:  # ロックを取得
            shared_data += 1

# スレッドを作成
threads = [threading.Thread(target=increment) for _ in range(4)]

# スレッドを開始
for thread in threads:
    thread.start()

# スレッドの完了を待機
for thread in threads:
    thread.join()

# 最終的な共有データの値を表示
print("最終的な値:", shared_data)

不要なロックを避ける

必要のない場合はロックを使用しないことで、パフォーマンスを向上させることができます。

import threading

shared_data = 0

# 共有データをインクリメントするタスクを定義
def increment():
    global shared_data
    for _ in range(100000):
        shared_data += 1

# スレッドを作成
threads = [threading.Thread(target=increment) for _ in range(4)]

# スレッドを開始
for thread in threads:
    thread.start()

# スレッドの完了を待機
for thread in threads:
    thread.join()

# 最終的な共有データの値を表示
print("最終的な値:", shared_data)

こちらのコードは意図的にロックを避けていますが、プログラムの規模が大きくなる際に意図せずデータ競合が発生する可能性があります。

最高効率を求める場合には有効ですが、闇雲にロックを避けることはおすすめしません。

スレッドプールを使用する

スレッドプールを使用することで、スレッドの作成と破棄のオーバーヘッドを減らし、パフォーマンスを向上させることができます。

from concurrent.futures import ThreadPoolExecutor
def task(n):
    print(f"タスク {n} を実行中")
with ThreadPoolExecutor(max_workers=4) as executor:
    for i in range(10):
        executor.submit(task, i)

以上が、スレッド同期のベストプラクティスです。

適切な同期オブジェクトの選択、デッドロックの回避、パフォーマンスの最適化を行うことで、効率的で安定したマルチスレッドプログラムを作成することができます。

目次に戻る

当サイトはリンクフリーです。出典元を明記していただければ、ご自由に引用していただいて構いません。

スレッド

関連記事

目次から探す