みーのぺーじ

みーが趣味でやっているPCやソフトウェアについて.Python, Javascript, Processing, Unityなど.

シングルトンを実装する

シングルトン とは,そのクラスのインスタンスが1つしか生成されないことを保証するデザインパターンのことであり,アプリケーション全体で1つの共通データがあればよい,という時に重宝します.

シングルトンが必要になったので作ってみました.

  • 通常のクラスとして使用できる.
  • マルチスレッドに対応する.

Pythonで実装

__new__(cls) を利用し,threading.Lock() を追加し,下記のように実装しました.

import threading

class ThreadSafeSingleton():
    _self = None
    _lock = threading.Lock()

    def __new__(cls):
        with cls._lock:
            if cls._self is None:
                cls._self = super().__new__(cls)
        return cls._self

unittestで動作検証します.

import threading
import unittest
from queue import Queue


class ThreadSafeSingleton():
    _self = None
    _lock = threading.Lock()

    def __new__(cls):
        with cls._lock:
            if cls._self is None:
                cls._self = super().__new__(cls)
        return cls._self

    def __init__(self):
        if not hasattr(self, "value"):
            self.value = 0

    def get_value(self):
        return self.value

    def set_value(self, value):
        self.value = value


class TestThreadSafeSingleton(unittest.TestCase):

    def test_singleton(self):
        a = ThreadSafeSingleton()
        b = ThreadSafeSingleton()
        self.assertIs(a, b)
        self.assertEqual(a.get_value(), 0)
        self.assertEqual(b.get_value(), 0)
        a.set_value(4)
        self.assertEqual(a.get_value(), 4)
        self.assertEqual(b.get_value(), 4)

    def test_singleton_in_multithreading(self):
        def worker1(q1):
            q1.put(ThreadSafeSingleton())

        def worker2(q2):
            b = ThreadSafeSingleton()
            b.set_value(1)
            q2.put(b)

        q1 = Queue()
        q2 = Queue()
        t1 = threading.Thread(target=worker1, args=[q1])
        t2 = threading.Thread(target=worker2, args=[q2])
        t1.start()
        t2.start()
        t1.join()
        t2.join()
        a = q1.get()
        b = q2.get()
        self.assertIs(a, b)
        self.assertEqual(a.get_value(), 1)
        self.assertEqual(b.get_value(), 1)

        def worker3():
            c = ThreadSafeSingleton()
            c.set_value(2)

        t3 = threading.Thread(target=worker3)
        t3.start()
        t3.join()
        self.assertEqual(a.get_value(), 2)
        self.assertEqual(b.get_value(), 2)


if __name__ == "__main__":
    unittest.main()

みーの環境では正常に実行されました.

ただし,この方法では,classに直接組み込むので,たくさんのsingletonを使う場合は大変です.デコレータにすると便利で,そういったライブラリーがありました.

singleton-decorator · PyPI

ソースコードを見てみると,

class _SingletonWrapper:
    """
    A singleton wrapper class. Its instances would be created
    for each decorated class. 
    """

    def __init__(self, cls):
        self.__wrapped__ = cls
        self._instance = None

    def __call__(self, *args, **kwargs):
        """Returns a single instance of decorated class"""
        if self._instance is None:
            self._instance = self.__wrapped__(*args, **kwargs)
        return self._instance

def singleton(cls):
    """
    A singleton decorator. Returns a wrapper objects. A call on that object
    returns a single instance object of decorated class. Use the __wrapped__
    attribute to access decorated class directly in unit tests
    """
    return _SingletonWrapper(cls)

こういう方法もありますね.勉強になります.

Rustで実装

Rustにはstaticな変数があるので,わざわざシングルトンを実装する必要がないです.ただし,staticな変数に代入できるデータはかなり制限がある*1ので,lazy-static.rs を使用すると便利です.