如何在Dramatiq的@dramatiq.actor装饰函数中获取worker可读名称

你需要的dramatiq-process-1-worker-3格式的worker名称可以通过读取Dramatiq内置环境变量+提取线程名的方式自行拼接得到，不需要依赖额外的第三方工具，实现逻辑如下：

实现原理 #

• Dramatiq启动worker进程时，会自动为每个子进程注入DRAMATIQ_WORKER_PROCESS_ID环境变量，值就是当前worker进程的编号，从1开始递增
• 每个worker进程内的工作线程，Dramatiq统一命名为WorkerThread-{序号}，序号同样从1开始递增，我们只需要提取线程名末尾的数字即可

完整代码示例 #

import os
import re
import threading
import dramatiq

# 封装成通用工具函数方便所有actor调用
def get_worker_name() -> str:
    # 获取进程编号，本地调试无该变量时默认返回1
    process_id = os.getenv("DRAMATIQ_WORKER_PROCESS_ID", "1")
    # 提取线程序号
    thread_name = threading.current_thread().name
    match_res = re.search(r"WorkerThread-(\d+)", thread_name)
    worker_id = match_res.group(1) if match_res else "0"
    # 拼接成你需要的格式
    return f"dramatiq-process-{process_id}-worker-{worker_id}"

@dramatiq.actor(broker=redis_broker, store_results=True, result_ttl=10*1000)
def ping():
    worker_name = get_worker_name()
    # 直接在日志里打印即可
    print(f"[{worker_name}] 执行ping任务")
    # 如果你用logging模块也可以直接把worker_name作为额外字段传入日志

补充说明 #

如果你只是需要日志里自动携带worker标识，不需要在业务逻辑里获取的话，可以直接修改Dramatiq的日志格式化配置，直接在格式串里加入对应变量即可，不需要额外写代码。该方法兼容Dramatiq所有主流版本，不需要依赖未公开的内部API，稳定性有保障。

内容的提问来源于stack exchange，提问作者Mikko Ohtamaa