Django & Celery 使用单例 class 实例属性作为全局
Django & Celery use a singleton class instance attribute as a global
我想这样做:
我在同一个模块(甚至同一个文件)中有 2 个函数:
def a():
while(True):
//do something
if global_var:
//do something else!
def b():
global_var = some_function_result
我想到了使用单例 class 作为全局存储的想法。
(我确实尝试过使用模块级全局,结果相同)
class Singleton(type):
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
return cls._instances[cls]
class MyClass(object):
__metaclass__ = Singleton
def __init__(self):
self.flag = 1
@shared_task
def add_var():
myclass = MyClass()
while(1):
myclass.flag += 1
print myclass.flag
@shared_task
def print_var():
myclass = MyClass()
while(1):
print myclass.flag
结果:
print_var 一直打印 1 并且 add_var 一直加 1 但它没有反映在 print_var
编辑:
错过了重要信息:
我在 celery 上 运行 处理这些进程 - 现在我开始明白 celery 和 django 运行 在不同的线程上。
但是当我运行都在芹菜里的时候,还是没看到效果。
如果这就是您想要实现的全部,那么您不需要单例;一个(静态)class 属性将完成这项工作:
class MyClass(object):
FLAG = 1
@shared_task
def add_var():
myclass = MyClass()
while(1):
myclass.FLAG += 1
print( myclass.FLAG )
@shared_task
def print_var():
myclass = MyClass()
while(1):
print( myclass.FLAG )
我想这样做:
我在同一个模块(甚至同一个文件)中有 2 个函数:
def a():
while(True):
//do something
if global_var:
//do something else!
def b():
global_var = some_function_result
我想到了使用单例 class 作为全局存储的想法。
(我确实尝试过使用模块级全局,结果相同)
class Singleton(type):
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
return cls._instances[cls]
class MyClass(object):
__metaclass__ = Singleton
def __init__(self):
self.flag = 1
@shared_task
def add_var():
myclass = MyClass()
while(1):
myclass.flag += 1
print myclass.flag
@shared_task
def print_var():
myclass = MyClass()
while(1):
print myclass.flag
结果:
print_var 一直打印 1 并且 add_var 一直加 1 但它没有反映在 print_var
编辑:
错过了重要信息: 我在 celery 上 运行 处理这些进程 - 现在我开始明白 celery 和 django 运行 在不同的线程上。 但是当我运行都在芹菜里的时候,还是没看到效果。
如果这就是您想要实现的全部,那么您不需要单例;一个(静态)class 属性将完成这项工作:
class MyClass(object):
FLAG = 1
@shared_task
def add_var():
myclass = MyClass()
while(1):
myclass.FLAG += 1
print( myclass.FLAG )
@shared_task
def print_var():
myclass = MyClass()
while(1):
print( myclass.FLAG )