Pythonでスレッド間通信を行う際の基本と活用8選

import threading
import time

# グローバル変数
shared_variable = 0

def increment_variable():
    global shared_variable
    for _ in range(1000000):
        shared_variable += 1

def main():
    thread1 = threading.Thread(target=increment_variable)
    thread2 = threading.Thread(target=increment_variable)

    thread1.start()
    thread2.start()

    thread1.join()
    thread2.join()

    print(f"最終的な値: {shared_variable}")

if __name__ == "__main__":
    main()

import threading
import queue
import time

def producer(q):
    for i in range(5):
        q.put(i)
        print(f"生産: {i}")
        time.sleep(1)

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"消費: {item}")
        q.task_done()

def main():
    q = queue.Queue()
    producer_thread = threading.Thread(target=producer, args=(q,))
    consumer_thread = threading.Thread(target=consumer, args=(q,))

    producer_thread.start()
    consumer_thread.start()

    producer_thread.join()
    q.put(None)  # 終了シグナル
    consumer_thread.join()

if __name__ == "__main__":
    main()

生産: 0
消費: 0
生産: 1
消費: 1
生産: 2
消費: 2
生産: 3
消費: 3
生産: 4
消費: 4

import threading
import time

def worker(event):
    print("作業者：待機中...")
    event.wait()
    print("作業者：仕事開始！")

def main():
    event = threading.Event()
    thread = threading.Thread(target=worker, args=(event,))
    thread.start()

    time.sleep(3)  # 3秒待機
    print("メイン：シグナル送信")
    event.set()

    thread.join()

if __name__ == "__main__":
    main()

作業者：待機中...
メイン：シグナル送信
作業者：仕事開始！

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

def main():
    threads = []
    for _ in range(5):
        thread = threading.Thread(target=increment)
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()

    print(f"カウンター最終値: {counter}")

if __name__ == "__main__":
    main()

import threading
import time

class Buffer:
    def __init__(self, size):
        self.buffer = []
        self.size = size
        self.condition = threading.Condition()

    def add(self, item):
        with self.condition:
            while len(self.buffer) == self.size:
                self.condition.wait()
            self.buffer.append(item)
            self.condition.notify()

    def remove(self):
        with self.condition:
            while len(self.buffer) == 0:
                self.condition.wait()
            item = self.buffer.pop(0)
            self.condition.notify()
            return item

def producer(buffer):
    for i in range(10):
        buffer.add(i)
        print(f"生産: {i}")
        time.sleep(0.1)

def consumer(buffer):
    for _ in range(10):
        item = buffer.remove()
        print(f"消費: {item}")
        time.sleep(0.2)

def main():
    buffer = Buffer(5)
    producer_thread = threading.Thread(target=producer, args=(buffer,))
    consumer_thread = threading.Thread(target=consumer, args=(buffer,))

    producer_thread.start()
    consumer_thread.start()

    producer_thread.join()
    consumer_thread.join()

if __name__ == "__main__":
    main()

生産: 0
生産: 1
消費: 0
生産: 2
生産: 3
消費: 1
生産: 4
生産: 5
消費: 2
生産: 6
消費: 3
生産: 7
消費: 4
生産: 8
消費: 5
生産: 9
消費: 6
消費: 7
消費: 8
消費: 9

import threading
import time
import random

class DatabaseConnection:
    def query(self):
        time.sleep(random.random())

semaphore = threading.Semaphore(3)  # 最大3つの同時接続

def worker(worker_id):
    with semaphore:
        print(f"ワーカー {worker_id} がデータベースに接続")
        connection = DatabaseConnection()
        connection.query()
        print(f"ワーカー {worker_id} が接続解除")

def main():
    threads = []
    for i in range(10):
        thread = threading.Thread(target=worker, args=(i,))
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()

if __name__ == "__main__":
    main()

ワーカー 0 がデータベースに接続
ワーカー 1 がデータベースに接続
ワーカー 2 がデータベースに接続
ワーカー 1 が接続解除
ワーカー 3 がデータベースに接続
ワーカー 0 が接続解除
ワーカー 4 がデータベースに接続
ワーカー 2 が接続解除
ワーカー 5 がデータベースに接続
ワーカー 3 が接続解除
ワーカー 6 がデータベースに接続
ワーカー 4 が接続解除
ワーカー 7 がデータベースに接続
ワーカー 5 が接続解除
ワーカー 8 がデータベースに接続
ワーカー 6 が接続解除
ワーカー 9 がデータベースに接続
ワーカー 7 が接続解除
ワーカー 8 が接続解除
ワーカー 9 が接続解除

import threading
import multiprocessing

def increment(value):
    for _ in range(1000000):
        with value.get_lock():
            value.value += 1

def main():
    value = multiprocessing.Value('i', 0)
    threads = []
    for _ in range(4):
        thread = threading.Thread(target=increment, args=(value,))
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()

    print(f"最終値: {value.value}")

if __name__ == "__main__":
    main()

import threading
import multiprocessing

def sender(conn):
    for i in range(5):
        conn.send(f"メッセージ {i}")
    conn.send(None)  # 終了シグナル

def receiver(conn):
    while True:
        message = conn.recv()
        if message is None:
            break
        print(f"受信: {message}")

def main():
    parent_conn, child_conn = multiprocessing.Pipe()

    sender_thread = threading.Thread(target=sender, args=(parent_conn,))
    receiver_thread = threading.Thread(target=receiver, args=(child_conn,))

    sender_thread.start()
    receiver_thread.start()

    sender_thread.join()
    receiver_thread.join()

if __name__ == "__main__":
    main()

