practical message queuing using rabbitmq (phpem, 3rd july 2014)
Post on 28-Aug-2014
226 Views
Preview:
DESCRIPTION
TRANSCRIPT
Practical Message Queueing Using
RabbitMQJames Titcumb
PHPem3rd July 2014
James Titcumbwww.jamestitcumb.comwww.protected.co.ukwww.phphants.co.uk@asgrim
Who is this guy?
Who are you?
https://www.flickr.com/photos/akrabat/10168019755/
What is message queueing?
Separation of Concerns
Scaling with Rabbit
RabbitMQApplication
Background processing
Scaling with Rabbit
RabbitMQApplication
Background processing
Background processing
Scaling with Rabbit
RabbitMQApplication
Background processing
Background processing
Background processing
Scaling with Rabbit
RabbitMQApplication
Background processing
Background processing
Background processing
Background processing
Scaling with Rabbit
RabbitMQApplication
Background processing
Background processing
Background processing
Background processing
Background processing
● Fast logging solution
Some real world uses...
● Fast logging solution● Background Processing
Some real world uses...
● Fast logging solution● Background Processing
○ Sending emails
Some real world uses...
● Fast logging solution● Background Processing
○ Sending emails○ Sending SMS
Some real world uses...
● Fast logging solution● Background Processing
○ Sending emails○ Sending SMS○ Analytics, reporting
Some real world uses...
(on precise64, other OSs may vary)
Installing RabbitMQ
● add apt repo○ deb http://www.rabbitmq.com/debian/ testing main
● add signing key○ http://www.rabbitmq.com/rabbitmq-signing-key-public.asc
● apt-get update● apt-get install rabbitmq-server● rabbitmq-plugins enable rabbitmq_management● sudo service rabbitmq-server restart
Using Apt
http://localhost:15672/
Basic Message Queuing
Objective: Basic Queuing
Producer Consumer
test_queue
1 2 3 4 5
composer.json{
"require": {
"videlalvaro/php-amqplib": "2.*"
}
}
then composer install
Please wait, connecting...use PhpAmqpLib\Connection\AMQPConnection;
$connection = new AMQPConnection(
'localhost',
5672,
'guest',
'guest',
'/'
);
$channel = $connection->channel();
basic/producer.phpuse PhpAmqpLib\Message\AMQPMessage;
$channel->queue_declare(
'test_queue',
false,
true,
false, false);
$message = new AMQPMessage('my test message');
$channel->basic_publish($message, '', 'test_queue');
basic/consumer.php$channel->basic_consume(
'test_queue', // Queue to consume
'', // Consumer identifier
false,
true, // No-ack means messages are "auto acknowledged"
false, // Exclusive - no other consumers can use the queue
false,
function(AMQPMessage $message) {
echo $message->body . "\n";
}
);
while (count($channel->callbacks)) {
$channel->wait();
}
What to expect...
Exchanges: Fanout
Objective: Fanout Exchange
test_exchange
amq.KfgPZ3PE
amq.cK5Cp3FC
Consumer
Consumer
Producer
1
1
2
2
3
3
4
4
5
5
fanout/producer.phpuse PhpAmqpLib\Message\AMQPMessage;
$channel->exchange_declare(
'test_exchange',
'fanout',
false, false, false);
$message = new AMQPMessage('my test message #' . $id);
$channel->basic_publish($message, 'test_exchange');
fanout/consumer.php$q = $channel->queue_declare(
'', // Lets RabbitMQ pick a name for queue
false, false, false,
true // Delete this queue
);
$queue_name = $q[0];
$channel->exchange_declare(
'test_exchange', 'fanout', false, false, false);
$channel->queue_bind($queue_name, 'test_exchange');
What to expect...
A word on Temporary Queues
● Only the “current” flow● Specific use cases
test_exchangeProducerMessages
go nowhere
Exchanges: Direct
Objective: Direct Exchange
test_direct
BK = apple
BK = banana, apple
Consumer
Consumer
Producer
3
Message Routing Keys1 = orange2 = banana3 = apple
2 3
BK = orange, banana, apple
Consumer1 2 3
direct/producer.php$channel->exchange_declare(
'test_direct', 'fanout', false, false, false);
$messageContent = 'my test message, key=' . $routingKey;
$message = new AMQPMessage($messageContent);
$channel->basic_publish($message, 'test_direct', $routingKey);
direct/consumer.php$q = $channel->queue_declare('', false, false, false, true);
$queue_name = $q[0];
$channel->exchange_declare(
'test_direct', 'direct', false, false, false);
// Bind for each routing key we want (BINDING KEY)
$channel->queue_bind($queue_name, 'test_direct', 'apple');
$channel->queue_bind($queue_name, 'test_direct', 'orange');
$channel->queue_bind($queue_name, 'test_direct', 'banana');
What to expect...
Exchanges: Topic
Objective:Topic Exchange
test_topic
BK = *.vegetable
BK = #
Consumer
Consumer
Producer
1 2
Message Routing Keys1 = red.vegetable2 = green.vegetable3 = green.fruit4 = red.meat5 = green.grass.long
1 2 3 4 5
BK = green.#
Consumer2 3 5
BK = *.grass.* / *.*.long
Consumer5
Real World Example
Fetch message
Logging Sequence
ApplicationBrowser Log Server
HTTP request
JSON via AMQP
Error!
HTTP response
RabbitMQ
Flexibility!
● Temporary Queues○ e.g. inspect “debug” messages
Flexibility!
● Temporary Queues● Queues to log to DB
Flexibility!
● Temporary Queues● Queues to log to DB● Queues to email “alert/emergency”
Flexibility!
● Temporary Queues● Queues to log to DB● Queues to email “alert/emergency”● Get creative with routing keys
○ RK = app.api.error … BK = #.api.error○ RK = app.form.debug … BK = #.debug
Problem: SPOF
Solution 1: Clustering
Clustering
RabbitMQNode 1
RabbitMQNode 3
RabbitMQNode 2
RabbitMQNode 4
RabbitMQNode 5
RabbitMQNode 6
Load Balance / Floating IP / Low TTL DNS etc.
Clustering
● Everything replicated, except queues
Clustering
● Everything replicated, except queues● Types:
○ RAM○ Disk
Clustering
● Everything replicated, except queues● Types:
○ RAM○ Disk
● Configuration:○ CLI (rabbitmqctl)○ Configuration files
Creating a clusternode1$ rabbitmqctl cluster_status
Cluster status of node rabbit@node1 ...
[{nodes,[{disc,[rabbit@node1]}]},{running_nodes,[rabbit@node1]}]
...done.
node2$ rabbitmqctl cluster_status
Cluster status of node rabbit@node2 ...
[{nodes,[{disc,[rabbit@node2]}]},{running_nodes,[rabbit@node2]}]
...done.
node3$ rabbitmqctl cluster_status
Cluster status of node rabbit@node3 ...
[{nodes,[{disc,[rabbit@node3]}]},{running_nodes,[rabbit@node3]}]
...done.
node2$ rabbitmqctl join_cluster --ram rabbit@node1
node3$ rabbitmqctl join_cluster rabbit@node2
node3$ rabbitmqctl cluster_status
Cluster status of node rabbit@node3 ...
[{nodes,[{disc,[rabbit@node3,rabbit@node1]},{ram,[rabbit@node2]}]},
{running_nodes,[rabbit@node2,rabbit@node1,rabbit@node3]}]
...done.
Creating a cluster
node1$ rabbitmqctl stop_app
node2$ rabbitmqctl forget_cluster_node rabbit@node1
node1$ rabbitmqctl reset
node1$ rabbitmqctl start_app
node2$ rabbitmqctl cluster_status
Cluster status of node rabbit@node2 ...
[{nodes,[{disc,[rabbit@node3]},{ram,[rabbit@node2]}]},
{running_nodes,[rabbit@node2,rabbit@node3]}]
...done.
Managing Nodes● Stopping/starting nodes● Removing nodes:
Solution 2: HA
HA + Queue Mirroring
RabbitMQNode 1
RabbitMQNode 2
Load Balance / Floating IP / Low TTL DNS etc.
https://github.com/asgrim/rmq-slides
Have a go yourself!
Questions?
James Titcumbwww.jamestitcumb.comwww.protected.co.ukwww.phphants.co.uk@asgrim
Thanks for watching!
top related