企业官网建设流程全解析

Python を書いていると、__new__という特殊メソッドに一度はぶつかります。名前は知っていても、実際に自分のコードでオーバーライドする機会は意外と少ない。私自身、これまで Python でバックエンドのバッチ処理や並行処理を書いてきた中で、__new__と「スレッド環境」という組み合わせに何度も悩まされてきました。とくに「シングルトンパターンをスレッドセーフに実装する」とか「スレッドごとに異なるオブジェクトを生成したい」といった要件が重なるところで、__new__の正しい理解が一気に効いてきます。

この記事では、__new__の役割を整理しつつ、マルチスレッド環境でそれを安全に使うための設計パターン、競合状態やデッドロックを避ける具体的な実装、スレッドローカル環境との組み合わせまでを解説します。対象読者は、「シングルトンを書いたらデータが混線した」「__new__をオーバーライドしたいけど何を気をつければいいかわからない」といった中級者の方です。コード例はコピーしてそのまま動くものを中心にしています。

1.__new__の役割とオブジェクト生成の流れ

1.1 インスタンス生成は__init__だけではない

Python でオブジェクトを作るとき、私たちが普通に書くクラス定義にはコンストラクタ相当の処理として__init__を実装します。しかし、実際のインスタンス生成は__init__が先に走るわけではありません。Python のオブジェクト生成の内部順序は次のようになっています。

1. __new__が呼ばれ、新しいインスタンスを生成する
2. 生成されたインスタンスを引数として__init__が呼ばれ、初期化する

__new__は静的メソッドに近い性質を持ち、第一引数がクラス自体である点が特徴です。通常はobject.__new__(cls)を呼び出してインスタンスを得て、必要に応じて属性を付与してから返します。要するに、__インスタンスを「作る」のが__new__、作られたインスタンスを「整える」のが__init__という役割分担です。

この順序を理解していないと、__new__内で初期化処理をがんばって書いてしまうことがあります。それ自体は動きますが、__init__が呼ばれたときに二重に初期化されるケースも出てくるので、役割を常に意識しておく必要があります。

1.2 型チェックと不変オブジェクトの扱い

__new__は、intやstr、tupleのような不変型のサブクラスを作るときにも使われます。不変オブジェクトは生成後に値を変更できないため、初期値を設定するタイミングが__new__しかありません。たとえばintを継承して新しい型を作る場合、__new__で値を設定してからインスタンスを返す必要があります。

class PositiveInt(int): def __new__(cls, value): if value <= 0: raise ValueError("value must be positive") return super().__new__(cls, value) p = PositiveInt(10) print(p + 1) # 11

このコードは、__init__を実装していませんが正常に動きます。intのサブクラスは生成後に内部値を変更できないため、値の検証と設定を__new__内で完結させるのが正しい書き方です。こうしたケースでは、__new__が唯一のフックになります。

1.3 なぜ__new__が「スレッド」と関係するのか

__new__がスレッド環境と深く関わる代表例は、シングルトンパターンです。通常のクラスでは、MyClass()と呼ぶたびに新しいオブジェクトが生成されますが、アプリケーション全体で共有したい設定オブジェクトや DB コネクションのようなリソースは「唯一のインスタンス」に制限したいことがあります。そこで__new__内でインスタンスの生成を制御し、既に生成済みのものを返すという実装をします。

このとき、複数のスレッドが同時にMyClass()を呼ぶと、__new__内の「インスタンスがまだ存在するかチェックして、なければ生成する」という処理が競合状態（レースコンディション）に陥ります。実は、これこそがタイトルにある「__new__环境线程」の正体です。スレッド環境で__new__を安全に使うには、原子性を確保する仕組みが必要になります。

2. マルチスレッド環境での__new__：競合状態とスレッドセーフな実装

2.1 GILだけでは防げない競合状態

Python には GIL（Global Interpreter Lock）があるため、複数のスレッドが同時に Python バイトコードを実行することは原則としてできません。そのため、「GILがあるから__new__の競合も起きないのでは」とよく誤解されますが、実際にはそんなことはありません。

