rabbit

consumer.rb

require 'bunny'
require 'logger'
require 'pry'

def consume(num, conn)
  logger = Logger.new(STDERR)
  channel = conn.create_channel
  queue = channel.queue('task_queue', durable: true)
  channel.prefetch(1)
  logger.info("Thread ##{num} Waiting for messages...")
  prng = Random.new
  queue.subscribe(manual_ack: true, block: true) do |dinfo, _properties, body|
    logger.info("Thread ##{num} Received message '#{body}'!")
    sleep(prng.rand(1..10))
    channel.ack(dinfo.delivery_tag)
  end
end

def main
  sleep(3)
  conn = Bunny.new(hostname: 'rabbitmq')
  conn.start
  threads = (1..3).map do |num|
    Thread.new { consume(num, conn) }
  end
  threads.map(&:join)
rescue Interrupt => _
  conn.close
end

main if $PROGRAM_NAME == __FILE__

docker-compose.yml

rabbitmq:
  image: rabbitmq:3.5.5
  expose:
    - 5672

producer:
  dockerfile: Dockerfile::Producer
  build: .
  command: bundle exec ruby producer.rb
  volumes:
    - .:/producer
  links:
    - rabbitmq

consumer:
  dockerfile: Dockerfile::Consumer
  build: .
  command: bundle exec ruby consumer.rb
  volumes:
    - .:/consumer
  links:
    - rabbitmq

Dockerfile::Consumer

FROM ruby:2.2.3

ENV APP_HOME=/consumer
RUN mkdir $APP_HOME
WORKDIR $APP_HOME
COPY Gemfile* $APP_HOME/
RUN bundle install

Dockerfile::Producer

FROM ruby:2.2.3

ENV APP_HOME=/producer
RUN mkdir $APP_HOME
WORKDIR $APP_HOME
COPY Gemfile* $APP_HOME/
RUN bundle install

Gemfile

source 'https://rubygems.org'
ruby '2.2.3'

gem 'bunny'
gem 'pry'

Gemfile.lock

GEM
  remote: https://rubygems.org/
  specs:
    amq-protocol (2.0.0)
    bunny (2.2.0)
      amq-protocol (>= 2.0.0)
    coderay (1.1.0)
    method_source (0.8.2)
    pry (0.10.2)
      coderay (~> 1.1.0)
      method_source (~> 0.8.1)
      slop (~> 3.4)
    slop (3.6.0)

PLATFORMS
  ruby

DEPENDENCIES
  bunny
  pry

BUNDLED WITH
   1.10.6

producer.rb

require 'bunny'
require 'logger'
require 'pry'

def main
  sleep(3)
  logger = Logger.new(STDERR)

  conn = Bunny.new(hostname: 'rabbitmq')
  conn.start

  channel = conn.create_channel
  queue   = channel.queue('task_queue', durable: true)

  loop do
    msg = %w(one two three).sample
    logger.info("Publishing message '#{msg}'...")
    queue.publish(msg, persistent: true)
    sleep(3)
  end
rescue Interrupt => _
  conn.close
end

main if __FILE__ == $PROGRAM_NAME