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.valuesconfig实例转化为字典,然后再用常规的**,即**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 中 *** 的参数传递

最后修改:2023 年 06 月 22 日
如果觉得我的文章对你有用,请随意赞赏