GIL が保証するのは「一定間隔でスレッドが切り替わる」というスケジューリングの安全性だけで、複数バイトコード命令にまたがる処理の一貫性は保証されません。具体的には、次のようなコードが危険です。

class Singleton: _instance = None def __new__(cls): if cls._instance is None: # ここでスレッド切り替えが起き得る cls._instance = super().__new__(cls) return cls._instance

一見問題なさそうに見えますが、cls._instance is Noneのチェックとcls._instance = ...の代入の間にスレッド切り替えが入ると、両方のスレッドが別々のインスタンスを生成し、最後に代入したスレッドのインスタンスだけが生き残ります。結果として、呼び出し元によっては異なるインスタンスが返る可能性があります。

2.2 ロックを使った正しいシングルトン実装

競合状態を確実に避けるには、threading.Lockで__new__全体を保護するのが基本です。ここでは私が実際に使っている実装例を示します。

import threading class Singleton: _instance = None _lock = threading.Lock() def __new__(cls): if cls._instance is not None: return cls._instance with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance

この実装では、まずロックを取らずに速いチェックを行い、すでにインスタンスが存在する場合はそのまま返します。存在しない場合だけロックを取得し、その中でもう一度チェックしてから生成します。この「二重チェックロッキング」パターンは、Java のシングルトン実装で有名ですが、Python でもそのまま使えます。

二重チェックが必要な理由は、「先にロックを取ってから生成する」だけでは不十分な場合があるからではありません。実は、with cls._lock:で囲めば単一チェックでも安全に生成できます。ただし、インスタンス生成後に毎回ロックを取得するのは無駄なので、高速パスとして最初のチェックを入れるのが定番です。

2.3 メタクラスを使ったシングルトンの別解

__new__を直接書く代わりに、メタクラスの__call__をフックする方法もあります。クラスを呼び出すとtype.__call__が動き、その中で__new__と__init__が呼ばれます。この__call__レベルでシングルトン化すると、__new__を書き換えずに済みます。

import threading class SingletonMeta(type): _instances = {} _lock = threading.Lock() def __call__(cls, *args, **kwargs): if cls in cls._instances: return cls._instances[cls] with cls._lock: if cls not in cls._instances: instance = super().__call__(*args, **kwargs) cls._instances[cls] = instance return cls._instances[cls] class Database(metaclass=SingletonMeta): def __init__(self): self.connection_id = id(self)

メタクラス版は、シングルトンにしたい複数のクラスで共通の仕組みを使い回せる利点があります。cls._instancesを辞書にしてクラスごとにインスタンスを管理しているので、DatabaseクラスとConfigクラスを両方シングルトン化したい場合にも、同じメタクラスを使い回せます。

2.4 GILの存在を前提とした簡易チェックの罠

GILがある以上、「アトミックな操作だけで済むならロックは不要」と考えることもできます。たとえば、不変オブジェクトを一度だけ設定するような場面では、_instanceへの代入自体は GIL の下でアトミックに見えます。ただし、厳密には「参照の読み書きはアトミックに見えるが、オブジェクト生成の順序が外部から観測できる」というメモリモデルの問題があります。

正直なところ、Python のメモリモデルは完全に標準化されておらず、実装依存の部分があります。私の経験則として、シングルトン生成に限っては、ロックを使うのが確実で、簡易チェックで頑張る必要はないと考えています。ロック取得のコストはコンテキストスイッチやキャッシュミスの影響でそれなりにありますが、シングルトン生成はクラスごとに一度しか起きない処理なので、性能上の問題にはほぼなりません。

3. スレッド環境の「環境」を作る：スレッドローカルと__new__

3.1 スレッドごとに異なるオブジェクトを返す要件

シングルトンは「グローバルで唯一」ですが、スレッド環境では「スレッドごとに唯一」という要件もよく出てきます。たとえば、Web アプリでリクエストごとにユーザー情報を保持したい場合、スレッドプールで処理が走るため、スレッド間でデータを共有すると別ユーザーの情報が混線します。こういうとき使うのがthreading.localです。

