RabbitMQ Asynchronous RPC with

Transcript

1. RabbitMQ Asynchronous RPC with
2. World without queues
3. Execution time Confirm payment Assign paid feature
4. Execution time Confirm payment Add to stats Assign paid feature
5. Execution time Create invoice Confirm payment Add to stats Assign paid feature Some users leave page
6. Execution time Create invoice Confirm payment Refresh user wallet Add to stats Assign paid feature Some users leave page
7. Execution time Send confirmation email Create invoice Confirm payment Refresh user wallet Add to stats Assign paid feature Some users leave page Many users leave page
8. Execution time Send confirmation email Send invoice via email Send printed invoice Create invoice Confirm payment Refresh user wallet Add to stats Assign paid feature Some users leave page Many users leave page
9. Execution time Send confirmation email Send invoice via email Send printed invoice Create invoice Confirm payment Refresh user wallet Add to stats Assign paid feature Some users leave page Many users leave page
10. Problems Multiple page refresh Data consistency problems Increase system load Require support for broken/retried processes Whole atomic transaction is very heavy, and when part of it fail, everything need to be retried After outage, all data is gone.
12. Execution time Send confirmation email Send invoice via email Send printed invoice Create invoice Confirm payment Refresh user wallet Add to stats Assign paid feature queue queue queue queue
14. Producer Producing task Consumer MySql Database queue (MySql table) Producer write rows to mysql table with tasks definition Task Task Task id:1 id:2 id:3 Task id:4
15. Producer Consuming task Consumer MySql Database queue (MySql table) Task Task Task id:2 id:3 id:4 Task id:1 Consumer executes select query to retrieve tasks to process
16. Database queue (MySql table) Consumer 2 Consumer 1 Task Task Task Task modulo 2 id:1 id:2 id:3 id:4
17. Database queue (MySql table) Consumer 2 Consumer 1 Task Task Task Task modulo 3 id:1 id:2 id:3 id:4 Scaling problem need to restart all workers Consumer 3
19. Producer Consuming task Consumer redis Producer executes RPUSH to put tasks on end of queue under specified queue name Task Task Task id:1 id:2 id:3 Task id:4 Redis queue
20. Producer Consuming task Consumer redis Database queue (MySql table) Task Task Task id:2 id:3 id:4 Task id:1 Consumer executes LRANGE to retrieve entries from the beginning of queue and LTRIM to remove them after processed
21. Redis queue Consumer Task Task Task Task id:1 id:2 id:3 id:4 queue.task
22. Redis queue Consumer 2 Consumer 1 Task Task Task Task id:1 id:3 id:5 id:7 Task Task Task Task id:2 id:4 id:6 id:8 queue.task.1 queue.task.2
23. Redis queue Consumer 1 Task Task Task Task id:1 id:2 id:3 id:8 Task Task Task Task id:4 id:5 id:6 id:7 queue.task processing.queue.task Consumer 2 C1 C1 C2 C2
24. Redis queue Consumer 1 Task Task Task Task id:1 id:2 id:3 id:8 Task Task Task Task id:4 id:5 id:6 id:7 queue.task processing.queue.task Consumer 2 C1 C1 C2 C2 Lost task requeue
26. RabbitMQ Consumer 1 Task Task Task Task id:1 id:2 id:3 id:4 queue.task
27. RabbitMQ Consumer 1 Task Task Task Task id:1 id:2 id:3 id:4 queue.task Consumer 2
28. RabbitMQ Consumer 1 Task Task Task Task id:1 id:2 id:3 id:4 queue.task Consumer 2 Consumer 3 Consumer 4 Consumer 5 Consumer 6 Consumer 7 Consumer 9 Consumer 10 Consumer 11 Consumer 12
29. Acknowledge
30. Creates rabbitmq server instance with management plugin enabled http://localhost:15672 Management panel http://localhost:5672 AMQP API > docker run -d --hostname localhost --name rabbitmq-server -p 15672:15672 -p 5672:5672 rabbitmq:3-management
31. $messageQueue = $container->get( 'messaging.rpc.producer' ); $task = new SendAdExpiredMail(); $task->setAdId($ad[AdFields::FIELD_ID]); $messageQueue->send($task); Producer Publishing new task RabbitMQ
32. Producer Publishing new task RabbitMQ class SendAdExpiredMail extends AbstractMessage { public function setAdId(int $adId) { $this->payload['adId'] = $adId; } public function getAdId(): int { return $this->payload['adId']; } public function getQueueName(): string { return QueueName::MAILER_AD_EXPIRED; } }
33. Producer Task json payload RabbitMQ { "payload": { "adId": 1 }, "metadata": { "className": "SendAdExpiredMail" } }
34. $mailer = new MailHandler(); $queueMapper = new QueueMapper(); $queueMapper->addMapping( new QueueMapping('AdExpired', [$mailer, 'sendAdExpiredEmail'] ) ); Producer Consuming task - listener Consumer RabbitMQ
35. Producer Consuming task - listener Consumer RabbitMQ $messageExecutor = $container->get( 'messaging.rpc.executor' ); $messageExecutor->runOnVhostWithMap( 'mailing', $queueMapper );
36. Producer Consuming task - handler Consumer RabbitMQ class MailHandler { public function sendAdExpiredEmail (SendAdExpiredMail $task): bool { (...) } }
38. class MessageQueue { private $writer; private $serializer; public function __construct( MessageQueueWriterInterface $writer, MessageSerializerInterface $serializer ) { $this->writer = $writer; $this->serializer = $serializer; } public function send(MessageInterface $message) { $this->writer->write( $message->getQueueName(), $this->serializer->serialize($message) ); } } Queuing class Write model High abstraction level
39. class RabbitMqMessageQueueWriter implements MessageQueueWriterInterface { public function __construct(RabbitMqConnection $connection) { $this->connection = $connection; } public function write(string $queueName, string $payload) { $this->connection->setVHost($this->getVHostFromName($queueName)); $this->connection->connect(); $channel = $this->connection->openChannel(); $channel->publish( new AMQPMessage($payload), $this->getExchangeFromName($queueName), $this->getRoutingKeyFromName($queueName) ); $channel->close(); $this->connection->disconnect(); } } Write model implementation
40. class RabbitMqChannel { public function publish( AMQPMessage $message, string $exchangeName, string $routingKey ) { $this->channel->exchange_declare( $exchangeName, 'topic', false, // passive true, // durable false // auto_delete ); $this->channel->basic_publish( $message, $exchangeName, $routingKey ); } } Queuing class Write model Low abstraction level
41. class JsonMessageSerializer implements MessageSerializerInterface { const FIELD_PAYLOAD = 'payload'; const FIELD_METADATA = 'metadata'; public function serialize(MessageInterface $message): string { $unserialized = [ self::FIELD_PAYLOAD => $message->getPayload(), self::FIELD_METADATA => $message->getMetadata() ]; return json_encode($unserialized); } public function unserialize(string $serializedMessage): MessageInterface { $unserialized = json_decode($serializedMessage, true); $payload = $unserialized[self::FIELD_PAYLOAD]; $metadata = $unserialized[self::FIELD_METADATA]; $messageType = $metadata['className']; $message = new $messageType(); $message->setPayload($payload); $message->setMetadata($metadata); return $message; } } Task serializer
42. Task abstraction abstract class AbstractMessage implements MessageInterface{ public function getQueueName(): string{ return $this->queueName; } public function getPayload(): array { return $this->payload; } public function getMetadata(): array { return array_merge($this->metadata, ['className' => static::class]); } public function setMetadata(array $metadata) { $this->metadata = $metadata; } public function setPayload(array $payload) { $this->payload = $payload; } }
43. class MessageQueueRPCExecutor { private $connection; public function __construct(RabbitMqConnection $connection) { $this->connection = $connection; } public function runOnVhostWithMap( string $vhost, QueueMapper $queueMapper) { $this->connection->setVHost($vhost); $messageExecutor = new MessageQueueExecutor( new RabbitMqMessageQueueExecutor( $this->connection ), new JsonMessageSerializer(), $queueMapper ); $messageExecutor->run(); } } Read model - executor
44. class RabbitMqMessageQueueExecutor implements MessageQueueExecutorInterface { public function __construct(RabbitMqConnection $connection) { $this->connection = $connection; } public function run() { $channel = $this->connection->openChannel(); foreach ($this->mapping as $queueName => $executor) {a $channel->bindQueueToExecutor($queueName, $executor); } $channel->listen(); $channel->close(); } public function mapQueueToExecutor(string $queueName, callable $executor) { $this->mapping[$queueName] = $this->getCallbackWrapper($executor); } Read model - executor
45. private function getCallbackWrapper(callable $executor) { return function (AMQPMessage $message) use ($executor) { $success = $executor(new RabbitMqMessage($message)); if ($success) { $message->delivery_info['channel']->basic_ack( $message->delivery_info['delivery_tag'] ); } else { $message->delivery_info['channel']->basic_nack( $message->delivery_info['delivery_tag'], false, true ); } return $success; }; } Low level processing
46. class RabbitMqChannel { public function publish( AMQPMessage $message, string $exchangeName, string $routingKey ) { $this->channel->exchange_declare($exchangeName, 'topic', false, true, false); $this->channel->basic_publish($message, $exchangeName, $routingKey); } public function bindQueueToExecutor(string $queueName, callable $executor) { $this->channel->queue_declare($queueName, false, true, false, false); $this->channel->basic_consume($queueName, '', false, false, false, false, $executor); } public function listen() { while(count($this->channel->callbacks)) { $this->channel->wait(); } } } Low level read
47. Producer Connect Open channel Declare exchange Basic publish Write flow
48. Connect Open channel Declare queue Basic consume Consumer Read flow
49. Consumer Producer Exchange Queue Application RabbitMQ Routing binding
50. AdExpired AdModerated AdExpired AdModerated mailing_olxpl mailing_olxpt vhost
51. Exchange types direct topic fanout header Exchange Queue A Queue B mailer.ad.expired mailer.ad.moderated mailer.ad.expired
53. Exchange types direct topic fanout headers Exchange Queue A Queue B mailer.ad.expired mailer.ad.* mailer.user.*
54. Exchange types direct topic fanout headers Exchange Mailer queue Stats queue mailer.ad.expired mailer.ad.* mailer.#
55. Exchange types direct topic fanout headers Exchange Mailer queue Mail stats queue mailer.ad.expired
56. Exchange types direct topic fanout headers Exchange Mailer queue Mail stats queue param1:1 param2:2
57. Exchange Mailer queue Mail stats queue mailer.ad.expired Delayed delivery Exchange Exchange to exchange binding rabbitmq-plugins enable rabbitmq_delayed_message_exchange
58. Delaying message delivery $channel->exchange_declare('delayed_exchange', 'x-delayed-message', false, true, false, false, false, new AMQPTable(array( "x-delayed-type" => "direct" ))); $hdrs = new AMQPTable(array("x-delay" => 300)); $msg = new AMQPMessage('5 minute delayed task', array('delivery_mode' => 2) ); $msg->set('application_headers', $hdrs); $channel->basic_publish($msg, 'delayed_exchange', 'routing_key');
59. Exchange Mailer queue Mail stats queue mailer.ad.expired Delayed delivery Exchange Exchange to exchange binding Delayed retries Producer Consumer Failed task execution
60. Handling failed tasks Ignore repetition { "payload": { "adId": 1 }, "metadata": { "className": "SendAdExpiredMail", "retryStrategy": "ignore" } }
61. Handling failed tasks Controlled repetition { "payload": { "adId": 1 }, "metadata": { "className": "SendAdExpiredMail", "retryStrategy": "repeat", "retryLimit": "5", "retryCount": "0", } }
62. Handling failed tasks Controlled repetition with declared intervals { "payload": { "adId": 1 }, "metadata": { "className": "SendAdExpiredMail", "retryStrategy": "repeat", "retryLimit": "5", "retryCount": "0", "retryInterval": [15, 30, 60, 300, 86400], } }
63. Fire and Forget
64. Producer Consumer Asynchronous RPC call
65. Producer Consumer Asynchronous RPC call Asynchronous RPC response
66. Send task Wait for response Create response hash Open response queue Continue processing
67. interface ResponseRPC { public function getResponseQueueName(): string; } class GetUserFriends extends AbstractMessage implements ResponseRPC { public function getResponseQueueName(): string { if (!isset($this->metadata['response_queue'])) { $this->metadata['response_queue'] = 'user-friends-' . $this->payload['user_id'] . '_' . uniqid('', true); } return $this->metadata['response_queue']; }
68. $messageQueue = $this->container->get('messaging.rpc.producer.responding'); $task = new GetUserFriends(); $task->setId($user[UserFields::FIELD_ID]); $messageQueue->send($task); /* do some processing */ $userFriends = $messageQueue->getAsyncResponse();
69. class RespondingMessageQueue extends MessageQueue { public function __construct( MessageQueue $writer, MessageQueueExecutorInterface $executor) { $this->writer = $writer; $this->executor = $executor; } public function send(MessageInterface $message) { if ($message instanceof ResponseRPC){ $this->openResponseQueue($message); } $this->writer->send($message); } public function getAsyncResponse(): MessageInterface { $this->executor->runSingle(10); return $this->response; } private function openResponseQueue(ResponseRPC $message) { $message->getResponseQueueName(); $this->executor->mapQueueToExecutor( $message->getResponseQueueName(),[$this, 'catchResponse']); } private function catchResponse(MessageInterface $message) { $this->response = $message; } }
70. Producer Consumer Asynchronous RPC calls Asynchronous RPC responses
71. user(id:27192) { active_ads observed_ads recommendations messages } GraphQL Sample request
72. getActiveAds getObservedAds getRecommendations getMessages Request Response
73. getActiveAds getObservedAds getRecommendations getMessages Request Response
74. $messageQueue = $this->container->get('messaging.rpc.producer.responding'); $getActiveAds = new GetActiveAds(); $getActiveAds->setUserId($user[UserFields::FIELD_ID]); $getObservedAds = new GetObservedAds(); $getObservedAds->setUserId($user[UserFields::FIELD_ID]); $getRecommendations = new GetRecommendations(); $getRecommendations->setUserId($user[UserFields::FIELD_ID]); $getMessages = new GetMessages(); $getMessages->setUserId($user[UserFields::FIELD_ID]); $messageQueue->sendMultiple( [$getActiveAds, $getObservedAds, $getRecommendations, $getMessages] ); /* do some processing */ $userAggregateData = $messageQueue->getAsyncResponses();
75. • more consumers • faster consumers • higher prefetch count
76. Useful links http://tryrabbitmq.com/ https://github.com/php-amqplib/php-amqplib https://www.cloudamqp.com/
77. RabbitMQ Asynchronous RPC with Dawid Mazurek PHP DEV at