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

import threading
# 共有リソース
shared_resource = 0
lock = threading.Lock()
def increment():
    global shared_resource
    lock.acquire()  # ロックを取得
    shared_resource += 1
    lock.release()  # ロックを解放
# スレッドの作成と実行
threads = [threading.Thread(target=increment) for _ in range(10)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()
print(shared_resource)  # 出力: 10

import threading
# 共有リソース
shared_resource = 0
rlock = threading.RLock()
def recursive_increment(count):
    global shared_resource
    rlock.acquire()  # RLockを取得
    if count > 0:
        shared_resource += 1
        recursive_increment(count - 1)  # 再帰呼び出し
    rlock.release()  # RLockを解放
recursive_increment(5)
print(shared_resource)  # 出力: 5

import threading
import time
# セマフォの作成
semaphore = threading.Semaphore(2)
def access_resource(thread_id):
    with semaphore:  # セマフォを取得
        print(f"スレッド {thread_id} がリソースにアクセス中...")
        time.sleep(2)  # リソースの使用
        print(f"スレッド {thread_id} がリソースの使用を終了しました。")
# スレッドの作成と実行
threads = [threading.Thread(target=access_resource, args=(i,)) for i in range(5)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

import threading
import time
# イベントの作成
event = threading.Event()
def wait_for_event():
    print("スレッドがイベントを待機中...")
    event.wait()  # イベントが立つまで待機
    print("イベントが発生しました！")
# スレッドの作成と実行
thread = threading.Thread(target=wait_for_event)
thread.start()
time.sleep(2)  # 2秒待機
event.set()  # イベントを立てる
thread.join()

import threading
# コンディションの作成
condition = threading.Condition()
shared_data = []
def producer():
    global shared_data
    with condition:
        shared_data.append(1)  # データを追加
        condition.notify()  # 待機しているスレッドに通知
def consumer():
    with condition:
        while not shared_data:  # データがない場合は待機
            condition.wait()
        print("データを消費しました。")
# スレッドの作成と実行
prod_thread = threading.Thread(target=producer)
cons_thread = threading.Thread(target=consumer)
cons_thread.start()
prod_thread.start()
prod_thread.join()
cons_thread.join()

from concurrent.futures import ThreadPoolExecutor
import time
def task(n):
    print(f"タスク {n} を実行中...")
    time.sleep(1)  # 模擬的な処理
    return f"タスク {n} が完了しました。"
# スレッドプールの作成
with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(task, range(5))
# 結果の表示
for result in results:
    print(result)

import requests
import threading
urls = [
    "http://example.com/page1",
    "http://example.com/page2",
    "http://example.com/page3",
]
def fetch_url(url):
    response = requests.get(url)
    print(f"{url} のステータスコード: {response.status_code}")
threads = [threading.Thread(target=fetch_url, args=(url,)) for url in urls]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

import sqlite3
import threading
def insert_data(value):
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()
    cursor.execute("INSERT INTO my_table (value) VALUES (?)", (value,))
    conn.commit()
    conn.close()
threads = [threading.Thread(target=insert_data, args=(i,)) for i in range(10)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

import threading
file_names = ["file1.txt", "file2.txt", "file3.txt"]
def read_file(file_name):
    with open(file_name, 'r') as file:
        data = file.read()
        print(f"{file_name} の内容: {data[:50]}")  # 最初の50文字を表示
threads = [threading.Thread(target=read_file, args=(file_name,)) for file_name in file_names]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

import threading
import time
import random
data_queue = []
def data_producer():
    while True:
        data = random.randint(1, 100)
        data_queue.append(data)
        print(f"データを生成: {data}")
        time.sleep(1)  # 1秒ごとにデータを生成
def data_consumer():
    while True:
        if data_queue:
            data = data_queue.pop(0)
            print(f"データを処理: {data}")
        time.sleep(1)  # 1秒ごとにデータを処理
producer_thread = threading.Thread(target=data_producer)
consumer_thread = threading.Thread(target=data_consumer)
producer_thread.start()
consumer_thread.start()

import threading
import time
def render():
    while True:
        print("ゲームを描画中...")
        time.sleep(0.1)  # 描画処理の模擬
def ai_processing():
    while True:
        print("AI処理を実行中...")
        time.sleep(0.5)  # AI処理の模擬
render_thread = threading.Thread(target=render)
ai_thread = threading.Thread(target=ai_processing)
render_thread.start()
ai_thread.start()