[email protected] rabbitmq©2014 allen i. holub ampq •open, wire-level binary protocol •very...

© 2014 Allen I. Holub www.holub.com

Allen I. Holub Holub Associates

www.holub.com [email protected] @allenholub

1

Allen I. Holub

www.holub.com [email protected] @allenholub

RabbitMQ

http://www.holub.com/slides


Producer

2

BrokerConsumer

PayloadMessage

Messaging


Why Messaging?

Easily accommodate new requirements. Painless (hot) scalability/updates Easy background/foreground processing Distributed development Polyglot applications (e.g. different languages) Promotes inter-module independence More efficient than DB for sharing info.

3

Speed developmentImproves flexibility

Get product out the door.


•Open Source written in Erlang (so needs Erlang VM)

•Polyglot java C#Erlang ruby python JavaScript/Node.js haskell python clojure perl COBOL! …

•Standard protocols, caninterface with other messaging systems.

4


AMPQ

•Open, wire-level binary protocol •Very efficient •Created by real users solving real problems • Interoperable between vendors •Highly scalable (phones to web clouds) •Runs over TCP/IP (or local)

5


Pub/Sub

6

Broker

Publisher Subscriber(s)

Topic


Producer

7

BrokerConsumer

Queues

pushpull


Producer

8

BrokerConsumer

Queues


Producer

9

Broker

Consumer

Or successive messages can round robin (load balancing)

Queues


Producer

10

Broker

Consumer


Queues


Producer

11

Broker

Consumer


Queues


Producer

12

Broker

Consumer


Queues


Queues

13

Bindings

Producers

ConsumersExchanges


(de)Coupling

• Logical –producer and consumer are unaware of each other

• Physical –producer and consumer are in different locations

• Temporal –message arrives at some future time

14

Could use a database, but messaging is better!


Queue/Exchange Properties

•Passive –can check to see if it exists

•Durable –continues to exist after broker restarts

• Exclusive –only one consumer can connect to it

•Auto-delete –deleted when there are no consumers waiting

15

© 2014 Allen I. Holub www.holub.com 16

Download git clone https://github.com/RabbitMQSimulator/ RabbitMQSimulator.git

Run minimal version as static HTML: cd RabbitMQSimulatoropen http://build/index.html

Run fully featured under node.js: brew install nodejs # or download from http://nodejs.org/ cd RabbitMQSimulatornpm install node app.js open http://localhost:3000


Direct (filter by routing key) & fanout (ignore routing key) routing

17

http://holub.com/RabbitMQSimulator/build


topic routing

18

This queueis interested only in this topic

* = any word # = 0 or more words


RPC call/response

19



REST-like

21

Topic Routing

Direct Routing

round robin}


Server Management Plugin

OverviewChannelsImport/ ExportQueues

Command- line tools also:

rabbitmqctl


Clustering

23

Multiple machines form one logical broker

Machines can mirror other machines.

SSL


Federated (Shoveled)

24

Exchange or queue on one broker

receives messages sent to exchange or queue on another broker.


Code

https://www.rabbitmq.com/getstarted.html


#!/usr/bin/env ruby# encoding: utf-8

require "bunny"

conn = Bunny.new(:automatically_recover => false)conn.start

ch = conn.create_channelq = ch.queue("hello")

ch.default_exchange.publish(“HelloWorld!”, :routing_key => q.name)puts " [x] Sent 'Hello World!'"

conn.close

26

Ruby



require "bunny"conn = Bunny.new(:automatically_recover => false)conn.start

ch = conn.create_channelq = ch.queue("hello")

begin puts " [*] Waiting for messages." q.subscribe(:block => true) do |delivery_info, properties, body| puts " [x] Received #{body}" endrescue Interrupt => _ conn.close

exit(0)end

27


Node.js (JavaScript)https://www.npmjs.org/package/rabbitmq-nodejs-client



class Send{ public static void Main() { var factory = new ConnectionFactory() { HostName = "localhost" }; using(var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { channel.QueueDeclare("hello", false, false, false, null); string message = "Hello World!"; var body = Encoding.UTF8.GetBytes(message);

channel.BasicPublish("", "hello", null, body); Console.WriteLine(" [x] Sent {0}", message); } } }}

30

C#


class Receive{ public static void Main() { var factory = new ConnectionFactory(){ HostName = "localhost" }; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { channel.QueueDeclare("hello", false, false, false, null);

var consumer = new QueueingBasicConsumer(channel); channel.BasicConsume("hello", true, consumer);

Console.WriteLine(" [*] Waiting for messages." + "To exit press CTRL+C"); while (true) { var ea= (BasicDeliverEventArgs)consumer.Queue.Dequeue(); var body = ea.Body; var message = Encoding.UTF8.GetString(body); Console.WriteLine(" [x] Received {0}", message); } } } }}

31


public class Send { private final static String QUEUE_NAME = "hello"; public static void main(String[] argv)

throws Exception { ConnectionFactory factory

= new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); String message = "Hello World!"; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println("[x] Sent '"+message + "'"); channel.close(); connection.close(); }}

32

Java


public class Recv { private final static String QUEUE_NAME = "hello"; public static void main(String[] argv) throws Exception{ ConnectionFactory factory = new ConnectionFactory() factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println(“ [*] Waiting for messages."); QueueingConsumer consumer = new QueueingConsumer(channel); channel.basicConsume(QUEUE_NAME, true, consumer); while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println("Received '" + message + "'"); } }}

33



require "bunny"

conn = Bunny.new(:automatically_recover => false)conn.start

ch = conn.create_channelx = ch.fanout("logs")

msg = ARGV.empty? ? "Hello World!" : ARGV.join(" ")

x.publish(msg)puts " [x] Sent #{msg}"

conn.close

34

Pub/Sub Ruby


#!/usr/bin/env ruby# encoding: utf-8require "bunny"conn = Bunny.new(:automatically_recover => false)conn.startch = conn.create_channelx = ch.fanout("logs")q = ch.queue("", :exclusive => true)q.bind(x)

puts " [*] Waiting for logs."begin q.subscribe(:block => true) do |delivery_info, properties, body| puts " [x] #{body}" endrescue Interrupt => _ ch.close conn.closeexit(0)end

35


Pub/Sub C#


Topics (C#)




RPC (C#)



DeliveryMode (persistent/transient) ContentType (mime type) ReplyTo (name of a callback queue) CorrelationId (correlate msg &


Allen Holub www.holub.com [email protected]

@allenholub

?49

[email protected] rabbitmq©2014 allen i. holub ampq •open, wire-level binary protocol •very...

Documents