在 Rails 5.x 中使用 DelayedJob 队列扩展 ActiveJob

Question

似乎无法从 activeJob 中访问 delay_job 对象。

到目前为止，我已经通过添加上下文作为额外参数（并在调用时将其删除）来解决这个问题，因此我可以添加我需要的任何额外信息（例如多租户）。这将在 handler 列中作为任何其他参数进行编码。

虽然我想找到重复项，但使用这种方法时会出现一些问题。所以我想定义一个额外的 job id j_id 列，用于查找重复项。

由于我无法访问delayed_job，我找不到填充它的方法。

这是我到目前为止所做的：

# already added column j_id to delayed_job table in migration file
class ApplicationJob < ActiveJob::Base
 before_enqueue :add_context

  def add_context
    # How can I set the j_id from here?
  end

Answer 1

Rails 的 ActiveJob 与 ActiveRecord 类似，因为它们都为后端（AR 的 DB 和 AJ 的队列后端）提供了稳定的 API。 DelayedJob 只是 AJ 的后端之一，就像 MySQL 或 Postgresql 是 AR 的后端一样。

如果您深入研究 DelayedJob 的代码（并且您使用 ActiveRecord 作为 DelayedJob 的后端，通过delayed_job_active_record gem），您会发现Delayed::Job 只是另一个ActiveRecord::Base 子类，就像您的任何模型一样。

这意味着您可以通过 Delayed::Job 类访问所有 DelayedJob 对象，以及您可以从 ActiveRecord 获得的所有查找和验证方法。

#<Delayed::Backend::ActiveRecord::Job:0x00007f00889c3210
 id: 1,
 priority: 0,
 attempts: 0,
 handler:
  "--- !ruby/object:ActiveJob::QueueAdapters::DelayedJobAdapter::JobWrapper\n" +
  "job_data:\n" +
  "  job_class: ApplicationJob\n" +
  "  job_id: 5f7539a3-120a-45b4-a6d7-12ec8c6cad3d\n" +
  "  provider_job_id: \n" +
  "  queue_name: default\n" +
  "  priority: \n" +
  "  arguments: []\n" +
  "  executions: 0\n" +
  "  exception_executions: {}\n" +
  "  locale: en\n" +
  "  timezone: UTC\n" +
  "  enqueued_at: '2020-03-02T05:38:22Z'\n",
 last_error: nil,
 run_at: Mon, 02 Mar 2020 05:38:22 UTC +00:00,
 locked_at: nil,
 failed_at: nil,
 locked_by: nil,
 queue: "default",
 created_at: Mon, 02 Mar 2020 05:38:22 UTC +00:00,
 updated_at: Mon, 02 Mar 2020 05:38:22 UTC +00:00>

这个类由一个名为 delayed_jobs 的表支持，正如您对任何 AR 类所期望的那样，因此您可以生成在该表上添加或更改列的迁移（添加 j_id）：

rails g migration AddJIdToDelayedJobs j:string

要添加验证，您可以创建 config/initializers/delayed_job.rb 和 monkey patch 类。

module Delayed
  class Job
     validates :j_id, uniqueness: true                                                                                                                                  
  end                                                                                                                                       
end

唯一剩下的就是在作业入队时向 j_id 列添加数据。通过这种方式，您可以添加使您的工作独一无二的任何信息，它可以是其他模型的 ID 或工作参数或您需要的任何内容。 DelayedJob 提供了一个很酷的plugin 解决方案，因此我们可以使用它来插入作业生命周期，并在作业入队时将信息添加到 j_id。回到config/initializers/delayed_job.rb，让我们添加新的插件类：

module Delayed
  class UniqueJobsPlugin < Plugin
    callbacks do |lifecycle|
      lifecycle.before(:enqueue) do |job|
        job.j_id = job #.anything inside the job, remember .handler is in YAML
      end
    end
  end
end
Delayed::Worker.plugins << Delayed::UniqueJobsPlugin

要获取 YAML 处理程序属性中的任何信息，您可以执行 handler_attributes = YAML.load(job.handler)

现在，如果 j_id 不是唯一的，则入队将失败，并出现常见的 AR 唯一性错误，您可以在将作业入队的控制器中处理该错误，或​​者将作业从入队的任何位置处理。