Pythonでスレッド終了を円滑に行う方法と活用方法10選

import threading

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

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

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

# メインスレッドの処理
print("メインスレッドも動いています")

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

スレッドが実行中です
メインスレッドも動いています

import threading
import time

def file_worker():
    with open("example.txt", "w") as f:
        f.write("Hello, World!")
        time.sleep(5)  # 長時間の処理を想定

thread = threading.Thread(target=file_worker)
thread.start()

# メインスレッドはすぐに終了
print("メインスレッドが終了しました")

import threading
import time

def file_worker():
    with open("example.txt", "w") as f:
        f.write("Hello, World!")
        time.sleep(5)  # 長時間の処理を想定

thread = threading.Thread(target=file_worker)
thread.start()

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

print("全ての処理が完了しました")

import threading
import time

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

def run_task(n):
    start = time.time()
    result = cpu_bound_task(n)
    end = time.time()
    print(f"計算結果: {result}, 実行時間: {end - start:.2f}秒")

# シングルスレッドで実行
run_task(50000000)

# マルチスレッドで実行
threads = []
for _ in range(2):
    thread = threading.Thread(target=run_task, args=(25000000,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

計算結果: 1249999975000000, 実行時間: 2.34秒
計算結果: 312499993750000, 実行時間: 1.18秒
計算結果: 312499993750000, 実行時間: 1.19秒

import threading
import time

def worker():
    print("作業を開始します")
    time.sleep(2)
    print("作業が完了しました")

thread = threading.Thread(target=worker)
thread.start()

print("スレッドの終了を待機中...")
thread.join()
print("スレッドが終了しました")

作業を開始します
スレッドの終了を待機中...
作業が完了しました
スレッドが終了しました

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()

# 3秒後にスレッドを停止
time.sleep(3)
stop_event.set()

thread.join()
print("スレッドが正常に終了しました")

作業中...
作業中...
作業中...
作業を終了します
スレッドが正常に終了しました

import threading
import time

def worker(stop_flag):
    while not stop_flag[0]:
        print("作業中...")
        time.sleep(1)
    print("作業を終了します")

stop_flag = [False]
thread = threading.Thread(target=worker, args=(stop_flag,))
thread.start()

# 3秒後にスレッドを停止
time.sleep(3)
stop_flag[0] = True

thread.join()
print("スレッドが正常に終了しました")

作業中...
作業中...
作業中...
作業を終了します
スレッドが正常に終了しました

import threading
import time

def worker():
    print("長時間の作業を開始します")
    time.sleep(10)
    print("作業が完了しました")

thread = threading.Thread(target=worker)
thread.start()

# 5秒間待機し、スレッドが終了しなければタイムアウト
timeout = 5
thread.join(timeout)

if thread.is_alive():
    print(f"{timeout}秒経過しましたが、スレッドはまだ実行中です")
else:
    print("スレッドは正常に終了しました")

長時間の作業を開始します
5秒経過しましたが、スレッドはまだ実行中です

import threading
import time
import ctypes

def worker():
    try:
        while True:
            print("作業中...")
            time.sleep(1)
    finally:
        print("クリーンアップ処理を実行します")

def terminate_thread(thread):
    if not thread.is_alive():
        return

    exc = ctypes.py_object(SystemExit)
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(thread.ident), exc)
    if res == 0:
        raise ValueError("無効なスレッドIDです")
    elif res != 1:
        ctypes.pythonapi.PyThreadState_SetAsyncExc(thread.ident, None)
        raise SystemError("PyThreadState_SetAsyncExc失敗")

thread = threading.Thread(target=worker)
thread.start()

# 3秒後にスレッドを強制終了
time.sleep(3)
terminate_thread(thread)

thread.join()
print("スレッドが強制終了されました")

作業中...
作業中...
作業中...
クリーンアップ処理を実行します
スレッドが強制終了されました

import threading
import time

def daemon_worker():
    while True:
        print("デーモンスレッドが動作中...")
        time.sleep(1)

daemon_thread = threading.Thread(target=daemon_worker, daemon=True)
daemon_thread.start()

# メインスレッドは5秒後に終了
time.sleep(5)
print("メインスレッドが終了します")

デーモンスレッドが動作中...
デーモンスレッドが動作中...
デーモンスレッドが動作中...
デーモンスレッドが動作中...
デーモンスレッドが動作中...
メインスレッドが終了します

from concurrent.futures import ThreadPoolExecutor
import time

def worker(n):
    print(f"タスク {n} を開始")
    time.sleep(2)
    return f"タスク {n} が完了"

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(worker, i) for i in range(5)]

    for future in futures:
        print(future.result())

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