threading.localはスレッドごとに独立した名前空間を持つオブジェクトで、同じオブジェクトを参照しても、スレッドごとに異なる属性値を持つことができます。これを__new__で返すクラスと組み合わせると、「各スレッドが同じクラス名なのに、実体はスレッド固有のインスタンスを持つ」という動作を作り込めます。

3.2__new__とthreading.localを組み合わせた設計

スレッドローカルなシングルトンを__new__で実装する場合、次のような形になります。

import threading class ThreadLocalService: _local = threading.local() def __new__(cls): # スレッドごとのタグを持ったインスタンスを作る instance = super().__new__(cls) instance.thread_name = threading.current_thread().name return instance @classmethod def current(cls): # スレッドローカルで共有するための入れ物 if not hasattr(cls._local, "service"): cls._local.service = cls() return cls._local.service

この場合、ThreadLocalService()を呼ぶたびに新しいインスタンスが返ります。current()を経由して取得すると、そのスレッド内では同じインスタンスを再利用できます。スレッドが異なれば別のインスタンスになるため、スレッドごとの分離ができます。

ここで重要なのは、threading.localのオブジェクトをクラス変数として持つことです。クラス変数は全スレッドから参照されますが、内部の属性はスレッドごとに独立しています。

3.3 スレッドローカルを__new__内で直接参照するパターン

「環境」を強く意識した設計として、__new__の中で直接threading.localを参照するパターンもあります。たとえば、同じ変数名でスレッドごとの設定を返したい場合です。

import threading class RequestContext: _local = threading.local() def __new__(cls): instance = super().__new__(cls) instance.request_id = getattr(cls._local, "request_id", None) return instance

こうすると、RequestContext()を呼ぶだけで、現在のスレッドに設定されたrequest_idを自動的に保持したオブジェクトが返ります。このとき、_local.request_idが設定されていなければデフォルト値のNoneが入ります。

ただし、注意すべき点があります。threading.localの値を__new__内で参照する場合、その値はスレッドプールが再利用される状況では前のリクエストの残骸が残っていることがあります。Web フレームワークのミドルウェアやリクエスト処理の開始時点で、きちんと初期化・クリーンアップする仕組みが必須です。

3.4 Python 3.7以降のcontextvarsとの住み分け

threading.localと似た用途で、Python 3.7 から標準ライブラリに入ったcontextvarsがあります。contextvarsは、非同期処理やジェネレータを含む実行コンテキストを追跡するための仕組みで、スレッドの切り替えではなく「コンテキストの切り替え」で値を分離します。

threading.localとcontextvarsの大きな違いを表にまとめます。

私の運用方針としては、純粋に「スレッド単位で閉じた処理」をしたい場合はthreading.localを使い、asyncio のタスクごとにコンテキストを分離したい場合はcontextvarsを選びます。__new__と組み合わせる場合も、どちらの仕組みを使うかによって「環境」の単位が変わる点を最初に明確にしておくのがコツです。

4. 実践：スレッドプールと__new__を組み合わせた設計例

4.1 シナリオ：タスクごとのコンテキストと共有リソース

ここまで説明した内容をまとめて、もう少し実用的な設計例を示します。想定シナリオは次のとおりです。

• スレッドプールを使って複数のタスクを並行処理する
• 各タスクは独自のrequest_idを持ち、ログやDBコネクションでそのIDを使いたい
• DBコネクション自体はアプリ全体で共有したい（シングルトン）

この要件を満たすには、「タスクごとのコンテキスト」と「アプリ全体のシングルトン」を混在させて扱う必要があります。ここでは__new__を軸に、シングルトンとスレッドローカルを分離して実装します。

4.2 シングルトンで共有するDBコネクション

まずは、アプリ全体で共有する DB コネクションです。スレッドセーフなシングルトンとして実装します。

