Ilya Grigorik recently wrote a nice summary about message handling with AMQP. Rany Keddo, the author of the Workling plugin, wrote another nice post about RabbitMQ with amqp. What is the most simple and basic example of using an AMQP message queue from Ruby? And what is an AMQP message queue anyway?

Message queues are useful to store a large number of items which can not be processed immediately and to achieve a loose coupling between application parts in general. There are two standard protocols for asynchronous message queues, AMQP and XMPP. AMQP is the lighter one which uses a binary format, while XMPP uses XML. AMQP means Advanced Message Queuing Protocol, it is an open standard application layer protocol for Message Oriented Middleware.

To use AMQP in Ruby, you can use either the carrot gem (a synchronous AMQP client without using EventMachine) or the amqp gem (a simple AMQP driver for Ruby using EventMachine). You can install them both at once with

$ sudo gem install amqp carrot

And you need to install a server implementing AMQP. RabbitMQ is an open source AMQP broker written in Erlang. To install the Erlang package and the RabbitMQ daemon under Ubuntu or Debian try

$ sudo apt-get install rabbitmq-server

Then you can start or stop the RabbitMQ daemon (the Linux counterpart of a service in Windows) with

$ sudo /etc/init.d/rabbitmq-server start
$ sudo /etc/init.d/rabbitmq-server stop

Here is a simple example for a publish/subscribe message exchange process using an AMQP message queue. The information flow in the publish/subscribe pattern is similar to the client/server pattern, the difference is that the server is sending (=publishing) and the client is listening (=subscribing), instead of the client-server pattern where the server is listening and the client is sending. Run ruby client.rb in one console and ruby server.rb in another:

client.rb (subscriber)

require 'rubygems'
require 'amqp'
require 'mq'

AMQP.start(:host => 'localhost' ) do
 q = MQ.new.queue('tasks')
 q.subscribe do |msg|
   puts msg
 end
end

server.rb (publisher)

require 'rubygems'
require 'amqp'
require 'mq'

AMQP.start(:host => 'localhost') do
 MQ.queue('tasks').publish("hello world")
 MQ.queue('tasks').publish("it is #{Time.now}")
 AMQP.stop { EM.stop }
end

It is also possible to use EM.run instead of AMQP.start. It is necessary to call AMQP.stop, otherwise you will be stuck in an endless loop. Another possibility is using the carrot gem:

client.rb (subscriber)

require 'rubygems'
require 'carrot'

q = Carrot.queue('tasks', :durable => true)

puts "count: #{q.message_count}"
while msg = q.pop(:ack => true)
 puts msg
 q.ack
end
Carrot.stop
end

server.rb (publisher)

require 'rubygems'
require 'carrot'
q = Carrot.queue('tasks')
q.publish("hello world")
q.publish("it is #{Time.now}")

development, ruby

, , , ,

2010/03/01

6 Comments