消息队列 delayed_job 的使用精解

1. 介绍

delayed_job是跟sidekiq一样的 ruby 编写的消息队列的系统，不过跟 sidekiq 一个很大的不同，在于 sidekiq 是使用 redis 来作为存储介质的，而 delayed_job 是使用 PostgreSQL，Mysql，Sqlite，或者 Mongodb 这样的数据库来作为存储系统的。sidekiq 的原理是不断地读取 (brpop) sidekiq 中的数据，delayed_job 也是一样，不过它读取的是表中的队列数据，所以是需要创建一张表来放队列的信息的。

2. 安装

添加下面一行到Gemfile文件中：

gem 'delayed_job_active_record'

创建存放队列的表：

$ rails generate delayed_job:active_record
$ rake db:migrate

输出如下：

create  bin/delayed_job
chmod  bin/delayed_job
create  db/migrate/20160103071229_create_delayed_jobs.rb

其中bin/delayed_job是用来启动 delayed_job 这个进程的。

而创建的表的名称为 delayed_jobs，我们来查看一下它的结构:

3. 使用

我们和 activejob 结合来使用 delayed_job。

在config/application.rb文件中设置queue_adapter。

config.active_job.queue_adapter = :delayed_job
config.eager_load_paths << Rails.root.join('app/jobs')

先来创建一个 job。

$ rails generate job update_article_visit_count

app/jobs/update_article_visit_count_job.rb的内容如下：

class UpdateArticleVisitCountJob < ActiveJob::Base
  queue_as :default

  def perform(article_id)
    # Do something later
    logger.info 'update article visit count begin'
    @article = Article.find(article_id)
    @article.visit_count += 1
    @article.save!(validate: false)
    logger.info 'update article visit count end'
  end

end

我们现在开启rails console来测试一下。

article = Article.first
UpdateArticleVisitCountJob.perform_later article.id

它产生的内容大体上是这样的：

现在我们准备开启 delayed_job 这个进程来消费这条队列，不过在开启之前，我们先把日志打开。

创建这个文件config/initializers/delayed_job_config.rb，内容如下：

Delayed::Worker.logger = Logger.new(File.join(Rails.root, 'log', 'delayed_job.log'))

然后我们监听日志的情况：

$ touch log/delayed_job.log
$ tail -f log/delayed_job.log

$ bin/delayed_job start

可能会报错，只要把下面这行添加到Gemfile文件中就好了:

gem 'daemons'

因为 delayed_job 是以后台的形式开启的，它是利用了 daemons 的功能来实现后台进程的。

成功执行的日志大体是这样的：

一成功，数据表 delayed_job 相关的数据也会被删除掉的。

如果执行失败，比如在 job 内有异常抛出或发生错误。

delayed_job 会不断地 retry，且数据库中的记录也是不会被删除掉的。

4. 高级选项

下面介绍 delayed_job 在使用上的高级功能。

4.1 delay 和 handle_asynchronously 方法

跟 sidekiq 一样，加上 delay 或 handle_asynchronously 方法就可以不用生成 job 或 worker，就可以让你原来的方法使用消息队列。

4.2 参数

每个 Job 分别可以指定三个参数：priority，run_at，queue。

这三个参数都是 delayed_jobs 表的列。

priority是指队列的优先级，run_at是指队列的运行时间，queue是指队列的名称，默认为default。

4.3 其他选项

比如指定最大的 retry 次数，队列最长的运行时间等。

Delayed::Worker.destroy_failed_jobs = false
Delayed::Worker.sleep_delay = 60
Delayed::Worker.max_attempts = 3
Delayed::Worker.max_run_time = 5.minutes
Delayed::Worker.read_ahead = 10
Delayed::Worker.default_queue_name = 'default'
Delayed::Worker.delay_jobs = !Rails.env.test?
Delayed::Worker.raise_signal_exceptions = :term

5. 相关的 gem

下面会简单介绍三个 gem：

• https://github.com/amitree/delayed_job_recurring
• https://github.com/ejschmitt/delayed_job_web
• https://github.com/thebestday/delayed-web

delayed_job_recurring 是一个能让 delayed_job 使用类似 cron 功能的 gem,也就是可以不结合 cron，轻易地直接利用数据库来实现任务计划的功能。

delayed_job_web 和 delayed-web 两个 gem 是相关于 delayed_job 的 web 端的任务查看监控页面。delayed_job_web 的功能多一些，delayed-web 简洁很多。

完结。