import threading class DatabaseConnection: _instance = None _lock = threading.Lock() def __new__(cls, *args, **kwargs): if cls._instance is not None: return cls._instance with cls._lock: if cls._instance is None: instance = super().__new__(cls) instance.connection = cls._create_connection() cls._instance = instance return cls._instance @staticmethod def _create_connection(): # 実際のDB接続の代わり return {"connected": True, "connection_id": id(object())} db1 = DatabaseConnection() db2 = DatabaseConnection() print(db1 is db2) # True

__new__内で_create_connection()という重い処理を呼んでいますが、ロックの中なのでスレッドが競合しても一度しか実行されません。ただし、DatabaseConnection()の呼び出し側に*argsや**kwargsを渡すと、シングルトンとして同じインスタンスを返すたびに無視されます。この点は仕様として明確にしておく必要があります。呼び出しごとに異なる接続パラメータを渡したい場合は、シングルトンではなくリポジトリパターンに切り替えるべきです。

4.3 タスクごとのコンテキストとログ連携

次に、スレッドプールで動くタスクごとにrequest_idを保持するコンテキストを用意します。ここでは、threading.localを内包したクラスを使います。

import threading import contextlib import uuid class RequestContext: _local = threading.local() def __init__(self, request_id): self.request_id = request_id self.thread_name = threading.current_thread().name @classmethod def get_current(cls): if not hasattr(cls._local, "current"): cls._local.current = cls("unknown") return cls._local.current @classmethod @contextlib.contextmanager def set_request_id(cls, request_id): previous = getattr(cls._local, "current", None) cls._local.current = cls(request_id) try: yield cls._local.current finally: if previous is None: del cls._local.current else: cls._local.current = previous def worker(task_id): with RequestContext.set_request_id(f"req-{task_id}"): ctx = RequestContext.get_current() db = DatabaseConnection() print(f"{ctx.thread_name} -> {ctx.request_id} / db={db.connection['connection_id']}") with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: list(executor.map(worker, range(5)))

この設計では、RequestContext.set_request_idというコンテキストマネージャがスレッドローカルに現在のリクエストIDを設定し、タスク終了時に元の状態へ戻します。threading.localを直接外から触らずに、RequestContextクラスのクラスメソッド経由でアクセスする意図があります。

__new__はここでは使っていません。コンテキスト生成に__init__が素直で、あえて__new__で制御する必要がないからです。逆に、__new__を使うべき場面は「インスタンスそのものを制御したい」ときであり、コンテキスト情報の付け替えはthreading.local側の役割です。

4.4 なぜ__new__で全部やろうとしないのか

私が経験した失敗の一つに、「スレッドローカルなシングルトンを全て__new__で実現しようとして、コードが複雑化した」というものがあります。__new__でインスタンスを制御すると、スレッドごとに異なるインスタンスを返すことができますが、その条件が増えるほどコードの分岐が増えます。

# これは「動作はするが、なぜこうなるのか」が分かりにくい悪例 class ComplexNew: _global_instance = None _local = threading.local() def __new__(cls, mode="global"): if mode == "global": if cls._global_instance is None: cls._global_instance = super().__new__(cls) return cls._global_instance else: if not hasattr(cls._local, "instance"): cls._local.instance = super().__new__(cls) return cls._local.instance

modeという引数で挙動を切り替えるこの実装は、呼び出し方次第で同じクラス名なのに別インスタンスが返るため追跡が困難です。__new__の制御は「単一のルール」に絞ったほうが、後に読む人（未来の自分も含む）にとって安全です。

私の設計指針は次のとおりです。

• グローバルに唯一：__new__によるシングルトンパターン
• スレッドごとに独立：threading.local＋ クラスメソッドのコンテキスト管理
• 両者の混在：シングルトンは共通インフラ、コンテキストはスレッドローカル、という分離

5. はまりどころとトラブルシューティング

5.1__new__内でのデッドロック

