ActiveJob + sidekiq做异步执行任务_qwbtc的博客

link之家

链接快照平台

输入网页链接，自动生成快照
标签化管理网页链接

初探ActiveJob

ActiveJob 是 Rails 4.2 新加入的功能。这个东西在beta阶段rubyChina就已经有很多高手关注了，无奈自己的项目使用的是4.1.5，升级到4.2 的时候其他gem又有很多依赖有问题，所以没在第一时间使用。今天补个课。

ActiveJob 是Rails自己开发运行后台程序的模块，常用于执行运行时间可能很长的工作（比如发送注册邮件）。

当然这种需求实际上非常普遍，所以rails 也有相应的第三方gem来解决这个需求，比如著名的Sidekiq和Resque等。**ActiveJob的出现不是为了代替他们**，而是统一了原来Resque、Sidekiq 等其他gem对后台运行程序的各种千奇百怪的写法。

ActiveJob 的使用官方文档已经给出了示例
首先在命令行中使用 rails g job JOBNAME 来新建一个任务

➜  my_rails42 git:(master) ✗ rails g job add_lots_of_users
      invoke  test_unit
      create    test/jobs/add_lots_of_users_job_test.rb
      create  app/jobs/add_lots_of_users_job.rb
➜  my_rails42 git:(master) ✗
 我们打开add_lots_of_users_job.rb，可以看到下列内容 
class AddLotsOfUsersJob < ActiveJob::Base
  queue_as :default
  def perform(*args)
    # Do something later
 我们可以将耗时的内容写入perform里 
  def perform(*args)
    # Do something later
    sleep 10
    1000.times do |index|
      user = User.new
      user.name = "atpking#{index}"
      user.save
 我们可以看到，我在job 里建立了一个作业，先睡10秒，再插入1000条数据到数据库中 
 至此，我们就成功建立了一个job了，接下来，就是使用了 
 官方demo讲的非常简单，就是在你使用的地方用这个句子：XXJob.perform_later PARAMS 
 比如我这里 
class UsersController < ApplicationController
  before_action :set_user, only: [:show, :edit, :update, :destroy]
  # GET /users
  # GET /users.json
  def index
    AddLotsOfUsersJob.perform_later
    @users = User.all
 我们运行一下试试 
 运行过程中发现**不像我们预想的那样**，而是访问index 的时候，活生生的等待了10多秒,在获得@users 
[ActiveJob] [AddLotsOfUsersJob] [e191c62d-68a9-425f-8a94-b9fe080c141c]    (0.8ms)  commit transaction
[ActiveJob] [AddLotsOfUsersJob] [e191c62d-68a9-425f-8a94-b9fe080c141c]    (0.0ms)  begin transaction
[ActiveJob] [AddLotsOfUsersJob] [e191c62d-68a9-425f-8a94-b9fe080c141c]   SQL (0.2ms)  INSERT INTO "users" ("name", "created_at", "updated_at") VALUES (?, ?, ?)  [["name", "atpking999"], ["created_at", "2015-04-20 04:30:29.320703"], ["updated_at", "2015-04-20 04:30:29.320703"]]
[ActiveJob] [AddLotsOfUsersJob] [e191c62d-68a9-425f-8a94-b9fe080c141c]    (0.6ms)  commit transaction
[ActiveJob] [AddLotsOfUsersJob] [e191c62d-68a9-425f-8a94-b9fe080c141c] Performed AddLotsOfUsersJob from Inline(default) in 12789.38ms
  User Load (2.1ms)  SELECT "users".* FROM "users"
  Rendered users/index.html.erb within layouts/application (186.7ms)
Completed 200 OK in 13210ms (Views: 392.6ms | ActiveRecord: 1166.0ms)
 注意看上面，很多歌ActiveJob 完毕了之后，输出Performed AddLotsOfUsersJob from Inline(default) in 12789.38ms,再执行的select * from users 
 也就是说，默认情况下的ActiveJob 跟我们使用的方法没什么区别，是阻塞的,实际上官方文档也说明了 
  4 Job Execution
 If no adapter is set, the job is immediately executedIf no adapter is set, the job is immediately executed. 
 那么我们需要给ActiveJob 指定一个Adapter 了。官方有支持以下的adapter，功能有所不同，需要注意。如果没设定，则是默认的 Active Job Inline ，可以看到一个悲剧，**不支持异步（Async）**，这也是为何我们刚刚等了很长时间 
 function  Async  Queues  Delayed  Priorities  Timeout  Retries
 Backburner  Yes  Yes  Yes  Yes  Job  Global
 Delayed Job  Yes  Yes  Yes  Job  Global  Global
 Qu  Yes  Yes  No  No  No  Global
 Que  Yes  Yes  Yes  Job  No  Job
 queue_classic  Yes  Yes  No*  No  No  No
 Resque  Yes  Yes  Yes (Gem)  Queue  Global  Yes
 Sidekiq  Yes  Yes  Yes  Queue  No  Job
 Sneakers  Yes  Yes  No  Queue  Queue  No
 Sucker Punch  Yes  Yes  No  No  No  No
 Active Job Inline(默认的)  No  Yes  N/A  N/A  N/A  N/A
 Active Job  Yes  Yes  Yes  No  No  No 
 我选用了sidekiq 作为Adapter，注意这里，你必须要安装过sidekiq, 没安装自然需要你Gemfile 里加一句咯 同时还要安装redis，之后在命令行使用 redis-server & 启用redis 同时也得在命令行里启用sidekiq,直接输入sidekiq & 即可 
 之后我们需要指定sidekiq 为我们的adapter，我们需要在application.rb 里加入一句话 
 config.active_job.queue_adapter = :sidekiq 
 再次运行index 页面，我们看rails 的日志，就变成了先执行Select * from Users，返回页面结果，再继续执行jobs的内容了。 
