bench.rb

require 'rubygems'
require 'bundler/setup'
Bundler.require(:default)
require 'benchmark'

queue = TorqueBox::Messaging::Queue.new('/queues/changes', host: 'localhost', :port => 5445)

puts "queue size: #{queue.count_messages}"
puts "queue processing started:"

# total: 205.970000  58.960000 264.930000
# total_time = Benchmark.measure do
#   100_000.times do |n|
#     queue.publish({offer_id: "a" * 32, changes: {shop_active: [true, false]}}, tx: false, persistent: false)
#   #   HardWorker.perform_async({offer_id: "a" * 32, changes: {shop_active: [true, false]}})
#     puts n if n % 1000 == 0
#   end
# end
# puts "total: #{total_time}"

# total:  40.290000  11.460000  51.750000 (382.962000)
# def process_queue(local_queue, session, queue)
#   local_queue.each do |message|
#     message = {offer_id: "a" * 32, changes: {shop_active: [true, false]}}
#     session.publish queue, message, queue.normalize_options(tx: false, persistent: false)
#   end
#   local_queue.clear
# end

# total:   9.200000   1.500000  10.700000 (  9.955000)
def process_queue(local_queue, session, queue)
  @producer ||= begin
    @jms_session = session.instance_variable_get('@jms_session')
    @jms_session.create_producer( session.java_destination( queue ) )
  end
  @options ||= queue.normalize_options(tx: false, persistent: false)
  @options[:properties] ||= {}
  local_queue.each do |message|
    payload = {offer_id: "a" * 32, changes: {shop_active: [true, false]}}
    # session.publish queue, message, queue.normalize_options(tx: false, persistent: false)
    message = TorqueBox::Messaging::Message.new( @jms_session, payload, @options[:encoding] )
    message.populate_message_headers(@options)
    message.populate_message_properties(@options[:properties])
    @producer.send( message.jms_message,
       @options.fetch(:delivery_mode, @producer.delivery_mode),
       @options.fetch(:priority, @producer.priority),
       @options.fetch(:ttl, @producer.time_to_live) )
    message
  end
  local_queue.clear
end

total_time = Benchmark.measure do
  queue.wait_for_destination(30) do
    queue.with_session() do |session|
      local_queue = []
      100_000.times do |n|
        local_queue << n
        if local_queue.size >= 1000
          process_queue(local_queue, session, queue)
        end
        puts n if n % 1000 == 0
      end
      process_queue(local_queue, session, queue)
    end
  end
end
puts "total: #{total_time}"

Gemfile

source 'http://rubygems.org'

ruby "1.9.3", :engine => "jruby", :engine_version => "1.7.19"

gem 'torquebox-messaging', '~> 3.1.2', platform: :jruby
gem 'torquebox', '~> 3.1.2', platform: :jruby

torquebox.rb

TorqueBox.configure do
  queue '/queues/changes'
end