resque-scheduler

resque-scheduler

Description

Resque-scheduler is an extension to Resque that adds support for queueing items in the future.

Job scheduling is supported in two different ways: Recurring (scheduled) and Delayed.

Scheduled jobs are like cron jobs, recurring on a regular basis. Delayed jobs are resque jobs that you want to run at some point in the future. The syntax is pretty explanatory:

Resque.enqueue_in(5.days, SendFollowupEmail, argument) # runs a job in 5 days, calling SendFollowupEmail.perform(argument)
# or
Resque.enqueue_at(5.days.from_now, SomeJob, argument) # runs a job at a specific time, calling SomeJob.perform(argument)

Documentation

This README covers what most people need to know. If you're looking for details on individual methods, you might want to try the rubydoc.

Installation

To install:

gem install resque-scheduler

If you use a Gemfile:

gem 'resque-scheduler'

Adding the resque:scheduler rake task:

require 'resque/scheduler/tasks'

Rake integration

By default, resque-scheduler depends on the "resque:setup" rake task. Since you probably already have this task, lets just put our configuration there. resque-scheduler pretty much needs to know everything resque needs to know.

…

The scheduler rake task is responsible for both queueing items from the schedule and polling the delayed queue for items ready to be pushed on to the work queues. For obvious reasons, this process never exits.

rake resque:scheduler

or, if you want to load the environment first:

rake environment resque:scheduler

Standalone Executable

The scheduler may also be run via a standalone resque-scheduler executable, which will be available once the gem is installed.

# Get some help
resque-scheduler --help

The executable accepts options via option flags as well as via environment variables.

Environment Variables

Both the Rake task and standalone executable support the following environment variables:

