Pythonにおけるthreadingイベントの基礎と活用事例10選

def worker():
    print("ワーカースレッドが実行中です")

# スレッドの作成
thread = threading.Thread(target=worker)

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

# メインスレッドの処理を続行
print("メインスレッドが実行中です")

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

ワーカースレッドが実行中です
メインスレッドが実行中です

class MyThread(threading.Thread):
    def run(self):
        print("カスタムスレッドが実行中です")

# スレッドの作成と開始
thread = MyThread()
thread.start()

print("メインスレッドが実行中です")

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

カスタムスレッドが実行中です
メインスレッドが実行中です

import threading
import time

def worker():
    print("ワーカースレッド開始")
    time.sleep(2)  # 重い処理をシミュレート
    print("ワーカースレッド終了")

# メインスレッドの処理
print("メインスレッド開始")

# ワーカースレッドの作成と開始
thread = threading.Thread(target=worker)
thread.start()

print("メインスレッドは他の処理を続行")

# ワーカースレッドの終了を待機
thread.join()

print("メインスレッド終了")

メインスレッド開始
ワーカースレッド開始
メインスレッドは他の処理を続行
ワーカースレッド終了
メインスレッド終了

import threading
import time

def task(name, duration):
    print(f"{name} 開始")
    time.sleep(duration)
    print(f"{name} 終了（所要時間: {duration}秒）")

# スレッドのリストを作成
threads = [
    threading.Thread(target=task, args=("タスク1", 2)),
    threading.Thread(target=task, args=("タスク2", 3)),
    threading.Thread(target=task, args=("タスク3", 1))
]

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

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

print("全てのタスクが完了しました")

タスク1 開始
タスク2 開始
タスク3 開始
タスク3 終了（所要時間: 1秒）
タスク1 終了（所要時間: 2秒）
タスク2 終了（所要時間: 3秒）
全てのタスクが完了しました

import threading
import time

def worker(name, delay, repeat):
    print(f"{name} スレッド開始")
    for i in range(repeat):
        time.sleep(delay)
        print(f"{name}: {i + 1}回目の処理")
    print(f"{name} スレッド終了")

# 異なるパラメータを持つ2つのスレッドを作成
thread1 = threading.Thread(target=worker, args=("スレッドA", 1, 3))
thread2 = threading.Thread(target=worker, args=("スレッドB", 0.5, 5))

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

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

print("全てのスレッドが終了しました")

スレッドA スレッド開始
スレッドB スレッド開始
スレッドB: 1回目の処理
スレッドA: 1回目の処理
スレッドB: 2回目の処理
スレッドB: 3回目の処理
スレッドA: 2回目の処理
スレッドB: 4回目の処理
スレッドB: 5回目の処理
スレッドB スレッド終了
スレッドA: 3回目の処理
スレッドA スレッド終了
全てのスレッドが終了しました

import threading
import time

class BankAccount:
    def __init__(self):
        self.balance = 1000
        self.lock = threading.Lock()

    def withdraw(self, amount):
        with self.lock:
            if self.balance >= amount:
                time.sleep(0.1)  # 処理時間をシミュレート
                self.balance -= amount
                print(f"引き出し成功: {amount}円 (残高: {self.balance}円)")
            else:
                print(f"引き出し失敗: 残高不足 (残高: {self.balance}円)")

def perform_transactions(account):
    for _ in range(5):
        account.withdraw(200)

account = BankAccount()

thread1 = threading.Thread(target=perform_transactions, args=(account,))
thread2 = threading.Thread(target=perform_transactions, args=(account,))

thread1.start()
thread2.start()

thread1.join()
thread2.join()

print(f"最終残高: {account.balance}円")

引き出し成功: 200円 (残高: 800円)
引き出し成功: 200円 (残高: 600円)
引き出し成功: 200円 (残高: 400円)
引き出し成功: 200円 (残高: 200円)
引き出し成功: 200円 (残高: 0円)
引き出し失敗: 残高不足 (残高: 0円)
引き出し失敗: 残高不足 (残高: 0円)
引き出し失敗: 残高不足 (残高: 0円)
引き出し失敗: 残高不足 (残高: 0円)
引き出し失敗: 残高不足 (残高: 0円)
最終残高: 0円

import threading
import time
import random

def wait_for_event(event, name):
    print(f"{name} が待機中...")
    event.wait()
    print(f"{name} が起動しました！")

def trigger_event(event):
    print("イベントトリガーの準備中...")
    time.sleep(random.randint(1, 5))  # ランダムな待機時間
    print("イベントをトリガーします！")
    event.set()