Started GET "/users" for ::1 at 2015-04-20 12:16:20 +0800
Processing by UsersController#index as HTML
[ActiveJob] Enqueued AddLotsOfUsersJob (Job ID: 50372234-6cf7-4ad2-b886-fd3029f4ea3d) to Sidekiq(default)
2015-04-20T04:16:20Z 79432 TID-ouzlum9ig ActiveJob::QueueAdapters::SidekiqAdapter::JobWrapper JID-eb96be6e314f6926583a1267 INFO: start
  User Load (0.3ms)  SELECT "users".* FROM "users"
  Rendered users/index.html.erb within layouts/application (4.1ms)
Completed 200 OK in 72ms (Views: 57.8ms | ActiveRecord: 0.5ms)
Started GET "/assets/jquery_ujs.self-8e98a7a072a6cee1372d19fff9ff3e6aa1e39a37d89d6f06861637d061113ee7.js?body=1" for ::1 at 2015-04-20 12:16:21 +0800
Started GET "/assets/users.self-e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855.css?body=1" for ::1 at 2015-04-20 12:16:21 +0800
Started GET "/assets/jquery.self-d03a5518f45df77341bdbe6201ba3bfa547ebba8ed64f0ea56bfa5f96ea7c074.js?body=1" for ::1 at 2015-04-20 12:16:21 +0800
Started GET "/assets/application.self-e80e8f2318043e8af94dddc2adad5a4f09739a8ebb323b3ab31cd71d45fd9113.css?body=1" for ::1 at 2015-04-20 12:16:21 +0800
Started GET "/assets/scaffolds.self-a98ac27100e3e5ca7065dbd7c898e5afa02690ec2ef84ccc02f65c4c20057b83.css?body=1" for ::1 at 2015-04-20 12:16:21 +0800
Started GET "/assets/turbolinks.self-c37727e9bd6b2735da5c311aa83fead54ed0be6cc8bd9a65309e9c5abe2cbfff.js?body=1" for ::1 at 2015-04-20 12:16:21 +0800
Started GET "/assets/users.self-877aef30ae1b040ab8a3aba4e3e309a11d7f2612f44dde450b5c157aa5f95c05.js?body=1" for ::1 at 2015-04-20 12:16:21 +0800
Started GET "/assets/application.self-3a3c8b61bda630ee689740ce7cbd0dd8ea6fdd45e2c42eef4661ab38cf268afe.js?body=1" for ::1 at 2015-04-20 12:16:21 +0800
2015-04-20T04:16:33Z 79432 TID-ouzlum9ig ActiveJob::QueueAdapters::SidekiqAdapter::JobWrapper JID-eb96be6e314f6926583a1267 INFO: done: 12.828 sec
 请注意上面的这两句话 
2015-04-20T04:16:20Z 79432 TID-ouzlum9ig ActiveJob::QueueAdapters::SidekiqAdapter::JobWrapper JID-eb96be6e314f6926583a1267 INFO: start
2015-04-20T04:16:33Z 79432 TID-ouzlum9ig ActiveJob::QueueAdapters::SidekiqAdapter::JobWrapper JID-eb96be6e314f6926583a1267 INFO: done: 12.828 sec
   Note that more advanced Sidekiq features cannot be controlled or configured via ActiveJob, e.g. saving backtraces.
  简单看了下 ActiveJob 里边的代码，发现 ActiveJobAdapter 里边都是主要只提供 .enqueue 这个接口，用于 ActiveJob 将任务塞进对应的队列，所以基本上断定 ActiveJob 并没有封装清理队列的相关逻辑，需要的直接用对应的 gem 提供的接口操作，比如这里提到的如何移除的方式：https://github.com/mperham/sidekiq/wiki/API#scheduled
                    初探ActiveJob简介ActiveJob 是 Rails 4.2 新加入的功能。这个东西在beta阶段rubyChina就已经有很多高手关注了，无奈自己的项目使用的是4.1.5，升级到4.2 的时候其他gem又有很多依赖有问题，所以没在第一时间使用。今天补个课。ActiveJob 是Rails自己开发运行后台程序的模块，常用于执行运行时间可能很长的工作（比如发送注册邮件
				ActiveJob :: Retriable
 自动重试失败的ActiveJob，并采用指数补偿。
 该gem旨在模仿Sidekiq的RetryJobs中间件的大多数功能，但可在ActiveJob层上运行。