• APP_NAME - Application name used in procline ($0) (default empty)
• BACKGROUND - Run in the background if non-empty (via Process.daemon, if supported) (default false)
• DYNAMIC_SCHEDULE - Enables dynamic scheduling if non-empty (default false)
• RAILS_ENV - Environment to use in procline ($0) (default empty)
• INITIALIZER_PATH - Path to a Ruby file that will be loaded before requiring resque and resque/scheduler (default empty).
• RESQUE_SCHEDULER_INTERVAL - Interval in seconds for checking if a scheduled job must run (coerced with Kernel#Float()) (default 5)
• LOGFILE - Log file name (default empty, meaning $stdout)
• LOGFORMAT - Log output format to use (either 'text', 'json' or 'logfmt', default 'text')
• PIDFILE - If non-empty, write process PID to file (default empty)
• QUIET - Silence most output if non-empty (equivalent to a level of MonoLogger::FATAL, default false)
• VERBOSE - Maximize log verbosity if non-empty (equivalent to a level of MonoLogger::DEBUG, default false)

Resque Pool integration

For normal work with the resque-pool gem, add the following task to wherever tasks are kept, such as ./lib/tasks/resque.rake:

task 'resque:pool:setup' do
  Resque::Pool.after_prefork do |job|
    Resque.redis.reconnect
  end
end

Delayed jobs

Delayed jobs are one-off jobs that you want to be put into a queue at some point in the future. The classic example is sending email:

Resque.enqueue_in(
  5.days,
  SendFollowUpEmail,
  user_id: current_user.id
)

This will store the job for 5 days in the resque delayed queue at which time the scheduler process will pull it from the delayed queue and put it in the appropriate work queue for the given job and it will be processed as soon as a worker is available (just like any other resque job).

NOTE: The job does not fire exactly at the time supplied. Rather, once that time is in the past, the job moves from the delayed queue to the actual resque work queue and will be completed as workers are free to process it.

Also supported is Resque.enqueue_at which takes a timestamp to queue the job, and Resque.enqueue_at_with_queue which takes both a timestamp and a queue name:

Resque.enqueue_at_with_queue(
  'queue_name',
  5.days.from_now,
  SendFollowUpEmail,
  user_id: current_user.id
)

The delayed queue is stored in redis and is persisted in the same way the standard resque jobs are persisted (redis writing to disk). Delayed jobs differ from scheduled jobs in that if your scheduler process is down or workers are down when a particular job is supposed to be queue, they will simply "catch up" once they are started again. Jobs are guaranteed to run (provided they make it into the delayed queue) after their given queue_at time has passed.

One other thing to note is that insertion into the delayed queue is O(log(n)) since the jobs are stored in a redis sorted set (zset). I can't imagine this being an issue for someone since redis is stupidly fast even at log(n), but full disclosure is always best.

Removing Delayed Jobs

If you have the need to cancel a delayed job, you can do like so:

# after you've enqueued a job like:
Resque.enqueue_at(5.days.from_now, SendFollowUpEmail, :user_id => current_user.id)
# remove the job with exactly the same parameters:
Resque.remove_delayed(SendFollowUpEmail, :user_id => current_user.id)

If you need to cancel a delayed job based on some matching arguments, but don't wish to specify each argument from when the job was created, you can do like so:

# after you've enqueued a job like:
Resque.enqueue_at(5.days.from_now, SendFollowUpEmail, :account_id => current_account.id, :user_id => current_user.id)
# remove jobs matching just the account:
Resque.remove_delayed_selection { |args| args[0]['account_id'] == current_account.id }
# or remove jobs matching just the user:
Resque.remove_delayed_selection { |args| args[0]['user_id'] == current_user.id }

If you need to cancel a delayed job based on some matching arguments AND by which class the job is, but don't wish to specify each argument from when the job was created, you can do like so:

# after you've enqueued a job like:
Resque.enqueue_at(5.days.from_now, SendFollowUpEmail, :account_id => current_account.id, :user_id => current_user.id)
# remove jobs matching just the account and that were of the class SendFollowUpEmail:
Resque.remove_delayed_selection(SendFollowUpEmail) { |args| args[0]['account_id'] == current_account.id }
# or remove jobs matching just the user and that were of the class SendFollowUpEmail:
Resque.remove_delayed_selection(SendFollowUpEmail) { |args| args[0]['user_id'] == current_user.id }

If you need to enqueue immediately a delayed job based on some matching arguments, but don't wish to specify each argument from when the job was created, you can do like so:

# after you've enqueued a job like:
Resque.enqueue_at(5.days.from_now, SendFollowUpEmail, :account_id => current_account.id, :user_id => current_user.id)
# enqueue immediately jobs matching just the account:
Resque.enqueue_delayed_selection { |args| args[0]['account_id'] == current_account.id }
# or enqueue immediately jobs matching just the user:
Resque.enqueue_delayed_selection { |args| args[0]['user_id'] == current_user.id }

Updating Delayed Jobs

Previously delayed jobs may be delayed even further into the future like so:

# after you've enqueued a job like:
Resque.enqueue_at(1.minute.from_now, SendNotifications, :user_id => current_user.id)
# delay running the job until two minutes from now
Resque.delay_or_enqueue_at(2.minutes.from_now, SendNotifications, :user_id => current_user.id)

You don't need to worry if a matching job has already been queued, because if no matching jobs are found a new job is created and enqueued as if you had called enqueue_at. This means you don't need any special conditionals to know if a job has already been queued. You simply create the job like so:

Resque.delay_or_enqueue_at(1.minute.from_now, SendNotifications, :user_id => current_user.id)

If multiple matching jobs are found, all of the matching jobs will be updated to have the same timestamp even if their original timestamps were not the same.

# enqueue multiple jobs with different delay timestamps
Resque.enqueue_at(1.minute.from_now, SendNotifications, :user_id => current_user.id)
Resque.enqueue_at(2.minutes.from_now, SendNotifications, :user_id => current_user.id)

# delay running the two jobs until 5 minutes from now
Resque.delay_or_enqueue_at(5.minutes.from_now, SendNotifications, :user_id => current_user.id)

The most useful case for increasing the delay of an already delayed job is to batch together work based on multiple events. For example, if you wanted to send a notification email to a user when an event triggers but didn't want to send 10 emails if many events happened within a short period, you could use this technique to delay the noficication email until no events have triggered for a period of time. This way you could send 1 email containing the 10 notifications once no events have triggered for 2 minutes. You could implement this like so:

# Send a notification when an event is created.
# app/models/event.rb
after_commit on: :create do
  Resque.delay_or_enqueue_in(2.minutes, SendNotifications, :user_id => user.id)
end

When the first event is created a job will be scheduled to send unsent notifications to the associated user. If another event is created within the 2 minute window, the timer will be reset to 2 minutes. This will continue as long as new events are created for the specific user before the 2 minute timer expires. Once the timer expires and the job is scheduled any new events that are created will schedule a new job and start the process over. By adjusting the window you can tweak the trade-off between sending notification emails quickly after an event happens and sending fewer emails.

Read more in the original PR

Scheduled Jobs (Recurring Jobs)

Scheduled (or recurring) jobs are logically no different than a standard cron job. They are jobs that run based on a schedule which can be static or dynamic.

Static schedules

Static schedules are set when resque-scheduler starts by passing a schedule file to resque-scheduler initialization like this (see Installation above for a more complete example):

Resque.schedule = YAML.load_file('your_resque_schedule.yml')

If a static schedule is not set resque-scheduler will issue a "Schedule empty!" warning on startup, but despite that warning setting a static schedule is totally optional. It is possible to use only dynamic schedules (see below).

The schedule file is a list of Resque job classes with arguments and a schedule frequency (in crontab syntax). The schedule is just a hash, but is usually stored in a YAML like this:

…

If you would like to setup a job that is executed manually you can configure like this in your YAML file.

ImportOrdersManual:
  custom_job_class: 'AmazonMws::ImportOrdersJob'
  never: "* * * * *"
  queue: high
  description: "This is a manual job for importing orders."
  args:
    days_in_arrears: 7

The queue value is optional, but if left unspecified resque-scheduler will attempt to get the queue from the job class, which means it needs to be defined. If you're getting "uninitialized constant" errors, you probably need to either set the queue in the schedule or require your jobs in your "resque:se