event = threading.Event()

threads = [
    threading.Thread(target=wait_for_event, args=(event, "スレッド1")),
    threading.Thread(target=wait_for_event, args=(event, "スレッド2")),
    threading.Thread(target=wait_for_event, args=(event, "スレッド3")),
    threading.Thread(target=trigger_event, args=(event,))
]

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

print("全てのスレッドが完了しました")

スレッド1 が待機中...
スレッド2 が待機中...
スレッド3 が待機中...
イベントトリガーの準備中...
イベントをトリガーします！
スレッド1 が起動しました！
スレッド2 が起動しました！
スレッド3 が起動しました！
全てのスレッドが完了しました

import threading
import time
import random

def worker(semaphore, name):
    with semaphore:
        print(f"{name} が作業を開始しました")
        time.sleep(random.uniform(1, 3))  # ランダムな作業時間
        print(f"{name} が作業を完了しました")

semaphore = threading.Semaphore(3)  # 同時に3つまでのスレッドを許可

threads = [
    threading.Thread(target=worker, args=(semaphore, f"ワーカー{i}"))
    for i in range(10)
]

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

print("全てのワーカーが作業を完了しました")

ワーカー0 が作業を開始しました
ワーカー1 が作業を開始しました
ワーカー2 が作業を開始しました
ワーカー1 が作業を完了しました
ワーカー3 が作業を開始しました
ワーカー0 が作業を完了しました
ワーカー4 が作業を開始しました
ワーカー2 が作業を完了しました
ワーカー5 が作業を開始しました
ワーカー3 が作業を完了しました
ワーカー6 が作業を開始しました
ワーカー4 が作業を完了しました
ワーカー7 が作業を開始しました
ワーカー5 が作業を完了しました
ワーカー8 が作業を開始しました
ワーカー6 が作業を完了しました
ワーカー9 が作業を開始しました
ワーカー7 が作業を完了しました
ワーカー8 が作業を完了しました
ワーカー9 が作業を完了しました
全てのワーカーが作業を完了しました

import concurrent.futures
import time
import random

def task(name):
    print(f"{name} が開始しました")
    sleep_time = random.uniform(1, 5)
    time.sleep(sleep_time)
    print(f"{name} が完了しました（処理時間: {sleep_time:.2f}秒）")
    return sleep_time

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    tasks = [executor.submit(task, f"タスク{i}") for i in range(10)]

    for future in concurrent.futures.as_completed(tasks):
        result = future.result()
        print(f"タスクが完了しました（処理時間: {result:.2f}秒）")

print("全てのタスクが完了しました")

タスク0 が開始しました
タスク1 が開始しました
タスク2 が開始しました
タスク1 が完了しました（処理時間: 1.23秒）
タスクが完了しました（処理時間: 1.23秒）
タスク3 が開始しました
タスク0 が完了しました（処理時間: 2.45秒）
タスクが完了しました（処理時間: 2.45秒）
タスク4 が開始しました
タスク2 が完了しました（処理時間: 3.67秒）
タスクが完了しました（処理時間: 3.67秒）
タスク5 が開始しました
タスク3 が完了しました（処理時間: 1.89秒）
タスクが完了しました（処理時間: 1.89秒）
タスク6 が開始しました
タスク4 が完了しました（処理時間: 2.34秒）
タスクが完了しました（処理時間: 2.34秒）
タスク7 が開始しました
タスク5 が完了しました（処理時間: 1.56秒）
タスクが完了しました（処理時間: 1.56秒）
タスク8 が開始しました
タスク6 が完了しました（処理時間: 3.12秒）
タスクが完了しました（処理時間: 3.12秒）
タスク9 が開始しました
タスク7 が完了しました（処理時間: 2.78秒）
タスクが完了しました（処理時間: 2.78秒）
タスク8 が完了しました（処理時間: 1.45秒）
タスクが完了しました（処理時間: 1.45秒）
タスク9 が完了しました（処理時間: 4.01秒）
タスクが完了しました（処理時間: 4.01秒）
全てのタスクが完了しました

import threading
import time

def background_task():
    while True:
        print("バックグラウンドタスクが動作中...")
        time.sleep(1)

def main_task():
    print("メインタスクが開始しました")
    time.sleep(5)
    print("メインタスクが完了しました")

background_thread = threading.Thread(target=background_task, daemon=True)
background_thread.start()

main_thread = threading.Thread(target=main_task)
main_thread.start()
main_thread.join()

print("プログラムが終了しました")

