Ruby Redis Pub/Sub Job Queue

We use Gearman at work so naturally I've been pondering simple alternatives. Obviously this code is not a 100% replacement for Gearman, but here's a (super) simple worker/job-queue that uses Redis to track job queues, and pub/sub to notify the workers of new tasks to complete.

The main guts of the worker are defined in a simple module. file: lib/worker.rb

require'redis'require'securerandom'require'json'moduleWorker# make module methods class methodsextendself# each worker gets a "unique" idWORKER_ID=SecureRandom.hex# module container for worker methodsmoduleWorkerMethodsend# method used to define worker tasksdefadd(task_name,*args,&block)raise"Block required"unlessblock_given?WorkerMethods.define_singleton_methodtask_name,blockenddefwork# process existing tasks in listwhileRedisWorker.got_tasks?RedisWorker.do_next_taskend# subscribe for new workRedisWorker.subscribeendclassRedisWorker# connect to redis for non-pub/sub commands@redis=Redis.newdefself.do_next_task(task=nil)# get task if not passed as argumenttask=task_counts.delete_if{|k,v|v==0}.keys.sampleiftask.nil?# pop task from redis listjson=@redis.lpoptaskreturnifjson.nil?# parse the taskdata=JSON.parsejson# debug outputputs"WORKER: #{WORKER_ID} - #{data}"WorkerMethods.sendtask,dataend# boolean if any tasks existdefself.got_tasks?total=task_counts.inject(0){|sum,n|sum+n[1]}returntotal>0?true:falseend# check redis list size for each worker taskdefself.task_countsWorkerMethods.singleton_methods.each_with_object({}){|task,hsh|hsh[task]=@redis.llentask}end# use redis pub/sub to subscribe to channel for new tasksdefself.subscribe@channel=defined?(CHANNEL)?CHANNEL:'job_server'@pubsub=Redis.new@pubsub.subscribe(@channel)do|on|on.messagedo|channel,msg|# messagedata=JSON.parse(msg)# process jobs this worker knows how to completeifWorkerMethods.respond_to?data['task']do_next_taskdata['task']endendendendendend

Workers can be created by including the worker module, and calling the add method with a block. file: worker_start.rb