タスク 0 を開始
タスク 1 を開始
タスク 2 を開始
タスク 0 が完了
タスク 3 を開始
タスク 1 が完了
タスク 4 を開始
タスク 2 が完了
タスク 3 が完了
タスク 4 が完了
全てのタスクが完了しました

import threading
import queue
import time

def worker(q):
    while True:
        item = q.get()
        if item is None:
            print("終了シグナルを受け取りました")
            break
        print(f"アイテム {item} を処理中")
        time.sleep(1)
    print("ワーカースレッドが終了します")

q = queue.Queue()
thread = threading.Thread(target=worker, args=(q,))
thread.start()

for i in range(3):
    q.put(i)

q.put(None)  # 終了シグナル
thread.join()
print("メインスレッドが終了します")

アイテム 0 を処理中
アイテム 1 を処理中
アイテム 2 を処理中
終了シグナルを受け取りました
ワーカースレッドが終了します
メインスレッドが終了します

import signal
import threading
import time

def signal_handler(signum, frame):
    global running
    print("終了シグナルを受け取りました")
    running = False

def worker():
    global running
    while running:
        print("作業中...")
        time.sleep(1)
    print("ワーカースレッドが終了します")

running = True
signal.signal(signal.SIGINT, signal_handler)

thread = threading.Thread(target=worker)
thread.start()

print("Ctrl+Cを押すとプログラムが終了します")
try:
    while running:
        time.sleep(0.1)
except KeyboardInterrupt:
    pass

thread.join()
print("プログラムが正常に終了しました")

Ctrl+Cを押すとプログラムが終了します
作業中...
作業中...
作業中...
^C終了シグナルを受け取りました
ワーカースレッドが終了します
プログラムが正常に終了しました

import threading
import time
from contextlib import ExitStack

def worker(name, event):
    while not event.is_set():
        print(f"{name} が作業中...")
        time.sleep(1)
    print(f"{name} が終了します")

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

with ExitStack() as stack:
    events_and_threads = [create_thread(f"スレッド{i}") for i in range(3)]

    for event, thread in events_and_threads:
        stack.callback(event.set)
        stack.callback(thread.join)

    print("3秒後に全てのスレッドを終了します")
    time.sleep(3)

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

スレッド0 が作業中...
スレッド1 が作業中...
スレッド2 が作業中...
3秒後に全てのスレッドを終了します
スレッド0 が作業中...
スレッド1 が作業中...
スレッド2 が作業中...
スレッド0 が作業中...
スレッド1 が作業中...
スレッド2 が作業中...
スレッド0 が終了します
スレッド1 が終了します
スレッド2 が終了します
全てのスレッドが終了しました

import threading
import time

def worker(file_path):
    try:
        with open(file_path, 'w') as f:
            f.write("スレッドが作業中です\n")
            time.sleep(2)
            f.write("スレッドが作業を完了しました\n")
    except Exception as e:
        print(f"エラーが発生しました: {e}")
    finally:
        print("リソースのクリーンアップを行います")

thread = threading.Thread(target=worker, args=("output.txt",))
thread.start()
thread.join()

print("メインスレッドが終了します")

import threading
import time

def worker(lock1, lock2):
    while True:
        print(f"スレッド {threading.current_thread().name} がロック1の取得を試みています")
        if lock1.acquire(timeout=1):
            try:
                print(f"スレッド {threading.current_thread().name} がロック1を取得しました")
                time.sleep(0.5)
                print(f"スレッド {threading.current_thread().name} がロック2の取得を試みています")
                if lock2.acquire(timeout=1):
                    try:
                        print(f"スレッド {threading.current_thread().name} がロック2を取得しました")
                        # 実際の処理をここに記述
                    finally:
                        lock2.release()
                else:
                    print(f"スレッド {threading.current_thread().name} がロック2の取得に失敗しました")
            finally:
                lock1.release()
        else:
            print(f"スレッド {threading.current_thread().name} がロック1の取得に失敗しました")
        time.sleep(0.1)

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

