Maison > Article > développement back-end > Premiers pas avec la file d'attente de messages RabbitMQ et explication détaillée des exemples PHP
Premiers pas avec la file d'attente de messages RabbitMQ et explication détaillée des exemples PHP
- WBOYavant
- 2022-05-12 18:33:527888parcourir
Cet article vous apporte des connaissances pertinentes sur PHP Il présente principalement l'introduction de la file d'attente de messages RabbitMQ et quelques détails pratiques. La file d'attente de messages est une méthode de communication entre les applications. J'espère que cela vous sera utile. tout le monde.
Apprentissage recommandé : "Tutoriel vidéo PHP"
Introduction aux scénarios d'application de file d'attente de messages et de file d'attente de messages
RabbitMQ
Explication
MQ (Message Queue ) est une file d'attente de messages, qui est entre les applications Méthode de communication, le message peut être renvoyé immédiatement après avoir été envoyé et le système de messagerie assure la livraison fiable du message. La « file d'attente des messages » est un conteneur qui enregistre les messages lors de leur transmission. C'est typique : modèle producteur, consommateur. Les producteurs continuent de produire des messages dans la file d'attente des messages et les consommateurs continuent d'obtenir des messages de la file d'attente. Parce que la production et la consommation de messages sont à la fois asynchrones et ne concernent que l’envoi et la réception de messages, il n’y a aucune intrusion de logique métier, réalisant ainsi le découplage entre producteurs et consommateurs.
Pourquoi utiliser un middleware de messages ?
La file d'attente de messages est un composant important dans les systèmes distribués, résolvant des problèmes tels que le découplage des applications, les messages asynchrones, l'écrêtement des pics de trafic, etc., et permettant d'obtenir une concurrence élevée, une haute disponibilité, une évolutivité et une architecture de cohérence éventuelle
Traitement asynchrone
Utilisateur inscription Après les informations, vous devez envoyer des e-mails et enregistrer des messages texte
1. Une fois les informations d'enregistrement de l'utilisateur écrites dans la base de données, même si l'enregistrement est réussi, les informations seront renvoyées
2. L'envoi d'e-mails et l'enregistrement de messages texte sont exécutés de manière asynchrone via la file d'attente des messages, et l'utilisateur n'a pas besoin d'attendre ces deux opérations
Découplage d'application
Une fois que l'utilisateur a passé une commande, le système de commande doit en informer le système d'inventaire. L'approche traditionnelle est que le système de commande appelle l'interface du système d'inventaire pour augmenter ou diminuer l'inventaire
1. L'utilisateur passe une commande de production et renvoie une invite de réussite
2. Le système d'inventaire de consommation de file d'attente augmente ou diminue l'inventaire 
Réduction des pics de trafic
L'écrêtage des pics de trafic est également un scénario courant dans les files d'attente de messages. Il est généralement largement utilisé dans les ventes flash ou les activités de saisie de groupe
1. Lorsqu'un groupe d'utilisateurs demande à entrer dans la file d'attente, contrôlez le nombre de files d'attente. . Si le nombre dépasse un certain nombre, la vente flash prendra fin
2. Ensuite, les files d'attente sont consommées une par une selon le principe du premier entré, premier sorti
Caractéristiques de Rabbitmq
Fiabilité (Fiabilité) RabbitMQ. utilise certains mécanismes pour garantir la fiabilité, tels que la persistance, la confirmation de transmission et la confirmation de publication.
Routage flexible Le message est acheminé via Exchange avant qu'il n'entre dans la file d'attente. Pour les fonctionnalités de routage typiques, RabbitMQ fournit déjà certaines implémentations Exchange intégrées. Pour des fonctions de routage plus complexes, plusieurs Exchanges peuvent être liés ensemble, ou votre propre Exchange peut être implémenté via le mécanisme de plug-in.
Message Clustering Plusieurs serveurs RabbitMQ peuvent former un cluster pour former un courtier logique.
Files d'attente hautement disponibles Les files d'attente peuvent être mises en miroir sur les machines du cluster, de sorte qu'elles soient toujours disponibles en cas de défaillance de certains nœuds.
Multi-protocole RabbitMQ prend en charge plusieurs protocoles de file d'attente de messages, tels que STOMP, MQTT, etc.
De nombreux clients RabbitMQ prend en charge presque tous les langages couramment utilisés, tels que PHP Java, .NET, Ruby, etc.
Interface utilisateur de gestion RabbitMQ fournit une interface utilisateur facile à utiliser qui permet aux utilisateurs de surveiller et de gérer de nombreux aspects du courtier de messages.
Traçage Si le message est anormal, RabbitMQ fournit un mécanisme de suivi des messages afin que les utilisateurs puissent découvrir ce qui s'est passé.
Plugin System RabbitMQ fournit de nombreux plug-ins pour l'étendre sous de nombreux aspects, et vous pouvez également écrire vos propres plug-ins.
Comment fonctionne RabbitMQ
Broker : Application qui reçoit et distribue des messages, RabbitMQ Server est le courtier de messages.
Hôte virtuel : Semblable à une base de données MySQL, lorsque plusieurs utilisateurs différents utilisent les services fournis par le même serveur RabbitMQ, plusieurs vhosts peuvent être divisés et chaque utilisateur crée un échange/une file d'attente, etc. dans son propre vhost.
Connexion : connexion TCP entre l'éditeur/consommateur et le courtier.
Canal : Si une connexion est établie à chaque accès à RabbitMQ, la surcharge liée à l'établissement d'une connexion TCP sera énorme et l'efficacité sera faible lorsque le volume de messages est important. Channel est une connexion logique établie à l'intérieur de la connexion. En tant que connexion légère, Channel réduit considérablement le coût d'établissement d'une connexion TCP par le système d'exploitation.
Exchange : Le message atteint le premier arrêt du courtier. Selon les règles de distribution, il correspond à la clé de routage dans la table de requête et distribue le message dans la file d'attente. Les types couramment utilisés sont : direct (point à point), sujet (publication-abonnement) et fanout (multidiffusion).
Queue : Le message est finalement envoyé ici pour être récupéré par le consommateur. Un message peut être copié dans plusieurs files d'attente en même temps.
L'installation de RabbitMQ démarre
Adresse officielle de RabbitMQ : http://www.rabbitmq.com
Pour installer Rabbitmq, vous devez d'abord installer Erlang
Étape 1 : installation d'Erlang
Pour installer Rabbitmq, vous devez installer Erlang tout d'abord, centos7 ne prend pas en charge l'installation de la version erlang 24 
Télécharger : 
# 系统 centos 7# 下载erlang包，手动下载后上传至服务器，我在使用wget下载后无法安装，这里没明白 # 安装 yum install erlang-23.3.4.4-1.el7.x86_64.rpm # 验证安装是否成功 erl
Step 2 : Installez Rabbitmq 
# 系统 centos 7# 下载rabbitmq包，手动下载后上传至服务器，我在使用wget下载后无法安装，这里没明白 # 安装 yum install rabbitmq-server-3.8.19-1.el7.noarch.rpm # 启动 systemctl start rabbitmq-server # 关闭 systemctl stop rabbitmq-server # 查看默认端口服务是否启动 netstat -tunlp
La file d'attente de messages PHP Rabbitmq est utilisée dans différents modes
Interface de gestion Rabbitmq et utilisation en ligne de commande
4369 : epmd (Erlang Port Mapper Daemon), port de service erlang
5672 : port de communication client
15672 : Client API HTTP, gestion L'UI (uniquement si le plugin de gestion est activé) ne démarre pas forcément
25672 : Utilisé pour la communication inter-nœuds (port du serveur de distribution Erlang)
Commandes de gestion Rabbitmq
Démarrage 15672 : Client API HTTP , interface utilisateur de gestion (Uniquement lorsque le plug-in de gestion est activé)
# 启动rabbitmq_management插件 rabbitmq-plugins enable rabbitmq_management # 查看所有插件 rabbitmq-plugins list
Testez l'accès à l'interface utilisateur : (Pour le moment, les adresses non locales ne peuvent pas se connecter)
http://192.168.10.105:15672/
rabbitmq Configurer l'interface de gestion
# 新增一个用户 rabbitmqctl add_user 【用户名Username】 【密码Password】 rabbitmqctl add_user root root # 删除一个用户 rabbitmqctl delete_user Username # 修改用户的密码 rabbitmqctl change_password Username Newpassword # 查看当前用户列表 rabbitmqctl list_users # 设置用户角色的命令为： rabbitmqctl set_user_tags User Tag rabbitmqctl set_user_tags root administrator # User为用户名， Tag为角色名(对应于上面的administrator，monitoring，policymaker，management，或其他自定义名称)。
Création en ligne de commande du vhost et installation de l'extension PHP
Semblable à la base de données mysql, lorsque plusieurs utilisateurs différents utilisent les services fournis par le même serveur RabbitMQ, plusieurs vhosts peuvent être divisés et chaque utilisateur crée un échange/une file d'attente, etc.
1) Afficher les vhosts de différents utilisateurs
Créer des vhosts et attribuer des autorisations
# 新增vhost rabbitmqctl add_vhost vhostname rabbitmqctl add_vhost order # 查看vhost列表 rabbitmqctl list_vhosts #为vhost添加用户 rabbitmqctl set_permissions -p vhostname username ".*" ".*" ".*"rabbitmqctl set_permissions -p order root ".*" ".*" ".*" ".*" ".*" ".*"后边三个.*分别代表：配置权限、写权限、读权限
2) Installer l'installation de l'extension RabbitMQ pour php
https://github.com/php-amqplib/php- Installation de l'extension amqplib
Modifiez l'image Alibaba Cloud
composer config -g repo.packagist composer https://mirrors.aliyun.com/composer/
Démarrez le téléchargement – Parfois, il sera téléchargé vers une version inférieure à 2.8, vous devez spécifier la version
Si le téléchargement échoue, mettez à niveau composer et php.ini Open. sockets pour étendre et changer d'images domestiques
# 升级composer composer self-update #php.ini 打开 sockets 扩展 #下载指定版本 composer require php-amqplib/php-amqplib=^3.0
simple mode producteur message pousser vers la file d'attente des messages
Documentation :
https://www.rabbitmq.com/tutorials/tutorial-one-php.html
Simple producteur et messager
Code producteur
http://localhost/rabbitmq/simple/pro.php
<?php require_once "../vendor/autoload.php"; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; //生产者 //Connection: publisher／consumer和broker之间的TCP连接 //Channel: 如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销。 //建立connction $connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order'); //Channel $channel = $connection->channel(); //声明队列名为：goods $queue_name = 'goods'; $channel->queue_declare($queue_name, false, true, false, false); //生产数据 $data = 'this is messge'; //创建消息 $msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]); //发布消息 $channel->basic_publish($msg, $exchange = '', $queue_name); //关闭连接 $channel->close(); $connection->close();
运行生产者脚本：
http://localhost/rabbitmq/simple/pro.php
点击goods队列可以进入到消息详情
simple模式消费者接受消息
http://localhost/rabbitmq/simple/con.php
<?php require_once "../vendor/autoload.php";
use PhpAmqpLib\Connection\AMQPStreamConnection;
//建立connction
$connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order');
//Channel
$channel = $connection->channel();
//声明队列名为：goods
$queue_name = 'goods';
$channel->queue_declare($queue_name, false, true, false, false);
echo " [*] Waiting for messages. To exit press CTRL+C\n";
$callback = function ($msg) {
echo 'received = ', $msg->body . "\n";
};
//开启消费
$channel->basic_consume($queue_name, '', false, true, false, false, $callback);
//不断的循环进行消费
while ($channel->is_open()) {
$channel->wait();
}
//关闭连接
$channel->close();
$connection->close();
worker模式生产消费消息
rabbitmq Work Queues
一个生产者对应多个消费者，消费特别慢时增加几个消费分发
生产者，和上文生产者不变
<?php require_once "../vendor/autoload.php"; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; //生产者 //Connection: publisher／consumer和broker之间的TCP连接 //Channel: 如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销。 //建立connction $connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order'); //Channel $channel = $connection->channel(); //声明队列名为：task_queue $queue_name = 'task_queue'; $channel->queue_declare($queue_name, false, true, false, false); for ($i = 0; $i AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]); //发布消息 $channel->basic_publish($msg, $exchange = '', $queue_name); } //关闭连接 $channel->close(); $connection->close();
消费者worker1
D:\phpstudy_pro\WWW\rabbitmq\worker\worker1.php
<?php require_once "../vendor/autoload.php";
use PhpAmqpLib\Connection\AMQPStreamConnection;
//建立connction
$connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order');
//Channel
$channel = $connection->channel();
//声明队列名为：task_queue
$queue_name = 'task_queue';
$channel->queue_declare($queue_name, false, true, false, false);
echo " [*] Waiting for messages. To exit press CTRL+C\n";
$callback = function ($msg) {
echo 'received = ', $msg->body . "\n";
};
//开启消费
$channel->basic_consume($queue_name, '', false, true, false, false, $callback);
//不断的循环进行消费
while ($channel->is_open()) {
$channel->wait();
}
//关闭连接
$channel->close();
$connection->close();
消费者worker2，代码和worker1一样，同时运行开启后会一起消费
D:\phpstudy_pro\WWW\rabbitmq\worker\worker2.php
消费者消费消息ack确认
用以确认不会丢失消息
消费消息
basic_consume($queue = ‘’, $consumer_tag = ‘’, $no_local = false, $no_ack = false, $exclusive = false, $nowait = false, $callback = null, $ticket = null, $arguments = array())
no_ack=false,设置为手动应答
开启后需要进行消息的消费确认后才会进行移除，否者该消息会一直存在消息队列中
消费端代码
D:\phpstudy_pro\WWW\rabbitmq\worker\worker1.php
<?php require_once "../vendor/autoload.php";
use PhpAmqpLib\Connection\AMQPStreamConnection;
//建立connction
$connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order');
//Channel
$channel = $connection->channel();
//声明队列名为：task_queue
$queue_name = 'task_queue';
$channel->queue_declare($queue_name, false, true, false, false);
echo " [*] Waiting for messages. To exit press CTRL+C\n";
$callback = function ($msg) {
echo 'received = ', $msg->body . "\n";
//确认消息已被消费，从生产队列中移除
$msg->ack();
};
//设置消费成功后才能继续进行下一个消费
$channel->basic_qos(null, 1, null);
//开启消费no_ack=false,设置为手动应答
$channel->basic_consume($queue_name, '', false, false, false, false, $callback);
//不断的循环进行消费
while ($channel->is_open()) {
$channel->wait();
}
//关闭连接
$channel->close();
$connection->close();
fanout模式生产者推送到交换器
发布/订阅模式
是要是公用一个交换机的消费端都能收到同样的消息，类似广播的功能
文档：rabbitmq Publish/Subscribe
https://www.rabbitmq.com/tutorials/tutorial-three-php.html
rabbitmq Exchange类型
交换器、路由键、绑定 Exchange：交换器。发送消息的AMQP实体。交换器拿到一个消息之后将它路由给一个或几个队列。它使用哪种路由算法是由交换机类型和被称作绑定（Binding）的规则所决定的。RabbitMQ有四种类型。 RoutingKey：路由键。生产者将消息发送给交换器。一般会指定一个RoutingKey,用来指定这个消息的路由规则，而这个RoutingKey需要与交换器类型和绑定键（BindingKey）联合使用才能最终失效。 Binding：绑定。绑定（Binding）是交换机（Exchange）将消息（Message）路由给队列（Queue）所需遵循的规则。 # 四种模式 Direct 定向 消息与一个特定的路由键完全匹配 Topic 通配符 路由键和某模式进行匹配 Fanout 广播 发送到该类型交换机的消息都会被广播到与该交换机绑定的所有队列 Headers 不处理路由键，而是根据发送的消息内容中的headers属性进行匹配
exchange_declare($exchange, $type, $passive = false, $durable = false, $auto_delete = true, $internal = false, $nowait = false, $arguments = array(), $ticket = null) 。试探性申请一个交换器，若该交换器不存在，则创建；若存在，则跳过。
生产者代码
D:\phpstudy_pro\WWW\rabbitmq\ps\pro.php
<?php require_once "../vendor/autoload.php"; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; //建立connction $connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order'); //Channel $channel = $connection->channel(); //声明交换器 $exc_name = 'exch'; $channel->exchange_declare($exc_name, 'fanout', false, false, false); //声明数据 $data = 'this is fanout message'; //创建消息 $msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]); //发布消息 $channel->basic_publish($msg, $exc_name); //关闭连接 $channel->close(); $connection->close();
fanout模式消费者消费消息
是要是公用一个交换机的消费端都能收到同样的消息，类似广播的功能
当消费端运行时才会显示该队列
消费端：
D:\phpstudy_pro\WWW\rabbitmq\ps\worker1.php
<?php require_once "../vendor/autoload.php";
use PhpAmqpLib\Connection\AMQPStreamConnection;
//建立connction
$connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order');
//Channel
$channel = $connection->channel();
//声明交换器
$exc_name = 'exch';
$channel->exchange_declare($exc_name, 'fanout', false, false, false);
//获取系统生成的消息队列名称
list($queue_name, ,) = $channel->queue_declare('', false, false, true, false);
//将队列名与交换器名进行绑定
$channel->queue_bind($queue_name,$exc_name);
$callback = function ($msg) {
echo 'received = ', $msg->body . "\n";
//确认消息已被消费，从生产队列中移除
$msg->ack();
};
//设置消费成功后才能继续进行下一个消费
$channel->basic_qos(null, 1, null);
//开启消费no_ack=false,设置为手动应答
$channel->basic_consume($queue_name, '', false, false, false, false, $callback);
//不断的循环进行消费
while ($channel->is_open()) {
$channel->wait();
}
//关闭连接
$channel->close();
$connection->close();
direct模式消息队列使用
文档：
https://www.rabbitmq.com/tutorials/tutorial-four-php.html
用来指定不同的交换机和指定routing_key，在消费端进行消费
生产者代码：
D:\phpstudy_pro\WWW\rabbitmq\routing\pro.php
<?php require_once "../vendor/autoload.php"; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; //建立connction $connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order'); //Channel $channel = $connection->channel(); //声明交换器 $exc_name = 'direct_log'; //指定routing_key $routing_key = 'info'; //指定交换机类型为direct $channel->exchange_declare($exc_name, 'direct', false, false, false); //声明数据 $data = 'this is ' . $routing_key . ' message'; //创建消息 $msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]); //发布消息 //指定使用的routing_key $channel->basic_publish($msg, $exc_name, $routing_key); //关闭连接 $channel->close(); $connection->close();
消费者代码
D:\phpstudy_pro\WWW\rabbitmq\routing\info.php
<?php require_once "../vendor/autoload.php";
use PhpAmqpLib\Connection\AMQPStreamConnection;
//建立connction
$connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order');
//Channel
$channel = $connection->channel();
//声明交换器
$exc_name = 'direct_log';
//指定routing_key
$routing_key = 'info';
$channel->exchange_declare($exc_name, 'direct', false, false, false);
//获取系统生成的消息队列名称
list($queue_name, ,) = $channel->queue_declare('', false, false, true, false);
//将队列名与交换器名进行绑定，并指定routing_key
$channel->queue_bind($queue_name,$exc_name,$routing_key);
$callback = function ($msg) {
echo 'received = ', $msg->body . "\n";
//确认消息已被消费，从生产队列中移除
$msg->ack();
};
//设置消费成功后才能继续进行下一个消费
$channel->basic_qos(null, 1, null);
//开启消费no_ack=false,设置为手动应答
$channel->basic_consume($queue_name, '', false, false, false, false, $callback);
//不断的循环进行消费
while ($channel->is_open()) {
$channel->wait();
}
//关闭连接
$channel->close();
$connection->close();
topic模式消息队列使用
通配符的匹配模式
如消费端中routing_key = ‘user.*’;
生产者：
指定routing_key= ‘user.top’
<?php require_once "../vendor/autoload.php"; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; //建立connction $connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order'); //Channel $channel = $connection->channel(); //声明交换器 $exc_name = 'topic_log'; //指定routing_key $routing_key = 'user.top'; //指定交换机类型为direct $channel->exchange_declare($exc_name, 'topic', false, false, false); //声明数据 $data = 'this is ' . $routing_key . ' message'; //创建消息 $msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]); //发布消息 //指定使用的routing_key $channel->basic_publish($msg, $exc_name, $routing_key); //关闭连接 $channel->close(); $connection->close();
消费者
消费端中routing_key = ‘user.*’;
<?php require_once "../vendor/autoload.php";
use PhpAmqpLib\Connection\AMQPStreamConnection;
//建立connction
$connection = new AMQPStreamConnection('192.168.10.105', 5672, 'root', 'root', 'order');
//Channel
$channel = $connection->channel();
//声明交换器
$exc_name = 'direct_log';
//指定routing_key
$routing_key = 'user.*';
$channel->exchange_declare($exc_name, 'topic', false, false, false);
//获取系统生成的消息队列名称
list($queue_name, ,) = $channel->queue_declare('', false, false, true, false);
//将队列名与交换器名进行绑定，并指定routing_key
$channel->queue_bind($queue_name,$exc_name,$routing_key);
$callback = function ($msg) {
echo 'received = ', $msg->body . "\n";
//确认消息已被消费，从生产队列中移除
$msg->ack();
};
//设置消费成功后才能继续进行下一个消费
$channel->basic_qos(null, 1, null);
//开启消费no_ack=false,设置为手动应答
$channel->basic_consume($queue_name, '', false, false, false, false, $callback);
//不断的循环进行消费
while ($channel->is_open()) {
$channel->wait();
}
//关闭连接
$channel->close();
$connection->close();
推荐学习：《PHP视频教程》
Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!
Articles Liés
Voir plus- Comment supprimer le 5ème élément d'un tableau en php
- Quels sont les outils de création d'environnement PHP ?
- Comment convertir un binaire en nombre à virgule flottante en php
- Comment trouver des éléments uniques dans un tableau en php
- Comment convertir une chaîne en tableau en php (séparés par des espaces)