Python中如何利用rq库的get_current_job()函数获取当前任务信息

在使用 rq 库时，可以使用 get_current_job() 函数获取当前任务的信息。get_current_job() 函数返回一个 Job 实例，可以通过该实例获取当前任务的相关属性和状态信息。

下面是一个使用例子，包含了创建任务函数、任务执行函数和获取任务信息函数：

from rq import Queue
from rq.job import Job
from rq.worker import Worker
from redis import Redis
import time


def create_task(queue_name, job_name, job_data):
    redis_conn = Redis()
    queue = Queue(name=queue_name, connection=redis_conn)
    job = queue.enqueue(job_function, job_name, job_data)
    return job


def job_function(job_name, job_data):
    # 获取当前任务信息
    current_job = get_current_job()
    print(f"Current job ID: {current_job.id}")
    print(f"Job name: {job_name}")
    print(f"Job data: {job_data}")

    # 模拟任务执行
    time.sleep(5)
    result = job_data * 2

    return result


def get_job_info(job_id):
    redis_conn = Redis()
    job = Job.fetch(job_id, connection=redis_conn)
    return {
        "ID": job.id,
        "Status": job.get_status(),
        "Result": job.result if job.is_finished else None,
        "Enqueued At": job.enqueued_at,
        "Started At": job.started_at,
        "Ended At": job.ended_at
    }

在这个例子中，create_task() 函数用于创建一个新的任务并将其加入到指定的队列中。该函数接受三个参数：队列名称、任务名称和任务数据。它使用 enqueue() 方法将任务加入队列，并返回创建的任务实例。

job_function() 函数是一个实际的任务执行函数。它接受任务名称和任务数据作为参数，并在执行过程中通过 get_current_job() 获取当前任务的信息。然后，它通过模拟一个耗时操作来执行任务，并返回结果。

get_job_info() 函数用于获取指定任务的信息。它接受任务 ID 作为参数，并使用 Job.fetch() 方法从 Redis 中获取任务实例。然后，它从任务实例中获取相关属性，如任务状态、结果、入队时间、开始时间和结束时间，并返回一个包含这些信息的字典。

下面是一个完整的例子，演示了如何使用上述函数创建、执行和获取任务的信息：

from rq import Queue, get_current_job
from redis import Redis


def create_task(queue_name, job_name, job_data):
    redis_conn = Redis()
    queue = Queue(name=queue_name, connection=redis_conn)
    job = queue.enqueue(job_function, job_name, job_data)
    return job


def job_function(job_name, job_data):
    current_job = get_current_job()
    print(f"Current job ID: {current_job.id}")
    print(f"Job name: {job_name}")
    print(f"Job data: {job_data}")

    # 模拟任务执行
    time.sleep(5)
    result = job_data * 2

    return result


def get_job_info(job_id):
    redis_conn = Redis()
    job = Job.fetch(job_id, connection=redis_conn)
    return {
        "ID": job.id,
        "Status": job.get_status(),
        "Result": job.result if job.is_finished else None,
        "Enqueued At": job.enqueued_at,
        "Started At": job.started_at,
        "Ended At": job.ended_at
    }


# 创建任务并加入队列
job = create_task("my_queue", "MyJob", 10)

# 获取任务信息
job_info = get_job_info(job.id)
print("Job info:")
for key, value in job_info.items():
    print(f"{key}: {value}")

在上面的例子中，首先使用 create_task() 函数创建了一个名为 "MyJob" 的任务，并将其加入到名为 "my_queue" 的队列中。然后，使用 get_job_info() 函数获取了该任务的信息，并打印出来。

这是一个基本的使用例子，展示了如何利用 rq 库的 get_current_job() 函数获取当前任务的信息，并使用相应的函数创建、执行和获取任务的信息。你可以根据自己的实际需求对代码进行进一步的修改和完善。