线程装饰器 [Python]
Threading Decorator [Python]
我正在尝试使用 python 套接字和线程库创建一个简单的程序。我想使用装饰器使以下过程自动化:
t = threading.Thread(target=function, args=(arg1, arg2))
t.start()
该程序是使用 OOP 构建的,因此我在主程序中定义了一个子程序class 来包含所有装饰器(我在这篇文章中读到了这种方法:https://medium.com/@vadimpushtaev/decorator-inside-python-class-1e74d23107f6)。因此我有这样的情况:
class Server(object):
class Decorators(object):
@classmethod
def threaded_decorator(cls, function):
def inner_function():
function_thread = threading.Thread(target=function)
function_thread.start()
return inner_function
def __init__(self, other_arguments):
# other code
pass
@Decorators.threaded_decorator
def function_to_be_threaded(self):
# other code
pass
但是当我尝试 运行 时,出现以下错误:TypeError: function_to_be_threaded() missing one required argument: 'self'
。我怀疑问题出在我调用 threading.Thread(target=function) 时的部分,它以某种方式没有传递整个函数 self.function_to_be_threaded。因此,如果你知道如何解决这个问题,请告诉我好吗?另外,你能告诉我是否有一种方法可以实现一个接受参数的装饰器,该参数将作为 args=(arguments_of_the_decorator)
传递给线程 class?
非常感谢你的时间,请原谅我的英语,我还在练习
使用 *args
语法移动 arguments.In 换句话说,使用 *args
将所有位置参数收集为一个元组并将其移动 threading.Thread
作为 args
.
import threading
import time
class Server(object):
class Decorators(object):
@classmethod
def threaded_decorator(cls, function):
def inner_function(*args):
function_thread = threading.Thread(target=function,args=args)
function_thread.start()
return inner_function
def __init__(self, count,sleep):
self.count = count
self.sleep = sleep
@Decorators.threaded_decorator
def function_to_be_threaded(self,id):
for xx in range(self.count):
time.sleep(self.sleep)
print("{} ==> {}".format(id,xx))
>>> Server(6,1).function_to_be_threaded('a')
>>> Server(2,3).function_to_be_threaded('b')
a ==> 0
a ==> 1
a ==> 2
b ==> 0
a ==> 3
a ==> 4
a ==> 5
b ==> 1
我正在尝试使用 python 套接字和线程库创建一个简单的程序。我想使用装饰器使以下过程自动化:
t = threading.Thread(target=function, args=(arg1, arg2))
t.start()
该程序是使用 OOP 构建的,因此我在主程序中定义了一个子程序class 来包含所有装饰器(我在这篇文章中读到了这种方法:https://medium.com/@vadimpushtaev/decorator-inside-python-class-1e74d23107f6)。因此我有这样的情况:
class Server(object):
class Decorators(object):
@classmethod
def threaded_decorator(cls, function):
def inner_function():
function_thread = threading.Thread(target=function)
function_thread.start()
return inner_function
def __init__(self, other_arguments):
# other code
pass
@Decorators.threaded_decorator
def function_to_be_threaded(self):
# other code
pass
但是当我尝试 运行 时,出现以下错误:TypeError: function_to_be_threaded() missing one required argument: 'self'
。我怀疑问题出在我调用 threading.Thread(target=function) 时的部分,它以某种方式没有传递整个函数 self.function_to_be_threaded。因此,如果你知道如何解决这个问题,请告诉我好吗?另外,你能告诉我是否有一种方法可以实现一个接受参数的装饰器,该参数将作为 args=(arguments_of_the_decorator)
传递给线程 class?
非常感谢你的时间,请原谅我的英语,我还在练习
使用 *args
语法移动 arguments.In 换句话说,使用 *args
将所有位置参数收集为一个元组并将其移动 threading.Thread
作为 args
.
import threading
import time
class Server(object):
class Decorators(object):
@classmethod
def threaded_decorator(cls, function):
def inner_function(*args):
function_thread = threading.Thread(target=function,args=args)
function_thread.start()
return inner_function
def __init__(self, count,sleep):
self.count = count
self.sleep = sleep
@Decorators.threaded_decorator
def function_to_be_threaded(self,id):
for xx in range(self.count):
time.sleep(self.sleep)
print("{} ==> {}".format(id,xx))
>>> Server(6,1).function_to_be_threaded('a')
>>> Server(2,3).function_to_be_threaded('b')
a ==> 0
a ==> 1
a ==> 2
b ==> 0
a ==> 3
a ==> 4
a ==> 5
b ==> 1