__new__の中で別のクラスの__new__を呼び出し、その相手もロックを取得しようとすると、デッドロックが発生することがあります。具体例としては、__new__内でログ出力クラスを生成し、そのログクラスが自前のロックを取る場合です。ロックの順序が逆転すると、互いに相手のロック解放を待って停止します。

デッドロックを避ける基本は、ロックの取得順を一定に保つことと、__new__内でできるだけ重い処理をしないことです。シングルトンの初期化処理が複雑で、他のオブジェクトと相互参照がある場合は、初期化を遅延させるか、別の初期化メソッド（initialize()やconnect()）に分離する方が安全です。

5.2threading.localのリセット忘れによるデータ混線

スレッドプールを使っていると、同じスレッドが複数のタスクを連続して処理します。このとき、前のタスクで設定したthreading.localの値が残っていると、次のタスクが誤った値を見てしまいます。

my 実務では、threading.localを扱うときは「値を設定したら、必ず元に戻す or 削除する」を徹底しています。先ほどのcontextmanagerを使う実装で、try/finallyで確実にクリーンアップしているのはこのためです。

@classmethod @contextlib.contextmanager def set_request_id(cls, request_id): previous = getattr(cls._local, "current", None) cls._local.current = cls(request_id) try: yield finally: if previous is None: del cls._local.current else: cls._local.current = previous

このパターンを使えば、設定し忘れは基本的に起きません。ただし、コンテキストマネージャの外側で直接cls._local.current = ...と代入している箇所が残っていると、そこが漏れます。プロジェクト内で「threading.localへの直接アクセスは禁止」というルールにするのも有効です。

5.3__new__が生成したオブジェクトの参照が解放されない問題

__new__の戻り値がクラス変数に保存されるシングルトンは、プロセスが終了するまで生き続けます。これは意図どおりですが、そのインスタンスが巨大なキャッシュやファイルハンドルを持っていると、メモリリークのような状態になります。

シングルトンであっても、不要になったら明示的にリソースを解放するコードを用意しておくのが実践的です。

class DatabaseConnection: # ... 省略 ... def close(self): if getattr(self, "_closed", False): return self._closed = True DatabaseConnection._instance = None

_instanceをNoneに戻すことで、次回呼び出し時に新しく接続を作り直せます。テストコードや長時間稼働するバッチ処理では、この「作り直し」の動線が地味に効きます。

5.4 よくある質問と簡易まとめ

__init__が毎回呼ばれる問題は意外と見落としがちです。シングルトン化したクラスに__init__を書いてしまうと、同じインスタンスを返した後も、呼び出すたびに__init__内の初期化処理が再実行されてしまいます。これが原因で「値を設定したのにリセットされた」という事象が起きます。シングルトンクラスでは、初期化を__new__か明示的なinitialize()メソッドに寄せるのが無難です。

6. 個人的な実装チェックリストと最後に

私自身が__new__をスレッド環境で使うとき、最終的に確認しているポイントをまとめておきます。

• [ ]__new__で制御する対象が「インスタンス生成自体」であり、単なる初期化処理ではないか
• [ ] 複数スレッドが同時に呼び出しても競合しないか（ロック or 二重チェック）
• [ ] シングルトンに__init__を実装していないか（毎回呼ばれる問題）
• [ ] スレッドごとのデータが必要ならthreading.localやcontextvarsを適切に選んでいるか
• [ ] スレッドプール再利用時にthreading.localの残骸が残らないか

最後に、私が実際に経験した小さなテクニックをひとつ。スレッドセーフなシングルトンの動作確認をするときは、concurrent.futures.ThreadPoolExecutorで大量に生成処理を呼び、返ってきたオブジェクトのid()がすべて一致することを確認するのが手軽です。time.sleepを挟みながら実行すると、より再現性高く競合を検出できます。__new__の面白さは、言語のオブジェクト生成という「一番根っこ」に手を入れられる点にあります。とくにスレッド環境だと、そこに並行性の難しさも重なってきます。この記事の内容が、そうした難しさを乗り越える手助けになれば幸いです。