【Python】multiprocessing.Processで並列処理（マルチプロセス）を実装する

2026年1月14日

概要

PythonでCPU負荷の高い処理（重い計算など）を並列化する場合、スレッド（threading）ではなく、マルチプロセス（multiprocessing）を使用するのが効果的です。

プロセスごとに独立したメモリ空間とPythonインタプリタを持つため、GIL（Global Interpreter Lock）の制約を受けずに、マルチコアCPUの性能をフルに引き出すことができます。

本記事では、プロセスの生成、引数の渡し方、終了待機（join）、デーモン化などの基本操作を解説します。

仕様（入出力）

multiprocessing.Process のパラメータ

パラメータ	型	意味
target	callable	プロセス起動時に実行される関数オブジェクト。
args	tuple	`target` 関数に渡す位置引数のタプル。要素が1つの場合もカンマ必須（例: `(1,)`）。
kwargs	dict	`target` 関数に渡すキーワード引数の辞書。
daemon	bool	`True` にするとデーモンプロセスとなり、メインプロセス終了時に道連れで強制終了されます。

主要メソッド

メソッド	説明
start()	プロセスを生成し、`target` 関数の実行を開始します。
join(timeout=None)	プロセスの終了を待ちます。呼び出さないとメイン処理が先に終わってしまう場合があります。

基本の使い方

関数を定義し、それを target に指定して Process インスタンスを作成します。

start() で開始し、join() で完了を待ち合わせます。

# 引数をタプルで渡す基本形
p = multiprocessing.Process(target=my_func, args=("値1",))
p.start()
p.join()

コード全文

「数値をカウントする処理」と「文字を出力する処理」を別々のプロセスで並列に実行する例です。

args（位置引数）と kwargs（キーワード引数）の両方の渡し方を記載しています。

import multiprocessing
import time
import os

def print_numbers(process_name: str, limit: int):
    """
    指定された回数だけ数字を出力する関数
    """
    print(f"[{process_name}] PID: {os.getpid()} 処理開始")
    for i in range(limit):
        print(f"  {process_name}: {i}")
        time.sleep(0.5)
    print(f"[{process_name}] 処理終了")

def print_letters(process_name: str, letters: list):
    """
    リスト内の文字を順に出力する関数
    """
    print(f"[{process_name}] PID: {os.getpid()} 処理開始")
    for char in letters:
        print(f"  {process_name}: {char}")
        time.sleep(0.7)
    print(f"[{process_name}] 処理終了")

def main():
    print(f"メインプロセス PID: {os.getpid()} 開始")

    # プロセス1: args（位置引数タプル）で渡す例
    # 注意: 要素が1つの場合でも (val, ) のようにカンマが必要です
    p1 = multiprocessing.Process(
        target=print_numbers,
        args=("NumProc", 3)
    )

    # プロセス2: kwargs（キーワード引数辞書）で渡す例
    p2 = multiprocessing.Process(
        target=print_letters,
        kwargs={"process_name": "CharProc", "letters": ["A", "B", "C"]}
    )

    # プロセスの開始
    p1.start()
    p2.start()

    print("--- プロセスを実行中 ---")

    # プロセスの終了待機（これがないとメインプロセスが先に終わる）
    p1.join()
    p2.join()

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

if __name__ == "__main__":
    # Windows/macOSでは、このガードブロックが必須です
    main()

実行結果例

2つのプロセスが並行して動いているため、出力が混ざり合います。PID（プロセスID）がそれぞれ異なる点に注目してください。

メインプロセス PID: 12345 開始
--- プロセスを実行中 ---
[NumProc] PID: 12346 処理開始
[CharProc] PID: 12347 処理開始
  NumProc: 0
  CharProc: A
  NumProc: 1
  CharProc: B
  NumProc: 2
[NumProc] 処理終了
  CharProc: C
[CharProc] 処理終了
全ての処理が終了しました

カスタムポイント

引数の渡し方
- args=("hoge",): 引数が1つの場合、タプルのカンマを忘れがちなので注意してください。
- kwargs={"key": "value"}: 引数が多い場合や、デフォルト引数を上書きする場合はこちらが可読性が高くおすすめです。
プロセスのリスト管理
- 大量のプロセスを起動する場合は、processes = [] リストに格納し、ループで start() と join() を行うのが一般的です。

注意点

if __name__ == "__main__": の必須性
- WindowsやmacOSでは、新しいプロセスを立ち上げる際にモジュール全体が再インポートされます。このガードブロックがないと、無限にプロセスが生成され続け（再帰爆発）、エラーになります。
メモリ空間の独立
- グローバル変数を書き換えても、それは「そのプロセス内のコピー」を書き換えているだけです。プロセス間でデータを共有するには Queue や Value などの専用機能が必要です。
ゾンビプロセス
- join() せずに親プロセスが長時間稼働し続けると、終了した子プロセスがリソースに残る場合があります。確実に join() するか、コンテキストマネージャ等を利用してください。

応用

デーモンプロセス (daemon=True) の使用例です。

メインプロセスが終了すると、デーモンプロセスは作業途中でも強制終了されます。「ログ監視」や「ヘルスチェック」など、メイン処理の終了と共に消えても良いタスクに使います。

import multiprocessing
import time

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

def main_daemon():
    # daemon=True に設定
    p = multiprocessing.Process(target=background_task, daemon=True)
    p.start()

    print("メイン処理を実行中(2秒間)...")
    time.sleep(2.0)
    
    print("メイン処理終了。デーモンもここで強制終了されます。")
    # p.join() を呼ばずに終了する

if __name__ == "__main__":
    main_daemon()