huey是一个轻量的任务队列,支持使用sqlite,适合用于在Linux/Windows多平台的软件使用
huey的相关教程较少
下一篇文章大概会放出,在嵌入到单个软件项目中使用的demo。
现象
在初始化ConsumerConfig时有两种写法,可以看看下面的demo:
from huey.consumer_options import ConsumerConfig
conf = {"workers":2, # 消费者数量
"worker_type":'thread', # 消费者类型
"initial_delay":60, # 最小的轮询间隔,与-d相同。
"backoff":2, # 指数退避使用这个速率,-b.
"max_delay":100, # 可能的最大轮询间隔, -m.
"periodic":True, # 允许周期性任务
"check_worker_health":True, # 启用工人健康检查。
"health_check_interval":1, # 指定了健康检查的轮询间隔
"utc":True, # 将时间戳转换为UTC时间
"consumer_timeout":10, # 消费者超时时间为10秒
}
config = ConsumerConfig(
workers=2, # 消费者数量
worker_type='thread', # 消费者类型
initial_delay=60, # 最小的轮询间隔,与-d相同。
backoff=2, # 指数退避使用这个速率,-b.
max_delay=100, # 可能的最大轮询间隔, -m.
periodic=True, # 允许周期性任务
check_worker_health=True, # 启用工人健康检查。
health_check_interval=1, # 指定了健康检查的轮询间隔
utc=True, # 将时间戳转换为UTC时间
consumer_timeout=10, # 消费者超时时间为10秒
)
def worker(**x):
print(x)
if __name__ == '__main__':
worker(**conf)
worker(**config.values)
安装完成环境后两种写法都对,但运行输出分别为:
{'workers': 2, 'worker_type': 'thread', 'initial_delay': 60, 'backoff': 2, 'max_delay': 100, 'periodic': True, 'check_worker_health': True, 'health_check_interval': 1, 'utc': True, 'consumer_timeout': 10}
{'workers': 2, 'worker_type': 'thread', 'initial_delay': 60, 'backoff': 2, 'max_delay': 100, 'check_worker_health': True, 'health_check_interval': 1, 'scheduler_interval': 1, 'periodic': True, 'flush_locks': False, 'extra_locks': None}
可以观察到,worker(config.values)的输出内容比worker(conf)要多。
但这是为什么呢?
原因
worker(**conf)
worker(**conf)
这个还是比较好理解的,毕竟是python的标准写法:
通过**声明的参数也是可以传入多个参数,但是传入的参数类型需要为字典的类型,且参数在函数内部将被存放在以形式名为标识符的dictionary中,这种方法在需要声明多个默认参数的时候特别有用。
worker(**config.values)
worker(**config.values)
就不太一样了,config是在类里面定义的对象,理论上应该是用vars() 函数,毕竟:
但当你实际使用vars() 函数会发现并不会起作用,而参考huey的demo:huey_consumer.py,可以看到作者是这样写的:
# 将 options 对象中的非空属性值提取出来,组成一个字典,并将结果赋值给 options 变量。
options = {k: v for k, v in options.__dict__.items()
if v is not None}
# 创建一个 ConsumerConfig 类的实例对象,使用 options 字典中的键值对参数进行初始化。
config = ConsumerConfig(**options)
# 调用 ConsumerConfig 实例对象的 validate 方法,用于验证参数的合法性。
config.validate()
# 根据 args 列表中的第一个元素加载一个 Huey 实例,并将结果赋值给 huey_instance 变量。
huey_instance = load_huey(args[0])
# 创建一个名为 'huey' 的 logger 对象。
logger = logging.getLogger('huey')
# 调用 ConsumerConfig 实例对象的 setup_logger 方法,用于设置 logger 对象的日志记录级别等参数。
config.setup_logger(logger)
# 调用 huey_instance 对象的 create_consumer 方法,创建一个消费者对象,并使用 config.values 中的键值对参数进行初始化。
consumer = huey_instance.create_consumer(**config.values)
# 调用消费者对象的 run 方法,开始消费消息。
consumer.run()
通过查看from huey.consumer_options import ConsumerConfig
中,ConsumerConfig
这个类所在的文件,我们能在末尾搜到:
@property
def values(self):
return dict((key, getattr(self, key)) for key in config_keys
if key not in ('logfile', 'verbose', 'simple_log'))
因此,就可以知道,在这里,通过config.values
将config
实例转化为字典,然后再用常规的**
,即**config.values
将该字典传入函数中,即可实现参数的传递。
而通过实例进行传参,可以有效的传递未定义的形参,避免再在from huey.consumer import Consumer
后,初始化Consumer
类的实例的时候导入不完整,这样写更标准。(虽然两种写法均不会报错)
拓展
但当我故意引入一个错误的参数"x":121212
,可以观察到两种导入方式的更明显的区别了:
from huey.consumer_options import ConsumerConfig
conf = {"workers":2, # 消费者数量
"worker_type":'thread', # 消费者类型
"initial_delay":60, # 最小的轮询间隔,与-d相同。
"backoff":2, # 指数退避使用这个速率,-b.
"max_delay":100, # 可能的最大轮询间隔, -m.
"periodic":True, # 允许周期性任务
"check_worker_health":True, # 启用工人健康检查。
"health_check_interval":1, # 指定了健康检查的轮询间隔
"utc":True, # 将时间戳转换为UTC时间
"consumer_timeout":10, # 消费者超时时间为10秒
"x":121212 # !!!!错误的参数
}
config = ConsumerConfig(
workers=2, # 消费者数量
worker_type='thread', # 消费者类型
initial_delay=60, # 最小的轮询间隔,与-d相同。
backoff=2, # 指数退避使用这个速率,-b.
max_delay=100, # 可能的最大轮询间隔, -m.
periodic=True, # 允许周期性任务
check_worker_health=True, # 启用工人健康检查。
health_check_interval=1, # 指定了健康检查的轮询间隔
utc=True, # 将时间戳转换为UTC时间
consumer_timeout=10, # 消费者超时时间为10秒
x=121212 # !!!!错误的参数
)
def worker(**x):
print(x)
if __name__ == '__main__':
worker(**conf)
worker(**config.values)
运行上面的代码可以得到下面的结果:
{'workers': 2, 'worker_type': 'thread', 'initial_delay': 60, 'backoff': 2, 'max_delay': 100, 'periodic': True, 'check_worker_health': True, 'health_check_interval': 1, 'utc': True, 'consumer_timeout': 10, 'x': 121212}
{'workers': 2, 'worker_type': 'thread', 'initial_delay': 60, 'backoff': 2, 'max_delay': 100, 'check_worker_health': True, 'health_check_interval': 1, 'scheduler_interval': 1, 'periodic': True, 'flush_locks': False, 'extra_locks': None}
可以看到,worker(**config.values)
不会将错误的参数引入到下一步中,通过这个方式增强了代码的健壮性,因此在作者的demo中也用的这个方式来传递参数。
参考:
python 中 *
与 **
的参数传递
此处评论已关闭