项目作者: ltello

项目描述 :
Ruby consumer/producer classes to RabbitMQ made simple
高级语言: Ruby
项目地址: git://github.com/ltello/rabbitmq-actors.git
创建时间: 2017-05-02T09:41:53Z
项目社区:https://github.com/ltello/rabbitmq-actors

开源协议:

下载


Gem Version
Build Status
Code Climate
Issue Count
Coverage Status
Dependency Status

RabbitMQ::Actors

rabbitmq-actors is a simple and direct way to use RabbitMQ in your Ruby applications.
It uses the excellent bunny Ruby client to RabbitMQ and basically, it provides a set of Ruby classes
implementing the most common producer-consumer patterns for multiple applications to exchange messages
using a RabbitMQ message broker:

  • master/worker
  • publisher/subscriber
  • routing producers/consumers
  • topic producers/consumers and
  • headers producers/consumers

Installation

If you don’t have RabbitMQ installed, you can do it following the instructions in RabbitMQ website.
On Mac OSX, the fastest way is via Homebrew:

  1. $ brew install rabbitmq
  2. $ rabbitmq-server

Make sure, /usr/local/sbin is in your $PATH

Once rabbitmq is installed locally or remotely accessible, add this line to your application’s Gemfile:

  1. gem 'rabbitmq-actors'

and execute:

  1. $ bundle

or install it yourself as:

  1. $ gem install rabbitmq-actors

Usage

To use rabbitmq-actors in your application, the first thing you need to do is set the url where the RabbitMQ messaging broker is running:

  1. RabbitMQ::Server.url = 'amqp://localhost' # talk to local RabbitMQ server

Patterns

There are different ways you can use RabbitMQ depending on the domain of your application.
RabbitMQ allows you to implement several message exchange patterns between producers and consumers.
Use the one/s that fit most the needs of your application:

Master - Worker pattern

Under this strategy, one or several programs (masters) produce and publish messages to a known work queue.
On the other hand, consumer programs (workers) bind to that queue so that RabbitMQ distributes all the received
messages among the workers in a round robin manner. Therefore, every message is routed to only one of the workers.

Creating a master producer

This gem provides the RabbitMQ::Actors::MasterProducer class to help you implement master producers in your
application:

  1. master = RabbitMQ::Actors::MasterProducer.new(
  2. queue_name: 'transactions',
  3. auto_delete: false,
  4. reply_queue_name: 'confirmations',
  5. logger: Rails.logger)

In the example above, master is a new MasterProducer instance routing messages to the "purchases"
queue (which won’t be automatically deleted by RabbitMQ when there are no consumers listening to it: auto_delete: false).

The reply_queue_name param will add a reply_to: "confirmations" property to every message published by
this master instance so that the consumer receiving the message knows the queue where it has to publish
any response message.

The activity happening inside master (when publishing messages, closing connections…) will be logged to
Rails.logger (STDOUT by default) with :info severity level.

Publishing messages

To publish a message using our fresh new master instance, we just call publish instance method with some
mandatory arguments (message and :message_id):

  1. message = { stock: 'Apple', number: 1000 }.to_json
  2. master.publish(message, message_id: '1234837325', content_type: "application/json")
  • message is a string containing the body of the message to be published to RabbitMQ
  • :message_id is a keyword argument and contains any id our application uses to identify the message.