バックグラウンドタスクが動作中...
メインタスクが開始しました
バックグラウンドタスクが動作中...
バックグラウンドタスクが動作中...
バックグラウンドタスクが動作中...
バックグラウンドタスクが動作中...
メインタスクが完了しました
プログラムが終了しました

import threading
import time

class StoppableThread(threading.Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name
        self._stop_event = threading.Event()

    def stop(self):
        self._stop_event.set()

    def stopped(self):
        return self._stop_event.is_set()

    def run(self):
        print(f"{self.name} が開始しました")
        while not self.stopped():
            print(f"{self.name} が動作中...")
            time.sleep(1)
        print(f"{self.name} が停止しました")

# スレッドの作成と開始
threads = [StoppableThread(f"スレッド{i}") for i in range(3)]
for thread in threads:
    thread.start()

# 5秒間スレッドを実行
time.sleep(5)

# スレッドの停止を要求
for thread in threads:
    thread.stop()

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

print("全てのスレッドが終了しました")

スレッド0 が開始しました
スレッド0 が動作中...
スレッド1 が開始しました
スレッド1 が動作中...
スレッド2 が開始しました
スレッド2 が動作中...
スレッド0 が動作中...
スレッド1 が動作中...
スレッド2 が動作中...
スレッド0 が動作中...
スレッド1 が動作中...
スレッド2 が動作中...
スレッド0 が動作中...
スレッド1 が動作中...
スレッド2 が動作中...
スレッド0 が動作中...
スレッド1 が動作中...
スレッド2 が動作中...
スレッド0 が停止しました
スレッド1 が停止しました
スレッド2 が停止しました
全てのスレッドが終了しました

import threading
import time

def check_server_status():
    print(f"サーバーの状態をチェック中... (現在時刻: {time.strftime('%H:%M:%S')})")
    # ここに実際のチェック処理を書く

    # 次回の実行をスケジュール
    threading.Timer(60, check_server_status).start()

# 最初の実行
check_server_status()

# メインスレッドを待機状態にして、プログラムが終了しないようにする
while True:
    time.sleep(1)

サーバーの状態をチェック中... (現在時刻: 14:30:00)
サーバーの状態をチェック中... (現在時刻: 14:31:00)
サーバーの状態をチェック中... (現在時刻: 14:32:00)
...

import threading
import requests
from bs4 import BeautifulSoup
import time

def scrape_website(url):
    response = requests.get(url)
    soup = BeautifulSoup(response.text, 'html.parser')
    title = soup.find('title').text
    print(f"URL: {url}\nタイトル: {title}\n")

urls = [
    "https://www.python.org",
    "https://www.google.com",
    "https://www.github.com",
    "https://www.stackoverflow.com",
    "https://www.reddit.com"
]

threads = []
start_time = time.time()

for url in urls:
    thread = threading.Thread(target=scrape_website, args=(url,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

end_time = time.time()
print(f"実行時間: {end_time - start_time:.2f}秒")

URL: https://www.github.com
タイトル: GitHub: Let's build from here · GitHub

URL: https://www.google.com
タイトル: Google

URL: https://www.python.org
タイトル: Welcome to Python.org

URL: https://www.stackoverflow.com
タイトル: Stack Overflow - Where Developers Learn, Share, & Build Careers

URL: https://www.reddit.com
タイトル: Reddit - Dive into anything

実行時間: 1.23秒

import tkinter as tk
import threading
import time

def long_running_task():
    for i in range(10):
        time.sleep(1)
        progress_var.set(i + 1)
        progress_label.config(text=f"進捗: {i + 1}/10")
    result_label.config(text="タスク完了!")

def start_task():
    start_button.config(state=tk.DISABLED)
    thread = threading.Thread(target=long_running_task)
    thread.start()

root = tk.Tk()
root.title("マルチスレッドGUIアプリ")

start_button = tk.Button(root, text="タスク開始", command=start_task)
start_button.pack()

progress_var = tk.IntVar()
progress_bar = tk.Scale(root, variable=progress_var, from_=0, to=10, orient=tk.HORIZONTAL, length=200, state=tk.DISABLED)
progress_bar.pack()

progress_label = tk.Label(root, text="進捗: 0/10")
progress_label.pack()

result_label = tk.Label(root, text="")
result_label.pack()

root.mainloop()

import threading
import time

def acquire_locks(lock1, lock2):
    while True:
        with lock1:
            if lock2.acquire(blocking=False):
                print(f"{threading.current_thread().name} が両方のロックを取得しました")
                lock2.release()
                break
        time.sleep(0.1)

lock1 = threading.Lock()
lock2 = threading.Lock()

thread1 = threading.Thread(target=acquire_locks, args=(lock1, lock2), name="スレッド1")
thread2 = threading.Thread(target=acquire_locks, args=(lock2, lock1), name="スレッド2")

thread1.start()
thread2.start()

thread1.join()
thread2.join()

スレッド1 が両方のロックを取得しました
スレッド2 が両方のロックを取得しました

import threading
import time

def worker(stop_event):
    while not stop_event.is_set():
        print("作業中...")
        time.sleep(1)
    print("スレッドが終了しました")

stop_event = threading.Event()
thread = threading.Thread(target=worker, args=(stop_event,))
thread.start()

# メインスレッドで5秒待機
time.sleep(5)

# スレッドに停止を通知
stop_event.set()

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

print("プログラムが終了しました")

import threading
import time

class ResourceManager:
    def __init__(self):
        self.resource = "重要なリソース"
        self.lock = threading.Lock()

    def use_resource(self):
        with self.lock:
            print(f"{threading.current_thread().name} がリソースを使用中: {self.resource}")
            time.sleep(1)

    def __del__(self):
        print("リソースが解放されました")

def worker(manager):
    for _ in range(3):
        manager.use_resource()

manager = ResourceManager()

threads = [threading.Thread(target=worker, args=(manager,)) for _ in range(3)]

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

print("全てのスレッドが終了しました")

import asyncio
import threading
import time

# asyncioを使用した非同期関数
async def async_task(name):
    print(f"{name} 開始")
    await asyncio.sleep(1)
    print(f"{name} 終了")

# threadingを使用した関数
def thread_task(name):
    print(f"{name} 開始")
    time.sleep(1)
    print(f"{name} 終了")

# asyncioの実行
async def run_async():
    start = time.time()
    await asyncio.gather(
        async_task("非同期タスク1"),
        async_task("非同期タスク2"),
        async_task("非同期タスク3")
    )
    end = time.time()
    print(f"asyncio実行時間: {end - start:.2f}秒")

# threadingの実行
def run_threading():
    start = time.time()
    threads = [
        threading.Thread(target=thread_task, args=(f"スレッドタスク{i+1}",))
        for i in range(3)
    ]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    end = time.time()
    print(f"threading実行時間: {end - start:.2f}秒")

# 両方を実行
asyncio.run(run_async())
run_threading()

非同期タスク1 開始
非同期タスク2 開始
非同期タスク3 開始
非同期タスク1 終了
非同期タスク2 終了
非同期タスク3 終了
asyncio実行時間: 1.00秒
スレッドタスク1 開始
スレッドタスク2 開始
スレッドタスク3 開始
スレッドタスク1 終了
スレッドタスク2 終了
スレッドタスク3 終了
threading実行時間: 1.00秒

import multiprocessing
import time

def cpu_bound_task(n):
    count = 0
    for i in range(n):
        count += i * i
    return count

def run_serial(numbers):
    start = time.time()
    results = [cpu_bound_task(num) for num in numbers]
    end = time.time()
    print(f"シリアル実行時間: {end - start:.2f}秒")
    return results

def run_parallel(numbers):
    start = time.time()
    with multiprocessing.Pool() as pool:
        results = pool.map(cpu_bound_task, numbers)
    end = time.time()
    print(f"並列実行時間: {end - start:.2f}秒")
    return results

if __name__ == "__main__":
    numbers = [10**7, 10**7, 10**7, 10**7]

    serial_results = run_serial(numbers)
    parallel_results = run_parallel(numbers)

    print("結果が一致:", serial_results == parallel_results)

シリアル実行時間: 7.23秒
並列実行時間: 2.14秒
結果が一致: True

import multiprocessing as mp
from multiprocessing import shared_memory
import numpy as np

def modify_array(shm_name):
    existing_shm = shared_memory.SharedMemory(name=shm_name)
    shared_array = np.ndarray((10,), dtype=np.int64, buffer=existing_shm.buf)
    shared_array *= 2
    existing_shm.close()

if __name__ == "__main__":
    original_array = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], dtype=np.int64)
    shm = shared_memory.SharedMemory(create=True, size=original_array.nbytes)
    shared_array = np.ndarray(original_array.shape, dtype=original_array.dtype, buffer=shm.buf)
    shared_array[:] = original_array[:]

    print("元の配列:", shared_array)

    process = mp.Process(target=modify_array, args=(shm.name,))
    process.start()
    process.join()

    print("変更後の配列:", shared_array)

    shm.close()
    shm.unlink()

元の配列: [ 1  2  3  4  5  6  7  8  9 10]
変更後の配列: [ 2  4  6  8 10 12 14 16 18 20]