受信: メッセージ 0
受信: メッセージ 1
受信: メッセージ 2
受信: メッセージ 3
受信: メッセージ 4

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

threads = [threading.Thread(target=increment) for _ in range(5)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

print(f"カウンター: {counter}")

import threading

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

def worker1():
    with lock1:
        print("ワーカー1: lock1獲得")
        with lock2:
            print("ワーカー1: lock2獲得")

def worker2():
    with lock1:  # lock2ではなくlock1を先に獲得
        print("ワーカー2: lock1獲得")
        with lock2:
            print("ワーカー2: lock2獲得")

thread1 = threading.Thread(target=worker1)
thread2 = threading.Thread(target=worker2)

thread1.start()
thread2.start()

thread1.join()
thread2.join()

ワーカー1: lock1獲得
ワーカー1: lock2獲得
ワーカー2: lock1獲得
ワーカー2: lock2獲得

import weakref

class Node:
    def __init__(self, value):
        self.value = value
        self.parent = None
        self.children = []

    def add_child(self, child):
        self.children.append(child)
        child.parent = weakref.ref(self)  # 弱参照を使用

root = Node(1)
child = Node(2)
root.add_child(child)

# rootとchildの参照を削除
del root
del child

# ガベージコレクションが自動的に行われる

import threading
import queue

def search_word(file_chunk, target_word, result_queue):
    count = 0
    for line in file_chunk:
        count += line.count(target_word)
    result_queue.put(count)

def parallel_search(filename, target_word, num_threads=4):
    with open(filename, 'r') as file:
        content = file.readlines()

    chunk_size = len(content) // num_threads
    chunks = [content[i:i+chunk_size] for i in range(0, len(content), chunk_size)]

    result_queue = queue.Queue()
    threads = []

    for chunk in chunks:
        thread = threading.Thread(target=search_word, args=(chunk, target_word, result_queue))
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()

    total_count = 0
    while not result_queue.empty():
        total_count += result_queue.get()

    return total_count

# 使用例
filename = "large_text_file.txt"
target_word = "Python"
occurrences = parallel_search(filename, target_word)
print(f"'{target_word}'の出現回数: {occurrences}")

import threading
import queue
import time
import random

def sensor(data_queue):
    while True:
        # センサーからデータを読み取る（ここではランダムな値を生成）
        data = random.randint(0, 100)
        data_queue.put(data)
        time.sleep(0.1)  # 100ミリ秒ごとにデータを生成

def processor(data_queue, result_queue):
    while True:
        data = data_queue.get()
        # データを処理（ここでは単純に2倍にする）
        result = data * 2
        result_queue.put(result)

def display(result_queue):
    while True:
        result = result_queue.get()
        print(f"処理結果: {result}")

data_queue = queue.Queue()
result_queue = queue.Queue()

sensor_thread = threading.Thread(target=sensor, args=(data_queue,))
processor_thread = threading.Thread(target=processor, args=(data_queue, result_queue))
display_thread = threading.Thread(target=display, args=(result_queue,))

sensor_thread.start()
processor_thread.start()
display_thread.start()

# メインスレッドで10秒間実行
time.sleep(10)

処理結果: 138
処理結果: 116
処理結果: 72
処理結果: 18
処理結果: 180
...

import threading
import time
import random

class Character:
    def __init__(self, name):
        self.name = name
        self.x = 0
        self.y = 0
        self.lock = threading.Lock()

    def move(self):
        with self.lock:
            self.x += random.randint(-1, 1)
            self.y += random.randint(-1, 1)

def character_movement(character):
    while True:
        character.move()
        time.sleep(0.5)  # 0.5秒ごとに移動

def game_logic(characters):
    while True:
        print("ゲーム状態:")
        for character in characters:
            with character.lock:
                print(f"{character.name}: ({character.x}, {character.y})")
        time.sleep(1)  # 1秒ごとに状態を表示

characters = [Character("プレイヤー"), Character("敵1"), Character("敵2")]

movement_threads = [threading.Thread(target=character_movement, args=(char,)) for char in characters]
logic_thread = threading.Thread(target=game_logic, args=(characters,))

for thread in movement_threads:
    thread.start()
logic_thread.start()

# メインスレッドで30秒間実行
time.sleep(30)

ゲーム状態:
プレイヤー: (1, -2)
敵1: (0, 1)
敵2: (-1, 0)
ゲーム状態:
プレイヤー: (2, -1)
敵1: (-1, 2)
敵2: (-2, 1)
...

import threading
import queue
import requests
from bs4 import BeautifulSoup

def scrape_url(url, result_queue):
    try:
        response = requests.get(url)
        soup = BeautifulSoup(response.text, 'html.parser')
        title = soup.title.string if soup.title else "No title"
        result_queue.put((url, title))
    except Exception as e:
        result_queue.put((url, f"Error: {str(e)}"))

def parallel_scrape(urls, max_threads=5):
    result_queue = queue.Queue()
    threads = []

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

        # スレッド数を制限
        if len(threads) >= max_threads:
            for thread in threads:
                thread.join()
            threads = []

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

    results = []
    while not result_queue.empty():
        results.append(result_queue.get())

    return results

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

results = parallel_scrape(urls)
for url, title in results:
    print(f"URL: {url}")
    print(f"Title: {title}")
    print()

URL: https://www.python.org
Title: Welcome to Python.org

URL: https://www.google.com
Title: Google

URL: https://www.github.com
Title: GitHub: Let's build from here · GitHub

URL: https://www.stackoverflow.com
Title: Stack Overflow - Where Developers Learn, Share, & Build Careers

URL: https://www.reddit.com
Title: Reddit - Dive into anything