Рецепты RabbitMQ
•
0 likes•980 views
Слайды с Highload++ 2016, Описаны паттерны использоваиния сообщений, Протокол AMQP, предложены рацепты использования RabbitMQ
![Публикация сообщения
$rabbit = new AMQPConnection();
$res = $rabbit->connect();
$channel = new AMQPChannel($rabbit);
$exchange = new AMQPExchange($channel);
$exchange->setName('to_worker');
$i=0;
while ($i < 20) {
$i++;
$data = json_encode([1,2,3,
rand(0, 1000)]);
$exchange->publish($data,
'logs',
0,
['delivery_mode'=>2,
'content_type'=> 'text/json']
);
}](https://image.slidesharecdn.com/rabbitmq-161115065214/85/RabbitMQ-49-320.jpg)
![Публикация сообщения
$rabbit = new AMQPConnection();
$res = $rabbit->connect();
$channel = new AMQPChannel($rabbit);
$exchange = new AMQPExchange($channel);
$exchange->setName('to_worker');
$i=0;
while ($i < 20) {
$i++;
$data = json_encode([1,2,3,
rand(0, 1000)]);
$exchange->publish($data,
'logs',
0,
['delivery_mode'=>2,
'content_type'=> 'text/json']
);
}
Routing key Persisten](https://image.slidesharecdn.com/rabbitmq-161115065214/85/RabbitMQ-50-320.jpg)
![Синхронное CONSUME
function processMessage($envelope, $queue) {
echo 'Message: [',
$envelope->getDeliveryTag(),
']',
$envelope->getBody(),PHP_EOL;
$queue->ack( $envelope->getDeliveryTag() );
}
$rabbit = new AMQPConnection();
$res = $rabbit->connect();
$channel = new AMQPChannel($rabbit);
$queue = new AMQPQueue($channel);
$queue->setName('worker');
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->consume("processMessage");](https://image.slidesharecdn.com/rabbitmq-161115065214/85/RabbitMQ-54-320.jpg)
![Синхронное CONSUME
function processMessage($envelope, $queue) {
echo 'Message: [',
$envelope->getDeliveryTag(),
']',
$envelope->getBody(),PHP_EOL;
$queue->ack( $envelope->getDeliveryTag() );
}
$rabbit = new AMQPConnection();
$res = $rabbit->connect();
$channel = new AMQPChannel($rabbit);
$queue = new AMQPQueue($channel);
$queue->setName('worker');
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->consume("processMessage");](https://image.slidesharecdn.com/rabbitmq-161115065214/85/RabbitMQ-55-320.jpg)
![function processMessage($envelope, $queue) {
echo 'Message: [',
$envelope->getDeliveryTag(),
']',
$envelope->getBody(),PHP_EOL;
$queue->ack( $envelope->getDeliveryTag() );
}
$rabbit = new AMQPConnection();
$res = $rabbit->connect();
$channel = new AMQPChannel($rabbit);
$queue = new AMQPQueue($channel);
$queue->setName('worker');
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->consume("processMessage");
Синхронное CONSUME](https://image.slidesharecdn.com/rabbitmq-161115065214/85/RabbitMQ-56-320.jpg)
![Несколько каналов
$exchange = new AMQPExchange($channel);
$exchange->setName('to_worker');
$ex->setType(AMQP_EX_TYPE_HEADERS);
$data = json_encode([1,2,3]);
$headers = ['operation' => 'parsing'];
$args = [
'content_type'=> 'text/json',
'headers' => $headers
];
$exchange->publish($data,'work',0,$args);](https://image.slidesharecdn.com/rabbitmq-161115065214/85/RabbitMQ-72-320.jpg)
![Несколько каналов
$exchange = new AMQPExchange($channel);
$exchange->setName('to_worker');
$ex->setType(AMQP_EX_TYPE_HEADERS);
$data = json_encode([1,2,3]);
$headers = ['operation' => 'parsing'];
$args = [
'content_type'=> 'text/json',
'headers' => $headers
];
$exchange->publish($data,'work',0,$args);](https://image.slidesharecdn.com/rabbitmq-161115065214/85/RabbitMQ-73-320.jpg)
![Correlation Identifier
$cr_id = $message->getArgument('correlation_id');
$exchange->publish( $message, '', ['correlation_id=>$cr_id]);](https://image.slidesharecdn.com/rabbitmq-161115065214/85/RabbitMQ-102-320.jpg)
Рецепты RabbitMQ
- 2. Пару слов об pipeline архитектуре https://habrahabr.ru/company/oleg-bunin/blog/310418/
- 6. Пример: аналог Yandex.market Загрузка YML Сведение в pricelist Парсинг YML Загрузка img Задачи за кадром
- 7. Пример: аналог Yandex.market Загрузка YML Сведение в pricelist Парсинг YML Загрузка img Задачи за кадром
- 9. Cообщения ● Позволяют асинхронно взаимодействоавть между элементами архитектуры ● Сохраняют слабую связанность между элементами архитектуры ● Позволяют взаимодействоавть с разными частями, PHP – Phython – Java - Go ...
- 14. Routers
- 17. Exchange ● Принимает сообщения ● Имеет имя ● Имеет тип: - fanout - direct - topic - headers AMQP : Exchange (обмен)
- 18. Exchange AMQP : Exchange (обмен) ● Принимает сообщения ● Имеет имя ● Имеет тип Имеет свойства: - autodelete - transit - durable
- 19. Exchange AMQP : Exchange (обмен) $exchange = new AMQPExchange($channel); $exchange->setName('MyExchange';) $exchange->declare();
- 20. AMQP : Queue (очередь) Queue ● Отдает сообщения адресату по принципу FIFO ● Имеет имя ● Имеет свойства: - autodelete - durable - exclusive
- 21. AMQP : Queue (очередь) Queue $queue = new AMQPQueue($channel); $queue->setName('MyQueue'); $queue->declare();
- 22. AMQP : Bind (связь) ● Между Exchange и Queue определяем связь (или маршрут): Bind ● Имеет ключ RoutingKey, в соответствии с которым определяется маршрут сообщения QueueExchange Bind
- 23. AMQP : Bind (связь) QueueExchange Bind $queue = new AMQPQueue($channel); $queue->setName('MyQueue'); $queue->bind('MyExchange', $key);
- 24. AMQP : Bind (связь) QueueExchange Bind $exchange = new AMQPExchange($channel); $exchange->setName('MyExchange'); $exchange->bind('MyQueue', $key);
- 25. AMQP : Bind (связь) QueueExchange Bind $exchange = new AMQPExchange($channel); $exchange->setName('MyExchange'); $exchange->bind('MyQueue', $key); Объявляется один раз
- 26. AMQP : Internals
- 27. AMQP : Message
- 28. Exchange Состоит: ● тела ● routing key ● заголовков ● имеет аттрибуты AMQP : Message
- 29. ● Expiration period ● Message publishing timeshamp Exchange Аттрибуты: ● Content-type ● Content-encoding ● Delivery-mode ● Message priority ● Application id AMQP : Message
- 30. Exchange AMQP : Message В зависимости от типа Exchange и routingKey сообщения определяется маршрут сообщения Queue 1 Queue 2
- 31. Exchange AMQP : Message Fanout – ключ не учитываем Direct – полное совпадение Topic – совпадение по маске Queue 1 Queue 2
- 33. Exchange AMQP : Message Fanout – самый быстрый Direct – середина на половине Topic – сомый медленный Queue 1 Queue 2
- 34. Exchange Отправить сообщение $exchange = new AMQPExchange($channel); $exchange->setName('MyExchange';) $exchange->publish($message, $key);
- 35. AMQP : ACK QueueExchange Bind На каждое принятое сообщение должна быть отослана “квитанция” ACK (Acknowledgements).
- 36. AMQP : ACK QueueExchange Bind На каждое принятое сообщение должна быть отослана “квитанция” ACK (Acknowledgements). Если Очередь имеет св-во AUTOACK то подтверждение специально отправлять не нужно ACK
- 39. AMQP : ACK workerWEB Заказ ACK Удачное соединение
- 40. AMQP : ACK workerWEB Заказ Удачное соединение Не удачноне соединение Делаем повторный запрос
- 42. Объявление очереди и обмена // создание очереди, обмена и их привязки друг к другу. $rabbit = new AMQPConnection(); $res = $rabbit->connect(); $channel = new AMQPChannel($rabbit); $exchange = new AMQPExchange($channel); $exchange->setFlags(AMQP_DURABLE); $exchange->setName('to_worker'); $exchange->setType('direct'); $exchange->declare(); $queue = new AMQPQueue($channel); $queue->setFlags(AMQP_DURABLE); $queue->setName('worker');
- 43. Объявление очереди и обмена // создание очереди, обмена и их привязки друг к другу. $rabbit = new AMQPConnection(); $res = $rabbit->connect(); $channel = new AMQPChannel($rabbit); $exchange = new AMQPExchange($channel); $exchange->setFlags(AMQP_DURABLE); $exchange->setName('to_worker'); $exchange->setType('direct'); $exchange->declare(); $queue = new AMQPQueue($channel); $queue->setFlags(AMQP_DURABLE); $queue->setName('worker'); AMQP_DURABLE AMQP_AUTODELETE AMQP_EXCLUSIVE AMQP_INTERNAL AMQP_IMMEDIATE AMQP_AUTOACK
- 44. Объявление очереди и обмена // создание очереди, обмена и их привязки друг к другу. $rabbit = new AMQPConnection(); $res = $rabbit->connect(); $channel = new AMQPChannel($rabbit); $exchange = new AMQPExchange($channel); $exchange->setFlags(AMQP_DURABLE); $exchange->setName('to_worker'); $exchange->setType('direct'); $exchange->declare(); $queue = new AMQPQueue($channel); $queue->setFlags(AMQP_DURABLE); $queue->setName('worker'); AMQP_DURABLE AMQP_AUTODELETE AMQP_EXCLUSIVE AMQP_INTERNAL AMQP_IMMEDIATE AMQP_AUTOACK Это можно сделать один раз
- 45. Привзка очереди к обмену $rabbit = new AMQPConnection(); $res = $rabbit->connect(); $channel = new AMQPChannel($rabbit); $queue = new AMQPQueue($channel); $queue->setName('worker'); // $queue->declare(); $queue->bind('to_worker', 'logs'); $rabbit->disconnect();
- 46. Привзка очереди к обмену $rabbit = new AMQPConnection(); $res = $rabbit->connect(); $channel = new AMQPChannel($rabbit); $queue = new AMQPQueue($channel); $queue->setName('worker'); // $queue->declare(); $queue->bind('to_worker', 'logs'); $rabbit->disconnect(); Routing key
- 47. Привзка очереди к обмену Это можно сделать один раз
- 48. Привзка очереди к обмену $rabbit = new AMQPConnection(); $res = $rabbit->connect(); $channel = new AMQPChannel($rabbit); $queue = new AMQPQueue($channel); $queue->setName('worker'); // $queue->declare(); $queue->bind('to_worker', 'logs'); $rabbit->disconnect(); Это можно вообще не делать а использовать HTTP интерфейс
- 49. Публикация сообщения $rabbit = new AMQPConnection(); $res = $rabbit->connect(); $channel = new AMQPChannel($rabbit); $exchange = new AMQPExchange($channel); $exchange->setName('to_worker'); $i=0; while ($i < 20) { $i++; $data = json_encode([1,2,3, rand(0, 1000)]); $exchange->publish($data, 'logs', 0, ['delivery_mode'=>2, 'content_type'=> 'text/json'] ); }
- 50. Публикация сообщения $rabbit = new AMQPConnection(); $res = $rabbit->connect(); $channel = new AMQPChannel($rabbit); $exchange = new AMQPExchange($channel); $exchange->setName('to_worker'); $i=0; while ($i < 20) { $i++; $data = json_encode([1,2,3, rand(0, 1000)]); $exchange->publish($data, 'logs', 0, ['delivery_mode'=>2, 'content_type'=> 'text/json'] ); } Routing key Persisten
- 52. Асинхронное GET $rabbit = new AMQPConnection(); $rabbit->connect(); $channel = new AMQPChannel($rabbit $queue = new AMQPQueue($channel); $queue->setName('worker'); $msg = $queue->get();
- 53. Асинхронное GET
- 54. Синхронное CONSUME function processMessage($envelope, $queue) { echo 'Message: [', $envelope->getDeliveryTag(), ']', $envelope->getBody(),PHP_EOL; $queue->ack( $envelope->getDeliveryTag() ); } $rabbit = new AMQPConnection(); $res = $rabbit->connect(); $channel = new AMQPChannel($rabbit); $queue = new AMQPQueue($channel); $queue->setName('worker'); $queue->setFlags(AMQP_EXCLUSIVE); $queue->consume("processMessage");
- 55. Синхронное CONSUME function processMessage($envelope, $queue) { echo 'Message: [', $envelope->getDeliveryTag(), ']', $envelope->getBody(),PHP_EOL; $queue->ack( $envelope->getDeliveryTag() ); } $rabbit = new AMQPConnection(); $res = $rabbit->connect(); $channel = new AMQPChannel($rabbit); $queue = new AMQPQueue($channel); $queue->setName('worker'); $queue->setFlags(AMQP_EXCLUSIVE); $queue->consume("processMessage");
- 56. function processMessage($envelope, $queue) { echo 'Message: [', $envelope->getDeliveryTag(), ']', $envelope->getBody(),PHP_EOL; $queue->ack( $envelope->getDeliveryTag() ); } $rabbit = new AMQPConnection(); $res = $rabbit->connect(); $channel = new AMQPChannel($rabbit); $queue = new AMQPQueue($channel); $queue->setName('worker'); $queue->setFlags(AMQP_EXCLUSIVE); $queue->consume("processMessage"); Синхронное CONSUME
- 57. Асинхронное GET
- 58. Publish/Consume
- 59. Publish/Consume В RannitMQ работает не так как ожидается
- 64. Publish/Consume Загрузка YML Загрузка картинок Загрузка картинок 33.3% 33.3% 33.3% Загрузка картинок
- 66. Несколько каналов Парсинг YML Загрузка картинок Channel -1 Channel -2 Загрузка YML content parsing uploadingExchange: Content, funout, Bind: parsing durable Bind: uploading Queue 1: parsing, durable Queue 2: uploading, durable
- 67. Несколько каналов(вар 2) Парсинг YML Загрузка картинок Channel -1 Channel -2 Загрузка YML content parsing uploadingExchange: Content, header, durable Queue 1: parsing, durable Queue 2: uploading, durable Bind: operation:x-parsing Bind: operation:x-uploading
- 68. Несколько каналов(вар 2) Парсинг YML Загрузка картинок Channel -1 Channel -2 Загрузка YML content parsing uploadingExchange: Content, header, durable Queue 1: parsing, durable Queue 2: uploading, durable Bind: operation:x-parsing Bind: operation:x-uploading
- 69. Несколько каналов(вар 2) Парсинг YML Загрузка картинок Channel -1 Channel -2 Загрузка YML content parsing uploadingExchange: Content, header, durable Queue 1: parsing, durable Queue 2: uploading, durable Bind: operation:x-parsing Bind: operation:x-uploading
- 70. Несколько каналов import pika connection = pika.BlockingConnection() ch = connection.channel() ch.queue_bind( exchange='content', queue='parser', arguments={'operation': 'parsing', 'x- match':'any'})
- 71. Несколько каналов import pika connection = pika.BlockingConnection() ch = connection.channel() ch.queue_bind( exchange='content', queue='parser', arguments={'operation': 'parser', 'x- match':'any'}) в PHP Нельзя задать аргументы для Bind
- 72. Несколько каналов $exchange = new AMQPExchange($channel); $exchange->setName('to_worker'); $ex->setType(AMQP_EX_TYPE_HEADERS); $data = json_encode([1,2,3]); $headers = ['operation' => 'parsing']; $args = [ 'content_type'=> 'text/json', 'headers' => $headers ]; $exchange->publish($data,'work',0,$args);
- 73. Несколько каналов $exchange = new AMQPExchange($channel); $exchange->setName('to_worker'); $ex->setType(AMQP_EX_TYPE_HEADERS); $data = json_encode([1,2,3]); $headers = ['operation' => 'parsing']; $args = [ 'content_type'=> 'text/json', 'headers' => $headers ]; $exchange->publish($data,'work',0,$args);
- 75. Синхронное / Асинхронное what is the question ? WEB AJAX лучше асинхронное WebSocket Синхронное
- 76. Синхронное / Асинхронное WEB AJAX лучше асинхронное AMQP REST https://github.com/akalend/amqp-rest
- 77. Синхронное / Асинхронное WEB AJAX лучше асинхронное AMQP REST https://github.com/akalend/amqp-rest NGX-AMQP https://github.com/WPMedia/nginx-amqp
- 78. Синхронное / Асинхронное WEB AJAX лучше асинхронное AMQP REST https://github.com/akalend/amqp-rest NGX-AMQP https://github.com/WPMedia/nginx-amqpНадо делать через Upstreem & subquery
- 79. Синхронное / Асинхронное WEB AJAX лучше асинхронное AMQP REST https://github.com/akalend/amqp-rest NGX-AMQP https://github.com/WPMedia/nginx-amqp NGX-RABBIT https://github.com/wingify/lua-resty-rabbitmqstomp
- 80. Синхронное / Асинхронное WebSocket Синхронное NGX-RABBIT https://github.com/wingify/lua-resty-rabbitmqstomp Node.js
- 81. Routers
- 82. Exchange Route Message Direct – совпадение по ключу ● Bind Queue_1 logs ● Bind Queue_2 worker Queue 1 Queue 2
- 83. Exchange Route Message Headers – совпадение по заголовку ● Bind Queue_1 {'operation' : 'logs', 'x-match':'any'} ● Bind Queue_2 {'operation' : 'worker' , 'x-match':'any'} Queue 1 Queue 2
- 84. Messages
- 89. Загрузка видео с видео хостинга Load Convert WEB Клиентский скрипт Получаем статус Выполнено Curl
- 90. Channels
- 92. Poin-to-Point Channel Exchange : durable Queue : durable, exclusive
- 94. Guaranteed Delivery Exchange : durable= 1 Message : delivery-type = 2 Queue : durable= 1, autodelete = 0, autoack = 1
- 95. Guaranteed Delivery $delivery-tag = $message->delivery-tag; $queue->ack($delivery-tag);
- 97. Запасной канал
- 98. $exchange = new AMQPExchange($channel); $exchange->setFlags(AMQP_DURABLE); $exchange->setName('to_worker'); $exchange->setType(AMQP_EX_TYPE_DIRECT); $exchange->declare(); $queue = new AMQPQueue($channel); $queue->setFlags(AMQP_AUTODELETE); $queue->setName('worker'); $queue->declare(); $queue->bind('to_worker', 'work'); $queue->setFlags(AMQP_DURABLE); $queue->setName('garbage'); $queue->declare(); $queue->bind('to_worker', 'work'); Запасной канал
- 100. Request Reply
- 102. Correlation Identifier $cr_id = $message->getArgument('correlation_id'); $exchange->publish( $message, '', ['correlation_id=>$cr_id]);
- 103. Datatype channel
- 104. debugging
- 106. Rabbit MQ 2015 № 11 Кролик в песочнице
- 107. Rabbit MQ 2015 № 12 RabbitMQ. Вырастаем из штанишек.
- 108. книги
- 109. книги
- 110. книги
- 111. книги
- 112. Cпасибо за внимание