ProcessBalancer is a background job runner that is targeted toward the specific use-case of long running jobs.

If you need a job runner that runs small background jobs, look to Sidekiq.

ProcessBalancer has built-in functionality to balance your jobs across multiple instances.

Installation

Add this line to your application’s Gemfile:

gem 'process_balancer'

And then execute:

$ bundle

Or install it yourself as:

$ gem install process_balancer

Usage

Build a Job class that works through an iteration of your processing. The iteration does not have to be one record, it can be working through 1000 records. The iteration needs to be designed to lock its work atomically so that multiple concurrent workers could be running. The ProcessBalancer takes care of scaling out to run however many workers you want running across however many nodes you have running. Each instance of the Job will have a unique worker_id to ensure they do not trample on each other.

class ProcessQueue < ProcessBalancer::Base
  # set a worker locking algorithm
  lock_driver :simple_redis

  LOCK_SQL = "    WITH T as (\n      SELECT ctid\n      FROM queue_table\n      WHERE status = \#{QueueRecord::QUEUED} AND lock IS NULL\n      ORDER BY id\n      LIMIT 1000\n      FOR UPDATE SKIP LOCKED\n    )\n    UPDATE queue_records\n    SET lock = :lock, updated_at = :now\n    WHERE ctid = ANY(ARRAY(SELECT ctid FROM T))\n  SQL\n\n  def lock_records\n    # grab a # of records and lock them with the worker_id\n    sql = ActiveRecord::Base.sanitize_sql([LOCK_SQL, {lock: worker_index, now: Time.now}])\n    ActiveRecord::Base.connection.execute(sql)\n    # process those records\n    QueueRecord.where(lock: worker_index)\n  end\n\n  def process_record(entry)\n    # do processing\n    # mark record as processed and release the lock on that record\n    entry.update(lock: nil, status: QueueRecord::PROCESSED)\n  end\n\n  def unlock_records\n    # if any error occurs unlock any of our unprocessed records\n    QueueRecord.where(lock: worker_index).update_all(lock: nil)\n  end\nend\n"

Configuration file

jobs:
  process_queue:
    class: 'ProcessQueue'

Contributing

Bug reports and pull requests are welcome on GitHub at https://github.com/NetsoftHoldings/process_balancer.

License

The gem is available as open source under the terms of the LGPLv3 License.