thread1 = threading.Thread(target=worker, args=(lock1, lock2), name="Thread-1")
thread2 = threading.Thread(target=worker, args=(lock2, lock1), name="Thread-2")

thread1.start()
thread2.start()

time.sleep(10)
print("プログラムを終了します")

import threading

class ThreadSafeCounter:
    def __init__(self):
        self._value = 0
        self._lock = threading.Lock()

    def increment(self):
        with self._lock:
            self._value += 1

    def decrement(self):
        with self._lock:
            self._value -= 1

    def get_value(self):
        with self._lock:
            return self._value

def worker(counter, num_operations):
    for _ in range(num_operations):
        counter.increment()

counter = ThreadSafeCounter()
num_threads = 5
num_operations = 100000

threads = []
for _ in range(num_threads):
    t = threading.Thread(target=worker, args=(counter, num_operations))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

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

import threading
import time

def worker():
    print("スレッドが作業を開始しました")
    time.sleep(2)
    print("スレッドが作業を完了しました")

thread = threading.Thread(target=worker)
thread.start()  # スレッドを開始してから
thread.join()   # join()を呼び出す

print("メインスレッドが終了します")

import threading
import time

def worker():
    print("スレッドが作業を開始しました")
    time.sleep(2)
    print("スレッドが作業を完了しました")

thread = threading.Thread(target=worker)

if not thread.is_alive():
    thread.start()

thread.join()

print("メインスレッドが終了します")

import threading
import time
import logging

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def worker(name):
    logger.debug(f"スレッド {name} が開始されました")
    while True:
        logger.debug(f"スレッド {name} が作業中です")
        time.sleep(1)

def monitor_threads():
    while True:
        active_threads = threading.enumerate()
        logger.info(f"アクティブなスレッド数: {len(active_threads)}")
        for thread in active_threads:
            logger.info(f"スレッド名: {thread.name}, 状態: {'実行中' if thread.is_alive() else '停止'}")
        time.sleep(5)

worker_thread = threading.Thread(target=worker, args=("Worker-1",), name="Worker-1")
worker_thread.start()

monitor_thread = threading.Thread(target=monitor_threads, name="Monitor")
monitor_thread.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    logger.info("プログラムを終了します")

import threading
import queue
import requests
from bs4 import BeautifulSoup
import time

def scraper(url_queue, result_queue, stop_event):
    while not stop_event.is_set():
        try:
            url = url_queue.get(timeout=1)
            response = requests.get(url)
            soup = BeautifulSoup(response.text, 'html.parser')
            title = soup.title.string if soup.title else "タイトルなし"
            result_queue.put((url, title))
            url_queue.task_done()
        except queue.Empty:
            continue
        except Exception as e:
            print(f"エラーが発生しました: {e}")

def main():
    urls = [
        "https://www.python.org",
        "https://docs.python.org",
        "https://pypi.org",
        "https://www.djangoproject.com",
        "https://flask.palletsprojects.com",
    ]

    url_queue = queue.Queue()
    result_queue = queue.Queue()
    stop_event = threading.Event()

    for url in urls:
        url_queue.put(url)

    threads = []
    for _ in range(3):  # 3つのスクレイパースレッドを作成
        t = threading.Thread(target=scraper, args=(url_queue, result_queue, stop_event))
        t.start()
        threads.append(t)

    start_time = time.time()
    url_queue.join()  # 全てのURLが処理されるまで待機
    stop_event.set()  # スレッドに終了シグナルを送信

    for t in threads:
        t.join()

    end_time = time.time()
    print(f"スクレイピングにかかった時間: {end_time - start_time:.2f}秒")

    while not result_queue.empty():
        url, title = result_queue.get()
        print(f"URL: {url}, タイトル: {title}")

if __name__ == "__main__":
    main()

import threading
import queue
import csv
import time
from concurrent.futures import ThreadPoolExecutor

