What Does RabbitMQ Do?<\/h2>\n
RabbitMQ provides a language-agnostic way for programs to send messages to each other. In simple terms, it allows a \u201cPublisher\/Producer\u201d to send a message and allows for a \u201cConsumer\u201d to listen for those messages.<\/p>\n
In one of its simpler models, it resembles what many Rails developers are used to with Sidekiq: the ability to distribute asynchronous tasks among one or more workers. Sidekiq is one of the first things I install on all my Rails projects. I don\u2019t think RabbitMQ would necessarily take its place, especially for things that work more easily within a Rails environment: sending emails, interacting with Rails models, etc.<\/p>\n
It doesn\u2019t stop there though. RabbitMQ can also handle Pub\/Sub functionality, where a single \u201cevent\u201d can be published and one or more consumers can subscribe to that event. You can take this further where consumers can subscribe only to specific events and\/or events that match the pattern they\u2019re watching for.<\/p>\n
Finally, RabbitMQ can allow for RPCs (Remote Procedure Calls), where you\u2019re looking for an answer right away from another program\u2026 basically calling a function that exists in another program.<\/p>\n
In this article, we\u2019ll be taking a look at both the \u201cTopic\u201d or pattern-based Pub\/Sub approach, as well as how an RPC can be accomplished.<\/p>\n
Event-based and asynchronous<\/h2>\n
The first example we\u2019ll be working with today is a sports news provider who receives incoming data about scores, goals, players, teams, etc. It has to parse the data, store it, and perform various tasks depending on the incoming data.<\/p>\n
To make things a little clearer, let\u2019s imagine that, in one of the incoming data streams, we\u2019ll be notified about soccer goals.<\/p>\n
When we discover that a goal has happened, there are a number of things that we need to do:<\/p>\n
- \n
- Parse and normalize the information<\/li>\n
- Store the details locally<\/li>\n
- Update the \u201cbox-score\u201d for the game that the goal took place in<\/li>\n
- Update the league leaderboard showing who the top goal scorer is<\/li>\n
- Notify all subscribers (push notification) of a particular league, team, or player<\/li>\n
- And any number of other tasks or analysis that we need to do<\/li>\n<\/ul>\n
Do we need to do all of those tasks in order? Should the program in charge of processing incoming data need to know about all of these tasks and how to accomplish them? I suggest that other than parsing\/normalizing the incoming data and maybe even saving it locally, the rest of the tasks can be done asynchronously and that the program shouldn\u2019t really know or care about all of these other tasks.<\/p>\n
What we can do is have the parser program emit an event (
soccer.mls.goal<\/code> for example), along with its accompanying information:<\/p>\n{\r\n league: 'MLS',\r\n team: 'Toronto FC',\r\n player: 'Sebastian Giovinco',\r\n opponent: 'New York City FC',\r\n time: '14:21'\r\n}<\/pre>\nThe parser can then forget about it! It\u2019s done its work of emitting the event. The rest of the work will be done by any number of consumers who have subscribed to this specific event.<\/p>\n
Producing in Ruby<\/h3>\n
To produce or emit events in Ruby, the first thing we need to do is install the bunny<\/a> client, which allows Ruby to communicate with RabbitMQ. For an example, here is some fake incoming data that needs to trigger the
goal<\/code> event for soccer.<\/p>\n# Imagine the parsing happens here :)\r\nsoccer = Soccer.new\r\nsoccer.emit_goal(\r\n league: 'MLS',\r\n team: 'Toronto FC',\r\n player: 'Sebastian Giovinco',\r\n opponent: 'New York City FC',\r\n time: '14:21'\r\n)<\/pre>\n
Let\u2019s next take a look at the
emit_goal<\/code> function inside of theSoccer<\/code> class, which builds the event slug and packages the data together to be included in the event being emitted:<\/p>\nclass Soccer\r\n include EventEmitter\r\n\r\n def emit_goal(raw_details)\r\n slug = \"soccer.#{raw_details[:league]}.goal\".downcase # \"soccer.mls.goal\"\r\n payload = raw_details.slice(:league, :team, :player, :opponent, :time)\r\n emit('live_events', slug, payload)\r\n end\r\nend<\/pre>\nThe
'live_events'<\/code> string has to do with which Exchange to publish the event to. An Exchange is basically like a router that decides which Queue(s) the event should be placed into. Theemit<\/code> method is inside of aModule<\/code> I created to simplify emitting events:<\/p>\nmodule EventEmitter\r\n def emit(topic, slug, payload)\r\n conn = Bunny.new\r\n conn.start\r\n\r\n ch = conn.create_channel\r\n x = ch.topic(topic)\r\n\r\n x.publish(payload.to_json, routing_key: slug)\r\n puts \" [OUT] #{slug}:#{payload}\"\r\n\r\n conn.close\r\n end\r\nend<\/pre>\nIt receives the
topic<\/code>, eventslug<\/code>, and eventpayload<\/code> and sends that information to RabbitMQ.<\/p>\nConsuming in Ruby<\/h3>\n
So far we have produced an event, but without a consumer to consume it, the event will be lost. Let\u2019s create a Ruby consumer that is listening for all soccer goal events.<\/p>\n
![\"screen-shot-2016-11-07-at-1-29-34-pm\"](\"http:\/\/www.webcodegeeks.com\/wp-content\/uploads\/2016\/11\/Screen-Shot-2016-11-07-at-1.29.34-PM.png\")
<\/a><\/p>\nYou may have noticed that what I was calling the event
slug<\/code> (or therouting_key<\/code>) looked like\"soccer.mls.goal\"<\/code>. Picking a pattern to follow is important, because consumers can choose which events to listen for based on a pattern such as\"soccer.*.goal\"<\/code>: all soccer goals regardless of the league.<\/p>\nThe consumer in this case will be some code which updates the leaderboard for the top goal scorers in the league. It is kicked off by running a Ruby file with this line:<\/p>\n
SoccerLeaderboard.new.live_updates<\/pre>\n
The
SoccerLeaderboard<\/code> class has a method calledlive_updates<\/code> which will call areceive<\/code> method provided be an includedModule<\/code>. It will provide the topic, the pattern of event slug\/routing_key to listen for, and a block of code to be called any time there is a new event to process.<\/p>\nclass SoccerLeaderboard\r\n include EventReceiver\r\n\r\n def live_updates\r\n receive('live_events', 'soccer.*.goal') do |payload|\r\n puts \"#{payload['player']} has scored a new goal.\"\r\n end\r\n end\r\nend<\/pre>\nThe
EventReceiver<\/code> Module is a little larger, but for the most part it\u2019s just setting up a connection to RabbitMQ and telling it what it wants to listen for.<\/p>\nmodule EventReceiver\r\n def receive(topic, pattern, \u2588)\r\n conn = Bunny.new\r\n conn.start\r\n\r\n ch = conn.create_channel\r\n x = ch.topic(topic)\r\n q = ch.queue(\"\", exclusive: true)\r\n q.bind(x, routing_key: pattern)\r\n\r\n puts \" [INFO] Waiting for events. To exit press CTRL+C\"\r\n\r\n begin\r\n q.subscribe(:block => true) do |delivery_info, properties, body|\r\n puts \" [IN] #{delivery_info.routing_key}:#{body}\"\r\n block.call(JSON.parse(body))\r\n end\r\n rescue Interrupt => _\r\n ch.close\r\n conn.close\r\n end\r\n end\r\nend<\/pre>\nConsuming in Elixir<\/h3>\n
I mentioned that RabbitMQ is language agnostic. What I mean by this is that we can not only have a consumer in Ruby listening for events, but we can have a consumer in Elixir listening for events at the same time.<\/p>\n
In Elixir, the package I used to connect to RabbitMQ was amqp<\/a>. One gotcha was that it relies on amqp_client<\/a> which was giving me problems with Erlang 19. To solve that, I had to link directly to the GitHub repository because it doesn\u2019t appear that the fix has been published to Hex yet.<\/p>\n
defp deps do\r\n [\r\n {:amqp_client, git: \"https:\/\/github.com\/dsrosario\/amqp_client.git\", branch: \"erlang_otp_19\", override: true},\r\n {:amqp, \"~> 0.1.5\"}\r\n ]\r\nend<\/pre>\nThe code to listen for events in Elixir looks like the following code below. Most of the code inside of the
start_listening<\/code> method is just creating a connection to RabbitMQ and telling it what to subscribe to. Thewait_for_messages<\/code> is where the event processing takes place.<\/p>\ndefmodule GoalNotifications do\r\n def start_listening do\r\n {:ok, connection} = AMQP.Connection.open\r\n {:ok, channel} = AMQP.Channel.open(connection)\r\n AMQP.Exchange.declare(channel, \"live_events\", :topic)\r\n {:ok, %{queue: queue_name}} = AMQP.Queue.declare(channel, \"\", exclusive: true)\r\n AMQP.Queue.bind(channel, queue_name, \"live_events\", routing_key: \"*.*.goal\")\r\n AMQP.Basic.consume(channel, queue_name, nil, no_ack: true)\r\n IO.puts \" [INFO] Waiting for messages. To exit press CTRL+C, CTRL+C\"\r\n wait_for_messages(channel)\r\n end\r\n\r\n def wait_for_messages(channel) do\r\n receive do\r\n {:basic_deliver, payload, meta} ->\r\n IO.puts \" [x] Received [#{meta.routing_key}] #{payload}\"\r\n\r\n wait_for_messages(channel)\r\n end\r\n end\r\nend\r\n\r\n\r\nGoalNotifications.start_listening<\/pre>\nRPC\u2026 when you need an answer right away<\/h2>\n
Remote Procedure Calls can be accomplished with RabbitMQ, but I\u2019ll be honest: It\u2019s more involved than the examples above for more of a typical Pub\/Sub approach. To me, it felt like each side (producer\/consumer) has to act as both a producer and a consumer.<\/p>\n
The flow is a little like this:<\/p>\n
- \n
- Program A asks Program B for some information, providing a unique ID for the request<\/li>\n
- Program A listens for responses that match the same unique ID<\/li>\n
- Program B receives request, does the work and provides a response with the same unique ID<\/li>\n
- Program A runs callback once matching unique ID is found in response from Program B<\/li>\n<\/ul>\n
In this example, we\u2019ll be talking about a product\u2019s inventory\u2026 an answer we need to know right away to be sure that there is stock available for a customer to purchase.<\/p>\n
inventory = ProductInventory.new('abc123').inventory\r\nputs \"Product has inventory of #{inventory}\"<\/pre>\nThe
ProductInventory<\/code> class is quite simple, mostly because I\u2019ve hidden the complexity of the RPC call inside of a class calledRemoteCall<\/code>.<\/p>\nclass ProductInventory\r\n attr_accessor :product_sku\r\n\r\n def initialize(product_sku)\r\n @product_sku = product_sku\r\n end\r\n\r\n def inventory\r\n RemoteCall.new('inventory').response(product_sku)\r\n end\r\nend<\/pre>\nNow let\u2019s take a look at how
RemoteCall<\/code> is handling it:<\/p>\nrequire 'bunny'\r\nrequire 'securerandom'\r\n\r\nclass RemoteCall\r\n attr_reader :lock, :condition\r\n attr_accessor :conn, :channel, :exchange, :reply_queue,\r\n :remote_response, :call_id, :queue_name\r\n\r\n def initialize(queue_name)\r\n @queue_name = queue_name\r\n @conn = Bunny.new\r\n @conn.start\r\n @channel = conn.create_channel\r\n @exchange = channel.default_exchange\r\n @reply_queue = channel.queue('', exclusive: true)\r\n end\r\n\r\n def response(payload)\r\n @lock = Mutex.new\r\n @condition = ConditionVariable.new\r\n response_callback(reply_queue)\r\n\r\n self.call_id = SecureRandom.uuid\r\n puts \"Awaiting call with correlation ID #{call_id}\"\r\n exchange.publish(payload,\r\n routing_key: queue_name,\r\n correlation_id: call_id,\r\n reply_to: reply_queue.name\r\n )\r\n lock.synchronize { condition.wait(lock) }\r\n\r\n remote_response\r\n end\r\n\r\n private\r\n\r\n def response_callback(reply_queue)\r\n that = self\r\n reply_queue.subscribe do |delivery_info, properties, payload|\r\n if properties[:correlation_id] == that.call_id\r\n that.remote_response = payload\r\n that.lock.synchronize { that.condition.signal }\r\n end\r\n end\r\n end\r\nend<\/pre>\nSo if all that code was for the Producer, what does the Consumer look like? It\u2019s kicked off with:<\/p>\n
server = InventoryServer.new\r\nserver.start<\/pre>\n
And the
InventoryServer<\/code> looks like:<\/p>\nrequire 'bunny'\r\n\r\nclass InventoryServer\r\n QUEUE_NAME = 'inventory'.freeze\r\n attr_reader :conn\r\n\r\n def initialize\r\n @conn = Bunny.new\r\n end\r\n\r\n def start\r\n conn.start\r\n channel = conn.create_channel\r\n queue = channel.queue(QUEUE_NAME)\r\n exchange = channel.default_exchange\r\n subscribe(queue, exchange)\r\n rescue Interrupt => _\r\n channel.close\r\n conn.close\r\n end\r\n\r\n def subscribe(queue, exchange)\r\n puts \"Listening for inventory calls\"\r\n\r\n queue.subscribe(block: true) do |delivery_info, properties, payload|\r\n puts \"Received call with correlation ID #{properties.correlation_id}\"\r\n product_sku = payload\r\n response = self.class.inventory(product_sku)\r\n exchange.publish(response.to_s,\r\n routing_key: properties.reply_to,\r\n correlation_id: properties.correlation_id\r\n )\r\n end\r\n end\r\n\r\n def self.inventory(product_sku)\r\n 42\r\n end\r\nend<\/pre>\nWow\u2026 that\u2019s a lot of work to make an RPC! RabbitMQ has a great guide explaining<\/a> how this works in a variety of different languages.<\/p>\n
Conclusion<\/h2>\n
Microservices don\u2019t always need to communicate synchronously, and they don\u2019t always need to communicate over HTTP\/JSON either. They can, but next time you\u2019re thinking about how they should speak to each other, why not consider doing it asynchronously using RabbitM? It comes with a great interface for monitoring the activity of the queue and has fantastic client support in a variety of popular languages. It\u2019s fast, reliable, and scalable.<\/p>\n
Microservices aren\u2019t free though\u2026 I think it\u2019s worthwhile considering whether the extra complexity involved in setting up separate services and providing them a way to communicate couldn\u2019t be better handled using something like Sidekiq and writing clean, modular code.<\/p>\n
\n