要安装gem，请将以下内容添加到您的Gemfile中：
 gem "activejob-retriable"
 如果您使用的是Rails 5，请使用Github上的分支母版获得支持。 一旦Rails 5的第一个稳定版本发布，我们将推出5.0版本的gem。
 gem "activejob-retriable" , github : "SimplyBuilt/activejob-retr
 在指定的时间（使用cron的符号运行的线程一起Sidekiq工人调度作业* * * * *通过解析，更多。
 每隔30秒检查一次新作业，并在运行多个Sidekiq作业器时不多次计划同一作业。
 仅在至少运行一个Sidekiq进程时才添加调度作业，但是在运行多个Sidekiq进程或节点的环境中使用Sidekiq-Cron是安全的。
 如果您想了解调度的工作原理，请查看
与ActiveJob配合使用（Rails 4.2+）
 您不需要Sidekiq PRO，您可以将此宝石与便装的Sidekiq一起使用。
从<0> = 3.5结合使用会导致作业创建失败。 Sidekiq-cron 1.0包含一个补丁，可从rufus-scheduler切换到rufus-scheduler的核心依赖项fugit。
需要Redis 2.8或更高版本。 （建议大规模使用Redis 3.0.3或
				Lambdakiq
 使用 gem在AWS Lambda中运行Rails时， 替代产品。
 Lambdakiq允许您最大程度地利用AWS的托管基础架构。 管理吊舱和漫长的轮询过程的日子已经一去不复返了。 相反，AWS直接将消息传递到Rails的工作职能部门，并根据需要对其进行放大和缩小。 可观察性是使用AWS CloudWatch指标，仪表板和警报内置的。 了解有关更多信息或立即开始。
独特的Web和Jobs Lambda功能。
 AWS完全托管的轮询。 事件驱动。
 最多12次重试。 每个作业都是可配置的。
 镜像Sidekiq的重试时间。
 最后重试时间为11小时30分钟。
 支持ActiveJob的等待/延迟。 最多15分钟。
 无效消息最多可存储14天。
该gem假定您的Rails应用程序在AWS Lambda上，最好与我们的 gem一起使用。 它可能使用了
				Rails中，ActiveJob可以用于实现类似于前面提到的分数计算这类的，添加需要周期性执行的程序。 
可以使用rails generate job xxx生成job并进行Job的编写。 
调用方式为xxx.perform_now或xxx.perform_later，later大概是会在空闲的时候自动进行执行。 
可以设定何时进行perform操作。由于我这边的需求是每天半夜都需要进行操作，所以...
如果没有设置连接器，任务会立即执行。 
新建一个类继承ActiveJob::Base并定义一个perform方法 
class FileGeneratorJob <. ActiveJob::Base 
    def perform(*arg) 
        #do something 
end之后可以使用FileGeneratorJob.perfo
# 此方法会重新设置redis的key，将会会进入到失败队列
# “cancelled-”是你的sidekiq中redis的命令空间 例如 sidekiq:status:  => 例如 Sidekiq.redis {|c| c.set("sidekiq:status:#{jid}", "123456") }
Sidekiq.redis {|c| c.set("
				例子：如 我需要额外的时间去批量注入数据库，但是我们不想用户浏览我们的时候，去同步这个操作，因为要很久等待时间+服务器压力，我们想这个操作浏览网页同时，异步去执行这个操作。
1.   创建队列
rails g job guests_cleanup
打开app/jobs下能看到刚创建的作业文件
里边的方法有 perform(*guests)，在这个方法里边写一些您要执行的脚本
2.   入
本篇介绍sidekiq基于redis队列的后端的异步任务处理，当然还有定时任务，延时处理等功能。
https://github.com/mperham/sidekiq/wiki
http://mperham.github.com/sidekiq/
sidekiq的宣传语：
What if 1 Sidekiq process could do the work of 20 Resqu...

function	Async	Queues	Delayed	Priorities	Timeout	Retries
Backburner	Yes	Yes	Yes	Yes	Job	Global
Delayed Job	Yes	Yes	Yes	Job	Global	Global
Qu	Yes	Yes	No	No	No	Global
Que	Yes	Yes	Yes	Job	No	Job
queue_classic	Yes	Yes	No*	No	No	No
Resque	Yes	Yes	Yes (Gem)	Queue	Global	Yes
Sidekiq	Yes	Yes	Yes	Queue	No	Job
Sneakers	Yes	Yes	No	Queue	Queue	No
Sucker Punch	Yes	Yes	No	No	No	No
Active Job Inline(默认的)	No	Yes	N/A	N/A	N/A	N/A
Active Job	Yes	Yes	Yes	No	No	No