def process_chunk(chunk, result_queue):
    total = sum(int(row['value']) for row in chunk if row['value'].isdigit())
    result_queue.put(total)

def read_csv_in_chunks(file_path, chunk_size=1000):
    with open(file_path, 'r') as f:
        reader = csv.DictReader(f)
        chunk = []
        for i, row in enumerate(reader):
            chunk.append(row)
            if (i + 1) % chunk_size == 0:
                yield chunk
                chunk = []
        if chunk:
            yield chunk

def main():
    file_path = 'large_dataset.csv'  # 大規模なCSVファイルへのパス
    chunk_size = 10000
    result_queue = queue.Queue()

    start_time = time.time()

    with ThreadPoolExecutor(max_workers=4) as executor:
        for chunk in read_csv_in_chunks(file_path, chunk_size):
            executor.submit(process_chunk, chunk, result_queue)

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

    end_time = time.time()
    print(f"合計値: {total_sum}")
    print(f"処理時間: {end_time - start_time:.2f}秒")

if __name__ == "__main__":
    main()

import threading
import queue
import random
import time

class DataProcessor:
    def __init__(self):
        self.data_queue = queue.Queue()
        self.result_queue = queue.Queue()
        self.stop_event = threading.Event()

    def data_generator(self):
        while not self.stop_event.is_set():
            data = random.randint(1, 100)
            self.data_queue.put(data)
            time.sleep(0.1)

    def data_processor(self):
        while not self.stop_event.is_set():
            try:
                data = self.data_queue.get(timeout=0.5)
                result = data * 2
                self.result_queue.put(result)
                self.data_queue.task_done()
            except queue.Empty:
                continue

    def result_handler(self):
        while not self.stop_event.is_set():
            try:
                result = self.result_queue.get(timeout=0.5)
                print(f"処理結果: {result}")
                self.result_queue.task_done()
            except queue.Empty:
                continue

    def run(self):
        threads = [
            threading.Thread(target=self.data_generator),
            threading.Thread(target=self.data_processor),
            threading.Thread(target=self.result_handler)
        ]

        for t in threads:
            t.start()

        try:
            while True:
                time.sleep(0.1)
        except KeyboardInterrupt:
            print("システムを終了します...")
            self.stop_event.set()

        for t in threads:
            t.join()

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

if __name__ == "__main__":
    processor = DataProcessor()
    processor.run()

import tkinter as tk
from tkinter import ttk
import threading
import time

class BackgroundTask:
    def __init__(self, update_callback):
        self.update_callback = update_callback
        self.running = False
        self.progress = 0

    def run(self):
        self.running = True
        while self.running and self.progress < 100:
            time.sleep(0.1)
            self.progress += 1
            self.update_callback(self.progress)

    def stop(self):
        self.running = False

class App(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title("バックグラウンドタスク例")
        self.geometry("300x150")

        self.progress_var = tk.DoubleVar()
        self.progressbar = ttk.Progressbar(self, variable=self.progress_var, maximum=100)
        self.progressbar.pack(pady=20)

        self.start_button = ttk.Button(self, text="開始", command=self.start_task)
        self.start_button.pack(side=tk.LEFT, padx=10)

        self.stop_button = ttk.Button(self, text="停止", command=self.stop_task, state=tk.DISABLED)
        self.stop_button.pack(side=tk.RIGHT, padx=10)

        self.task = BackgroundTask(self.update_progress)
        self.task_thread = None

    def start_task(self):
        self.task = BackgroundTask(self.update_progress)
        self.task_thread = threading.Thread(target=self.task.run)
        self.task_thread.start()
        self.start_button.config(state=tk.DISABLED)
        self.stop_button.config(state=tk.NORMAL)

    def stop_task(self):
        if self.task_thread and self.task_thread.is_alive():
            self.task.stop()
            self.task_thread.join()
        self.start_button.config(state=tk.NORMAL)
        self.stop_button.config(state=tk.DISABLED)

    def update_progress(self, value):
        self.progress_var.set(value)
        if value >= 100:
            self.stop_task()

    def on_closing(self):
        self.stop_task()
        self.destroy()

if __name__ == "__main__":
    app = App()
    app.protocol("WM_DELETE_WINDOW", app.on_closing)
    app.mainloop()