There are lots of optional params you can add to publish a message (see the
MasterProducer code documentation and Bunny::Exchange#publish code documentation):

  • persistent: Should the message be persisted to disk?. Default true.
  • mandatory: Should the message be returned if it cannot be routed to any queue?
  • timestamp: A timestamp associated with this message
  • expiration: Expiration time after which the message will be deleted
  • type: Message type, e.g. what type of event or command this message represents. Can be any string
  • reply_to: Queue name other apps should send the response to. Default to `:reply_queue_name if set at creation time.
  • content_type: Message content type (e.g. application/json)
  • content_encoding: Message content encoding (e.g. gzip)
  • correlation_id: Message correlated to this one, e.g. what request this message is a reply for
  • priority: Message priority, 0 to 9. Not used by RabbitMQ, only applications
  • user_id: Optional user ID. Verified by RabbitMQ against the actual connection username
  • app_id: Optional application ID

Closing the channel

Finally, close the channel when no more messages are going to be published by the master instance:

  1. master.close

or using a chained method call after publishing the last message:

  1. master.publish(message, message_id: '1234837325', content_type: "application/json").and_close

Actors (consumers and producers of all types) can share a connection to RabbitMQ broker
but each one connects via its own private channel. Although RabbitMQ can have thousands
of channels open simulataneously, it is a good practice to close it when an actor is not
being used anymore.

Defining a worker consumer

To define worker consumers of master produced messages, subclass RabbitMQ::Actors::Worker class and
define a private perform instance method to process a received message like this:

  1. class MyListener < RabbitMQ::Actors::Worker
  2. def initialize
  3. super(queue_name: 'transactions',
  4. manual_ack: true
  5. logger: Rails.logger,
  6. on_cancellation: ->{ ActiveRecord::Base.connection.close }
  7. end
  8. private
  9. def perform(**task)
  10. message, message_id = JSON.parse(task[:body]), task[:properties][:message_id]
  11. ...
  12. # your code to process the message
  13. end
  14. end

RabbitMQ::Actors::Worker class requires a mandatory keyword argument to initialize instances:

  • queue_name: string containing the name of the durable queue where to receive messages from.

Other optional params you can add to initialize a worker (see the Worker code documentation):

  • manual_ack: tells RabbitMQ to wait for a manual acknowledge from the worker before
    marking a message as “processed” and remove it from the queue. If true, the acknowledgement will be
    automatically sent by the worker after returning from perform method. Defaults to false.
  • logger: where to log worker activity with :info severity. Defaults to STDOUT if none provided.
  • on_cancellation: a Proc/Lambda object to be called right before the worker is terminated.

Running a worker consumer

Call #start! method on a worker instance to start listening and processing messages from the associated queue.

  1. MyListener.new.start!

Press ^c on a console to stop a worker or send a process termination signal.

Publish - Subscribe pattern

The idea behind this strategy is to broadcast messages to all the subscribed consumers (subscribers)
as opposed to only one consumer as it was the case in the Master - Worker pattern.

Creating a publisher

Instantiate the class RabbitMQ::Actors::Publisher to create publishers of messages under this scheme:

  1. publisher = RabbitMQ::Actors::Publisher.new(exchange_name: 'sports', logger: Rails.logger)

exchange_name: param is mandatory and contains the name of the RabbitMQ exchange where to publis the
messages.

As we know from the master producer, the activity happening inside publisher can be sent to a logger
instance (Rails.logger in the example) with :info severity level.

Again, like master producers, RabbitMQ::Actors::Publisher you can pass a reply_queue_name: keyword
param to create a new instance.

Publishing messages

The way for a publisher to publish messages is identical to that of master producers. See documentation
above.

  1. publisher.publish(message, message_id: '1232357390', content_type: "application/json")

Closing the channel

  1. publisher.close

or using the chained way:

  1. publisher.publish(message, message_id: '1234837390', content_type: "application/json").and_close

Defining a subscriber

To define your subscriber class, make it a subclass of RabbitMQ::Actors::Subscriber class and
define a private perform instance method to process a received message like this:

  1. class ScoresListener < RabbitMQ::Actors::Subscriber
  2. def initialize
  3. super(exchange_name: 'scores',
  4. logger: Rails.logger,
  5. on_cancellation: ->{ ActiveRecord::Base.connection.close })
  6. end
  7. private
  8. def perform(**task)
  9. match_data = JSON.parse(task[:body])
  10. process_match(match_data)
  11. end
  12. ...
  13. end

RabbitMQ::Actors::Publiser class requires the following mandatory keyword argument:

  • exchange_name: name of the exchange where to consume messages from.

You can also add logger: and on_cancellation: keyword params (see worker documentation above)

Running a subscriber

Like every consumer actor, call #start! method on an instance to start listening and processing messages.

  1. ScoresListener.new.start!

Press ^c on a console or send a process termination signal to stop it.

Routing pattern

The routing pattern is similar to publish/subscribe strategy but messages are not routed to all consumers but
only to those bound to the value a property of the message named routing_key.
Every consumer bound to the exchange states those routing_key values it is interested in. When a message
arrives it comes with a certain routing_key value, so the exchange routes it to only those consumers
interested in that particular value.

Creating a routing producer

Instantiate the class RabbitMQ::Actors::RoutingProducer to create publishers of messages under this scheme:

  1. routing_producer = RabbitMQ::Actors::RoutingProducer.new(
  2. exchange_name: 'sports',
  3. replay_queue_name: 'scores',
  4. logger: Rails.logger)

The description of the 3 params is similar to that of the previous producer classes. See above.

Publishing messages

  1. message = {
  2. championship: 'Wimbledon',
  3. match: {
  4. player_1: 'Rafa Nadal',
  5. player_2: 'Roger Federed',
  6. date: '01-Jul-2016'
  7. } }.to_json
  8. routing_producer.publish(message, message_id: '1234837633', content_type: "application/json", routing_key: 'tennis')

The way to publish messages is similar to that of the rest of producers. Note the mandatory param routing_key:

  • routing_key: send the message only to queues bound to this string value.

Closing the channel

  1. routing_producer.close

or using the chained way:

  1. routing_producer.publish(message, message_id: '1234837633', content_type: "application/json", routing_key: 'tennis').and_close

Defining a routing consumer

Use the class RabbitMQ::Actors::RoutingConsumer in a similar way as to the rest of consumer types:

  1. class TennisListener < RabbitMQ::Actors::RoutingConsumer
  2. def initialize
  3. super(exchange_name: 'sports',
  4. binding_keys: ['tennis'],
  5. logger: Rails.logger,
  6. on_cancellation: ->{ ActiveRecord::Base.connection.close })
  7. end
  8. private
  9. def perform(**task)
  10. match_data = JSON.parse(task[:body])
  11. process_tennis_match(match_data)
  12. end
  13. ...
  14. end
  15. class FootballListener < RabbitMQ::Actors::RoutingConsumer
  16. def initialize
  17. super(exchange_name: 'sports',
  18. binding_keys: ['football', 'soccer'],
  19. logger: Rails.logger,
  20. on_cancellation: ->{ ActiveRecord::Base.connection.close })
  21. end
  22. private
  23. def perform(**task)
  24. match_data = JSON.parse(task[:body])
  25. process_footbal_match(match_data)
  26. end
  27. ...
  28. end

RabbitMQ::Actors::RoutingConsumer class requires the following mandatory keyword arguments:

  • exchange_name: name of the exchange where to consume messages from.
  • binding_keys: a string or list of strings with the routing key values this consumer is interested in.

You can also add logger: and on_cancellation: keyword params (see worker documentation above)

Running a routing consumer

Like every consumer actor, call #start! method on an instance to start listening and processing messages.

  1. TennisListener.new.start!
  2. FootballListener.new.start!

Press ^c on a console or send a process termination signal to stop it.

Topics pattern

The topics pattern is very similar to the routing one. However routing keys are not free string values. Instead,
every routing key is a dot separated list of words.

Binding keys can use special chars to match one or several words:

* can substitute for exactly one word
# can substitute for zero or more words

Creating a topic producer

Instantiate the class RabbitMQ::Actors::TopicProducer to create publishers of messages under this scheme:

  1. topic_producer = RabbitMQ::Actors::TopicProducer.new(topic_name: 'weather', logger: Rails.logger)

where

  • topic_name: (mandatory) is the name of the exchange to send messages to.

reply_queue_name: and logger are the 2 optional params that can be added.

Publishing messages

  1. message = { temperature: 20, rain: 30%, wind: 'NorthEast' }.to_json
  2. topic_producer.publish(message, message_id: '1234837633', content_type: "application/json", routing_key: 'Europe.Spain.Madrid')

Note the special format of the mandatory routing_key: param

Closing the channel

  1. topic_producer.close

or using the chained way:

  1. topic_producer.publish(message, message_id: '1234837633', content_type: "application/json", routing_key: 'Europe.Spain.Madrid').and_close

Defining a topic consumer

Use the class RabbitMQ::Actors::TopicConsumer in a similar way as to the rest of consumer types:

  1. class SpainTennisListener < RabbitMQ::Actors::TopicConsumer
  2. def initialize
  3. super(topic_name: 'sports',
  4. binding_keys: '#.tennis.#.spain.#',
  5. logger: Rails.logger,
  6. on_cancellation: ->{ ActiveRecord::Base.connection.close })
  7. end
  8. private
  9. def perform(**task)
  10. match_data = JSON.parse(task[:body])
  11. process_tennis_match(match_data)
  12. end
  13. def process_tennis_match(data)
  14. ...
  15. end
  16. end
  17. class AmericaSoccerListener < RabbitMQ::Actors::TopicConsumer
  18. def initialize
  19. super(exchange_name: 'sports',
  20. binding_keys: '#.soccer.#.america.#',
  21. logger: Rails.logger,
  22. on_cancellation: ->{ ActiveRecord::Base.connection.close })
  23. end
  24. private
  25. def perform(**task)
  26. match_data = JSON.parse(task[:body])
  27. process_soccer_match(match_data)
  28. end
  29. def process_soccer_match(data)
  30. ...
  31. end
  32. end

RabbitMQ::Actors::TopicConsumer class requires the following mandatory keyword arguments:

  • topic_name: name of the exchange where to consume messages from.
  • binding_keys: a string or list of strings with the routing key matching patterns this consumer
    is interested in.

As always, you can also add logger: and on_cancellation: keyword params (see worker documentation above)

Running a topic consumer

Like every consumer actor, call #start! method on an instance to start listening and processing messages.

  1. SpainTennisListener.new.start!
  2. AmericaSoccerListener.new.start!

Press ^c on a console or send a process termination signal to stop it.

Headers pattern

The headers pattern is a strategy based on headers instead of routing keys to deliver messages
to consumers. Messages add a headers: property including pairs of key-value entries.
Consumers show interest in certain headers to get messages sent.

Creating a headers producer

Instantiate the class RabbitMQ::Actors::HeadersProducer to create publishers of messages under this scheme:

  1. headers_producer = RabbitMQ::Actors::HeadersProducer.new(headers_name: 'reports', logger: Rails.logger)

where

  • headers_name: (mandatory) is the name of the exchange to send messages to.

reply_queue_name: and logger are the 2 optional params that can be added.

Publishing messages

  1. message = 'A report about USA economy'
  2. headers_producer.publish(
  3. message,
  4. message_id: '1234837633',
  5. headers: { 'type' => :economy, 'area' => 'USA'})

where

  • headers: send the message only to consumers bound to this exchange and matching any/all
    of these header pairs.

As usual, message and message_id: are also mandatory params. See documentation above for
all the optional message params.

Closing the channel

  1. headers_producer.close

or using the chained way:

  1. headers_producer.publish(...).and_close

Defining a headers consumer

Use the class RabbitMQ::Actors::HeadersConsumer in a similar way as to the rest of consumer types:

  1. class NewYorkBranchListener < RabbitMQ::Actors::HeadersConsumer
  2. def initialize
  3. super(headers_name: 'reports',
  4. binding_headers: { 'type' => :econony, 'area' => 'USA', 'x-match' => 'any' },
  5. logger: Rails.logger,
  6. on_cancellation: ->{ ActiveRecord::Base.connection.close })
  7. end
  8. private
  9. def perform(**task)
  10. report_data = JSON.parse(task[:body])
  11. process_report(report_data)
  12. end
  13. def process_report(data)
  14. ...
  15. end
  16. end
  17. class LondonBranchListener < RabbitMQ::Actors::HeadersConsumer
  18. def initialize
  19. super(headers_name: 'reports',
  20. binding_headers: { 'type' => :industry, 'area' => 'Europe', 'x-match' =>'any' },
  21. logger: Rails.logger,
  22. on_cancellation: ->{ ActiveRecord::Base.connection.close })
  23. end
  24. private
  25. def perform(**task)
  26. report_data = JSON.parse(task[:body])
  27. process_report(report_data)
  28. end
  29. def process_report(data)
  30. ...
  31. end
  32. end

RabbitMQ::Actors::HedersConsumer class requires the following mandatory keyword arguments:

  • headers_name: name of the exchange where to consume messages from.
  • binding_headers: hash of headers this consumer is interested in.

Note the special mandatory binding header 'x-match'. Its value can be one of these:

  • 'any' receive the message if any of the message headers matches any of the binding headers.
  • 'all' receive the message only if all of the binding headers are included in the message headers.

Optional params logger: and on_cancellation:

Running a headers consumer

Like every consumer actor, call #start! method on an instance to start listening and processing messages.

  1. NewYorkBranchListener.new.start!
  2. LondonBranchListener.new.start!

Press ^c on a console or send a process termination signal to stop it.

Contributing

Bug reports and pull requests are welcome on GitHub at https://github.com/ltello/rabbitmq-actors