Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Messaging Patterns With RabbitMQ

Alvaro Videla
September 26, 2011

Messaging Patterns With RabbitMQ

Alvaro Videla

September 26, 2011
Tweet

More Decks by Alvaro Videla

Other Decks in Programming

Transcript

  1. Can we also notify the user friends when she uploads

    a new image? Tuesday, June 14, 2011
  2. Can we also notify the user friends when she uploads

    a new image? I forgot to mention we need it for tomorrow… Tuesday, June 14, 2011
  3. We need to give badges to users for each picture

    upload Tuesday, June 14, 2011
  4. We need to give badges to users for each picture

    upload and post uploads to Twitter Tuesday, June 14, 2011
  5. Dumb! You’re delivering full size images! The bandwidth bill has

    tripled! We need this fixed for yesterday! Tuesday, June 14, 2011
  6. I need to call your PHP stuff but from Python

    And also Java starting next week Tuesday, June 14, 2011
  7. I don’t want to wait till your app resizes my

    image! Tuesday, June 14, 2011
  8. %% image_controller handle('PUT', "/user/image", ReqData) -> {ok, Image} = image_handler:do_upload(ReqData:get_file()),

    resize_image(Image), notify_friends(ReqData:get_user()), ok. Third Implementation: Tuesday, June 14, 2011
  9. %% image_controller handle('PUT', "/user/image", ReqData) -> {ok, Image} = image_handler:do_upload(ReqData:get_file()),

    resize_image(Image), notify_friends(ReqData:get_user()), add_points_to_user(ReqData:get_user()), ok. Fourth Implementation: Tuesday, June 14, 2011
  10. %% image_controller handle('PUT', "/user/image", ReqData) -> {ok, Image} = image_handler:do_upload(ReqData:get_file()),

    resize_image(Image), notify_friends(ReqData:get_user()), add_points_to_user(ReqData:get_user()), tweet_new_image(User, Image), ok. Final Implementation: Tuesday, June 14, 2011
  11. What if • We need to speed up image conversion

    • User notification has to be sent by email Tuesday, June 14, 2011
  12. What if • We need to speed up image conversion

    • User notification has to be sent by email • Stop tweeting about new images Tuesday, June 14, 2011
  13. What if • We need to speed up image conversion

    • User notification has to be sent by email • Stop tweeting about new images • Resize in different formats Tuesday, June 14, 2011
  14. %% image_controller handle('PUT', "/user/image", ReqData) -> {ok, Image} = image_handler:do_upload(ReqData:get_file()),

    Msg = #msg{user = ReqData:get_user(), image = Image}, publish_message('new_image', Msg). First Implementation: Tuesday, June 14, 2011
  15. %% image_controller handle('PUT', "/user/image", ReqData) -> {ok, Image} = image_handler:do_upload(ReqData:get_file()),

    Msg = #msg{user = ReqData:get_user(), image = Image}, publish_message('new_image', Msg). First Implementation: %% friends notifier on('new_image', Msg) -> notify_friends(Msg.user, Msg.image). Tuesday, June 14, 2011
  16. %% image_controller handle('PUT', "/user/image", ReqData) -> {ok, Image} = image_handler:do_upload(ReqData:get_file()),

    Msg = #msg{user = ReqData:get_user(), image = Image}, publish_message('new_image', Msg). First Implementation: %% friends notifier on('new_image', Msg) -> notify_friends(Msg.user, Msg.image). %% points manager on('new_image', Msg) -> add_points(Msg.user, 'new_image'). Tuesday, June 14, 2011
  17. %% image_controller handle('PUT', "/user/image", ReqData) -> {ok, Image} = image_handler:do_upload(ReqData:get_file()),

    Msg = #msg{user = ReqData:get_user(), image = Image}, publish_message('new_image', Msg). First Implementation: %% friends notifier on('new_image', Msg) -> notify_friends(Msg.user, Msg.image). %% points manager on('new_image', Msg) -> add_points(Msg.user, 'new_image'). %% resizer on('new_image', Msg) -> resize_image(Msg.image). Tuesday, June 14, 2011
  18. Messaging • Share data across processes • Processes can be

    part of different apps Tuesday, June 14, 2011
  19. Messaging • Share data across processes • Processes can be

    part of different apps • Apps can live in different machines Tuesday, June 14, 2011
  20. Messaging • Share data across processes • Processes can be

    part of different apps • Apps can live in different machines • Communication is Asynchronous Tuesday, June 14, 2011
  21. Main Concepts • Messages are sent by Producers • Messages

    are delivered to Consumers Tuesday, June 14, 2011
  22. Main Concepts • Messages are sent by Producers • Messages

    are delivered to Consumers • Messages goes through a Channel Tuesday, June 14, 2011
  23. RabbitMQ • Enterprise Messaging System • Open Source MPL •

    Written in Erlang/OTP • Commercial Support • Messaging via AMQP Tuesday, June 14, 2011
  24. Features • Reliable and High Scalable • Easy To install

    • Easy To Cluster • Runs on: Windows, Solaris, Linux, OSX • AMQP 0.8 - 0.9.1 Tuesday, June 14, 2011
  25. Client Libraries • Java • .NET/C# • Erlang • Ruby,

    Python, PHP, Perl, AS3, Lisp, Scala, Clojure, Haskell Tuesday, June 14, 2011
  26. AMQP • Advanced Message Queuing Protocol • Suits Interoperability •

    Completely Open Protocol • Binary Protocol Tuesday, June 14, 2011
  27. AMQP Model • Exchanges • Message Queues • Bindings •

    Rules for binding them Tuesday, June 14, 2011
  28. Competing Consumers Create multiple Competing Consumers on a single channel

    so that the consumers can process multiple messages concurrently. Tuesday, June 14, 2011
  29. Publisher Code init(Exchange, Queue) -> #'exchange.declare'{exchange = Exchange, type =

    <<"direct">>, durable = true}, #'queue.declare'{queue = Queue, durable = false}, #'queue.bind'{queue = Queue, exchange = Exchange}. publish_msg(Exchange, Payload) -> Props = #'P_basic'{content_type = <<"application/json">>, delivery_mode = 2}, %% persistent publish(Exchange, #amqp_msg{props = Props, payload = Payload}). Tuesday, June 14, 2011
  30. Consumer Code init_consumer(Exchange, Queue) -> init(Exchange, Queue), #'basic.consume'{queue = Queue}.

    on(#'basic.deliver'{delivery_tag = DeliveryTag}, #amqp_msg{} = Msg) -> do_something_with_msg(Msg), #'basic.ack'{delivery_tag = DeliveryTag}. Tuesday, June 14, 2011
  31. Publish/Subscribe How can the sender broadcast an event to all

    interested receivers? Tuesday, June 14, 2011
  32. Publish/Subscribe Send the event on a Publish-Subscribe Channel, which delivers

    a copy of a particular event to each receiver. Tuesday, June 14, 2011
  33. Publisher Code init(Exchange, Queue) -> #'exchange.declare'{exchange = Exchange, type =

    <<"fanout">>, %% different type durable = true} %% same as before ... publish_msg(Exchange, Payload) -> Props = #'P_basic'{content_type = <<"application/json">>, delivery_mode = 2}, %% persistent publish(Exchange, #amqp_msg{props = Props, payload = Payload}). Tuesday, June 14, 2011
  34. Consumer Code A init_consumer(Exchange, ResizeImageQueue) -> init(Exchange, ResizeImageQueue), #'basic.consume'{queue =

    ResizeImageQueue}. on(#'basic.deliver'{delivery_tag = DeliveryTag}, #amqp_msg{} = Msg) -> resize_message(Msg), #'basic.ack'{delivery_tag = DeliveryTag}. Tuesday, June 14, 2011
  35. Consumer Code B init_consumer(Exchange, NotifyFriendsQueue) -> init(Exchange, NotifyFriendsQueue), #'basic.consume'{queue =

    NotifyFriendsQueue}. on(#'basic.deliver'{delivery_tag = DeliveryTag}, #amqp_msg{} = Msg) -> notify_friends(Msg), #'basic.ack'{delivery_tag = DeliveryTag}. Tuesday, June 14, 2011
  36. Consumer Code C init_consumer(Exchange, LogImageUpload) -> init(Exchange, LogImageUpload), #'basic.consume'{queue =

    LogImageUpload}. on(#'basic.deliver'{delivery_tag = DeliveryTag}, #amqp_msg{} = Msg) -> log_image_upload(Msg), #'basic.ack'{delivery_tag = DeliveryTag}. Tuesday, June 14, 2011
  37. Request/Reply When an application sends a message, how can it

    get a response from the receiver? Tuesday, June 14, 2011
  38. Return Address How does a replier know where to send

    the reply? Tuesday, June 14, 2011
  39. Return Address The request message should contain a Return Address

    that indicates where to send the reply message. Tuesday, June 14, 2011
  40. Correlation Identifier How does a requestor that has received a

    reply know which request this is the reply for? Tuesday, June 14, 2011
  41. Correlation Identifier Each reply message should contain a Correlation Identifier,

    a unique identifier that indicates which request message this reply is for. Tuesday, June 14, 2011
  42. RPC Client init() -> #'queue.declare_ok'{queue = SelfQueue} = #'queue.declare'{exclusive =

    true, auto_delete = true}, #'basic.consume'{queue = SelfQueue, no_ack = true}, SelfQueue. Tuesday, June 14, 2011
  43. RPC Client init() -> #'queue.declare_ok'{queue = SelfQueue} = #'queue.declare'{exclusive =

    true, auto_delete = true}, #'basic.consume'{queue = SelfQueue, no_ack = true}, SelfQueue. request(Payload, RequestId) -> Props = #'P_basic'{correlation_id = RequestId, reply_to = SelfQueue}, publish(ServerExchange, #amqp_msg{props = Props, payload = Payload}). Tuesday, June 14, 2011
  44. RPC Client init() -> #'queue.declare_ok'{queue = SelfQueue} = #'queue.declare'{exclusive =

    true, auto_delete = true}, #'basic.consume'{queue = SelfQueue, no_ack = true}, SelfQueue. request(Payload, RequestId) -> Props = #'P_basic'{correlation_id = RequestId, reply_to = SelfQueue}, publish(ServerExchange, #amqp_msg{props = Props, payload = Payload}). on(#'basic.deliver'{}, #amqp_msg{props = Props, payload = Payload}) -> CorrelationId = Props.correlation_id, do_something_with_reply(Payload). Tuesday, June 14, 2011
  45. RPC Server on(#'basic.deliver'{}, #amqp_msg{props = Props, payload = Payload}) ->

    CorrelationId = Props.correlation_id, ReplyTo = Props.reply_to, Reply = process_request(Payload), NewProps = #'P_basic'{correlation_id = CorrelationId}, publish("", %% anonymous exchange #amqp_msg{props = NewProps, payload = Reply}, ReplyTo). %% routing key Tuesday, June 14, 2011
  46. Control Bus How can we effectively administer a messaging system

    that is distributed across multiple platforms and a wide geographic area? Tuesday, June 14, 2011
  47. Control Bus Use a Control Bus to manage an enterprise

    integration system. Tuesday, June 14, 2011
  48. Control Bus • Send Configuration Messages • Start/Stop Services •

    Inject Test Messages • Collect Statistics Tuesday, June 14, 2011
  49. Detour How can you route a message through intermediate steps

    to perform validation, testing or debugging functions? Tuesday, June 14, 2011
  50. Detour Construct a Detour with a context-based router controlled via

    the Control Bus. In one state the router routes incoming messages through additional steps while in the other it routes messages directly to the destination channel. Tuesday, June 14, 2011
  51. Wire Tap How do you inspect messages that travel on

    a point-to-point channel? Tuesday, June 14, 2011
  52. Wire Tap Insert a simple Recipient List into the channel

    that publishes each incoming message to the main channel and a secondary channel. Tuesday, June 14, 2011
  53. Smart Proxy How can you track messages on a service

    that publishes reply messages to the Return Address specified by the requestor? Tuesday, June 14, 2011
  54. Smart Proxy Use a Smart Proxy to store the Return

    Address supplied by the original requestor and replace it with the address of the Smart Proxy. When the service sends the reply message route it to the original Return Address. Tuesday, June 14, 2011