複数のプロセスが同じファイルに同時に書き込むと、データが混在したり破損したりする。安全に書き込むにはファイルロックや排他制御が必要だ。
問題の例
# ❌ 複数プロセスが同時に実行すると壊れる def log_message(message): with open('app.log', 'a') as f: f.write(f'{message}\n')
複数プロセスが同時にこの関数を呼ぶと、メッセージが混在したり、一部が上書きされたりする可能性がある。
ファイルロックを使う
filelock ライブラリで排他制御を行う。
pip install filelock
from filelock import FileLock def log_message_safe(message): lock = FileLock('app.log.lock') with lock: with open('app.log', 'a') as f: f.write(f'{message}\n')
ロックを取得したプロセスだけが書き込みでき、他のプロセスは待機する。
logging モジュールを使う
Python 標準の logging モジュールはマルチプロセス対応のハンドラを持っている。
import logging from logging.handlers import RotatingFileHandler # プロセスごとに別のログファイルに書く方法 import os handler = RotatingFileHandler( f'app_{os.getpid()}.log', maxBytes=10*1024*1024, backupCount=5 ) logger = logging.getLogger() logger.addHandler(handler) logger.setLevel(logging.INFO) logger.info('メッセージ')
QueueHandler を使う（推奨）
マルチプロセスで単一のログファイルに書く場合は、QueueHandler と QueueListener を使う。
import logging from logging.handlers import QueueHandler, QueueListener from multiprocessing import Queue def setup_logging(): log_queue = Queue() # ファイルハンドラ（リスナー側） file_handler = logging.FileHandler('app.log') file_handler.setFormatter(logging.Formatter( '%(asctime)s - %(processName)s - %(message)s' )) # リスナーを開始 listener = QueueListener(log_queue, file_handler) listener.start() # 各プロセスはキューに書き込む queue_handler = QueueHandler(log_queue) logger = logging.getLogger() logger.addHandler(queue_handler) logger.setLevel(logging.INFO) return listener listener = setup_logging() # プログラム終了時に listener.stop() を呼ぶ
fcntl を使う（Unix 系）
Unix 系 OS では fcntl で低レベルなファイルロックができる。
import fcntl def append_with_lock(filepath, data): with open(filepath, 'a') as f: fcntl.flock(f.fileno(), fcntl.LOCK_EX) try: f.write(data) finally: fcntl.flock(f.fileno(), fcntl.LOCK_UN)
プロセスごとにファイルを分ける
ロックを避けるシンプルな方法は、プロセスごとに別のファイルに書き、後でマージすることだ。
import os from multiprocessing import Pool def worker(task): pid = os.getpid() with open(f'result_{pid}.txt', 'a') as f: result = process(task) f.write(f'{result}\n') return result # 各ワーカーが別ファイルに書く with Pool(4) as pool: pool.map(worker, tasks) # 後でマージ import glob with open('final_result.txt', 'w') as out: for filepath in glob.glob('result_*.txt'): with open(filepath) as f: out.write(f.read())
アトミック書き込み + ユニークファイル名
各書き込みを別ファイルにし、ディレクトリで管理する方法もある。
import uuid import os def atomic_append(directory, data): # ユニークなファイル名を生成 filename = f'{uuid.uuid4()}.txt' filepath = os.path.join(directory, filename) # アトミック書き込み with open(filepath, 'w') as f: f.write(data)
この方法ならロック不要で、高い並列性を実現できる。
データベースを使う
複雑な並行書き込みが必要な場合は、SQLite などのデータベースを使う方が安全だ。SQLite は適切なロック機構を内蔵している。
import sqlite3 def log_to_db(message): conn = sqlite3.connect('app.db') cursor = conn.cursor() cursor.execute( 'INSERT INTO logs (message, timestamp) VALUES (?, datetime("now"))', (message,) ) conn.commit() conn.close()
