Skip to content

Instantly share code, notes, and snippets.

@victorusachev
Created April 18, 2020 01:00
Show Gist options
  • Select an option

  • Save victorusachev/e2aa328e2125a8a16b019e46d836a201 to your computer and use it in GitHub Desktop.

Select an option

Save victorusachev/e2aa328e2125a8a16b019e46d836a201 to your computer and use it in GitHub Desktop.
import functools
import threading
from time import sleep
lock = threading.Lock()
class SingletonMeta(type):
_instance = None
_args = None
_kwargs = None
_lock = None
def __new__(mcs, name, bases, attrs, **kwargs):
cls = super().__new__(mcs, name, bases, attrs)
cls._lock = threading.Lock()
return cls
def __call__(cls, *args, **kwargs):
if not cls._instance:
print(f'{cls.__name__}: ожидание блокировки')
with cls._lock:
print(f'{cls.__name__}: взятие блокировки')
sleep(5)
if not cls._instance:
cls._args = args
cls._kwargs = kwargs
cls._instance = super(SingletonMeta, cls).__call__(*args, **kwargs)
print(f'{cls.__name__}: создание нового экземпляра')
print(f'{cls.__name__}: освобождение блокировки')
else:
print(f'{cls.__name__}: использование существующего экземпляра')
if not isinstance(cls._instance, cls):
raise TypeError
if cls._args != args or cls._kwargs != kwargs:
raise ValueError
return cls._instance
class Singleton(metaclass=SingletonMeta):
"""Синглтон на основе блокировки с двойной проверкой"""
def test_singleton(classes):
threads = [
threading.Thread(target=cls)
for cls in classes
]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
def make_singleton_classes(names):
return [
type(f'Class{name}', (Singleton,), {})
for name in names
]
if __name__ == '__main__':
classes = make_singleton_classes('ABC')
test_singleton(classes * 2)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment