6

如果这是一个愚蠢的问题，我道歉并且会羞愧地掩饰我的头，但是：

我正在使用 rq 在 Python 中对作业进行排队。我希望它像这样工作：

  1. 作业 A 开始。作业 A 通过 Web API 获取数据并存储。
  2. 作业 A 运行。
  3. 作业 A 完成。
  4. 完成 A 后，作业 B 开始。作业 B 检查作业 A 存储的每条记录并添加一些额外的响应数据。
  5. 完成作业 B 后，用户会收到一封愉快的电子邮件，说明他们的报告已准备就绪。

到目前为止我的代码：

redis_conn = Redis()
use_connection(redis_conn)
q = Queue('normal', connection=redis_conn) # this is terrible, I know - fixing later
w = Worker(q)
job = q.enqueue(getlinksmod.lsGet, theURL,total,domainid)
w.work()

我认为我最好的解决方案是让 2 名工人，一名负责工作 A，一名负责 B。工作 B 工人可以监控工作 A，当工作 A 完成后，开始工作 B。

我无法想办法挽救我的生命是如何让一名工人监控另一名工人的状态。我可以使用 job.id 从作业 A 中获取作业 ID。我可以用 w.name 获取工人姓名。但是对于我如何将任何这些信息传递给其他工作人员，这还不是最模糊的。

或者，有没有一种我完全想念的更简单的方法来做到这一点？

4

4 回答 4

7

更新 2015 年 1 月，这个 pull request 现在被合并了，参数被重命名为depends_on，即：

second_job = q.enqueue(email_customer, depends_on=first_job)

对于运行旧版本等的人来说，原始帖子保持不变：

我已经提交了一个拉取请求（https://github.com/nvie/rq/pull/207）来处理 RQ 中的作业依赖关系。合并此拉取请求后，您将能够：

def generate_report():
    pass

def email_customer():
    pass

first_job = q.enqueue(generate_report)
second_job = q.enqueue(email_customer, after=first_job)
# In the second enqueue call, job is created,
# but only moved into queue after first_job finishes

现在，我建议编写一个包装函数来顺序运行您的作业。例如：

def generate_report():
     pass

def email_customer():
    pass

def generate_report_and_email():
    generate_report()
    email_customer() # You can also enqueue this function, if you really want to

# Somewhere else
q.enqueue(generate_report_and_email)
于 2013-06-18T16:14:21.673 回答
2

从文档的这个页面，rq看起来每个job对象都有一个result属性，可调用 by job.result，您可以检查它。如果工作尚未完成，它将是None，但如果您确保您的工作返回一些值（"Done"甚至job.result只是value，意味着第一个工人已经完成。

于 2012-10-08T07:12:36.543 回答
0

将作业 A 和作业 B 所做的事情组合在一个函数中，然后使用 eg multiprocessing.Pool（它的map_async方法）将其分摊到不同的进程中。

我不熟悉 rq，但multiprocessing它是标准库的一部分。默认情况下，它使用的进程数与 CPU 的内核数一样多，根据我的经验，这通常足以使机器饱和。

于 2012-08-23T22:58:08.373 回答
0

您可能对项目太深入而无法切换，但如果没有，请查看Twisted. http://twistedmatrix.com/trac/我现在正在将它用于一个访问 API、抓取 Web 内容等的项目。它并行运行多个作业，以及按顺序组织某些作业，所以作业 B 没有t 执行直到作业 A 完成。

如果您想尝试，这是学习 Twisted 的最佳教程。http://krondo.com/?page_id=1327

于 2012-08-24T13:29:48.177 回答