diff --git a/README.md b/README.md index 5e47359..339d763 100644 --- a/README.md +++ b/README.md @@ -401,6 +401,7 @@ By default, Backburner comes with the following workers built-in: | ------- | ------------------------------- | | `Backburner::Workers::Simple` | Single threaded, no forking worker. Simplest option. | | `Backburner::Workers::Forking` | Basic forking worker that manages crashes and memory bloat. | +| `Backburner::Workers::Threading` | Basic worker that can start a given number of threads per queue | | `Backburner::Workers::ThreadsOnFork` | Forking worker that utilizes threads for concurrent processing. | You can select the default worker for processing with: @@ -426,6 +427,11 @@ $ QUEUE=newsletter-sender,push-message THREADS=2 GARBAGE=1000 rake backburner:th For more information on the threads_on_fork worker, check out the [ThreadsOnFork Worker](https://github.com/nesquena/backburner/wiki/ThreadsOnFork-worker) documentation. Please note that the `ThreadsOnFork` worker does not work on Windows due to its lack of `fork`. +``` +$ QUEUE=newsletter-sender:2,push-message rake backburner:threading:work +``` + +Will create two threads listening to newsletter-sender queue and one thread listening to push-message queue. Additional workers such as individual `threaded` and `forking` strategies will hopefully be contributed in the future. If you are interested in helping out, please let us know. diff --git a/lib/backburner.rb b/lib/backburner.rb index b7a4acc..dbcdb55 100644 --- a/lib/backburner.rb +++ b/lib/backburner.rb @@ -13,6 +13,7 @@ require 'backburner/workers/simple' require 'backburner/workers/forking' require 'backburner/workers/threads_on_fork' +require 'backburner/workers/threading' require 'backburner/queue' module Backburner diff --git a/lib/backburner/tasks.rb b/lib/backburner/tasks.rb index 964fd0b..3517f43 100644 --- a/lib/backburner/tasks.rb +++ b/lib/backburner/tasks.rb @@ -27,6 +27,16 @@ end end # forking + namespace :threading do + # QUEUE=twitter:10,parse_page rake backburner:threads_on_fork:work + # twitter tube will have 10 threads, parse_page only 1 thread + desc "Starts backburner worker using threading processing" + task :work => :environment do + queues = (ENV["QUEUE"] ? ENV["QUEUE"].split(',') : nil) rescue nil + Backburner.work queues, :worker => Backburner::Workers::Threading + end + end # threading + namespace :threads_on_fork do # QUEUE=twitter:10:5000:5,parse_page,send_mail,verify_bithday THREADS=2 GARBAGE=1000 rake backburner:threads_on_fork:work # twitter tube will have 10 threads, garbage after 5k executions and retry 5 times. @@ -40,4 +50,4 @@ Backburner.work queues, :worker => Backburner::Workers::ThreadsOnFork end end # threads_on_fork -end \ No newline at end of file +end diff --git a/lib/backburner/workers/threading.rb b/lib/backburner/workers/threading.rb new file mode 100644 index 0000000..1e4a4df --- /dev/null +++ b/lib/backburner/workers/threading.rb @@ -0,0 +1,61 @@ +module Backburner + module Workers + class Threading < Worker + # Used to prepare job queues before processing jobs. + # Setup beanstalk tube_names and watch all specified tubes for jobs. + # + # @raise [Beaneater::NotConnected] If beanstalk fails to connect. + # @example + # @worker.prepare + # + def prepare + @queue_options = {} + self.tube_names.map! { |name| + # queue can be in format queue:number_of_threads + # number_of_threads is 1 by default + data = name.split(":") + tube_name = expand_tube_name(data.first) + @queue_options[tube_name] = {} + @queue_options[tube_name][:number_of_threads] = data[1].nil? || data[1].empty? ? 1 : data[1].to_i + tube_name + }.uniq! + log_info "[Threading worker] Working #{tube_names.size} queues: [ #{tube_names.join(', ')} ]" + end + + # Starts processing new jobs indefinitely. + # Primary way to consume and process jobs in specified tubes. + # + # @example + # @worker.start + # + def start + prepare + + threads = [] + self.tube_names.each do |tube_name| + @queue_options[tube_name][:number_of_threads].times do + threads << Thread.new do + log_info "[Threading worker] Watching queue : #{tube_name}" + Thread.current[:tube_name] = tube_name + Thread.current[:conn] = new_connection() + Thread.current[:conn].tubes.watch!(Thread.current[:tube_name]) + loop { + work_one_job(Thread.current[:conn]) + } + end + end + end + + threads.each { |thread| thread.join } + end + + def on_reconnect(conn) + unless Thread.current[:tube_name].nil? then + log_info "[Threading worker] Trying to recover connection for queue #{Thread.current[:tube_name]}" + Thread.current[:conn] = conn + Thread.current[:conn].tubes.watch!(Thread.current[:tube_name]) + end + end + end + end +end diff --git a/test/workers/threading_worker_test.rb b/test/workers/threading_worker_test.rb new file mode 100644 index 0000000..c271453 --- /dev/null +++ b/test/workers/threading_worker_test.rb @@ -0,0 +1,18 @@ +require File.expand_path('../../test_helper', __FILE__) +require File.expand_path('../../fixtures/test_forking_jobs', __FILE__) + +describe "Backburner::Workers::Threading module" do + + before do + Backburner.default_queues.clear + @worker_class = Backburner::Workers::Threading + end + + describe "for prepare method" do + it "should make tube names array always unique to avoid duplication" do + worker = @worker_class.new(["foo", "demo.test.foo"]) + worker.prepare + assert_equal ["demo.test.foo"], worker.tube_names + end + end # prepare +end