Maison > développement back-end > tutoriel php > 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

WBOY
Libérer: 2023-04-11 06:20:01
avant
7944 Les gens l'ont consulté

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.

Premiers pas avec la file d'attente de messages RabbitMQ et explication détaillée des exemples PHP

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
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

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
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

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
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

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
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

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
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP
Télécharger :
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

# 系统  centos 7# 下载erlang包,手动下载后上传至服务器,我在使用wget下载后无法安装,这里没明白


# 安装
yum install erlang-23.3.4.4-1.el7.x86_64.rpm

# 验证安装是否成功
erl
Copier après la connexion

Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

Step 2 : Installez Rabbitmq
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

# 系统  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
Copier après la connexion

Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

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
Copier après la connexion

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,或其他自定义名称)。
Copier après la connexion

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
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

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 ".*" ".*" ".*"
 ".*" ".*" ".*"后边三个.*分别代表:配置权限、写权限、读权限
Copier après la connexion

Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

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/
Copier après la connexion

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

Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

# 升级composer
composer self-update

#php.ini 打开 sockets 扩展

#下载指定版本
composer require php-amqplib/php-amqplib=^3.0
Copier après la connexion

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
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP
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(&#39;192.168.10.105&#39;, 5672, &#39;root&#39;, &#39;root&#39;, &#39;order&#39;);
//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();
Copier après la connexion

运行生产者脚本:
http://localhost/rabbitmq/simple/pro.php
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP
点击goods队列可以进入到消息详情
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

simple模式消费者接受消息

http://localhost/rabbitmq/simple/con.php

<?php require_once "../vendor/autoload.php";

use PhpAmqpLib\Connection\AMQPStreamConnection;

//建立connction
$connection = new AMQPStreamConnection(&#39;192.168.10.105&#39;, 5672, &#39;root&#39;, &#39;root&#39;, &#39;order&#39;);
//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();
Copier après la connexion

worker模式生产消费消息

rabbitmq Work Queues
一个生产者对应多个消费者,消费特别慢时增加几个消费分发
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples 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(&#39;192.168.10.105&#39;, 5672, &#39;root&#39;, &#39;root&#39;, &#39;order&#39;);
//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();
Copier après la connexion

消费者worker1
D:\phpstudy_pro\WWW\rabbitmq\worker\worker1.php

<?php require_once "../vendor/autoload.php";

use PhpAmqpLib\Connection\AMQPStreamConnection;

//建立connction
$connection = new AMQPStreamConnection(&#39;192.168.10.105&#39;, 5672, &#39;root&#39;, &#39;root&#39;, &#39;order&#39;);
//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();
Copier après la connexion

消费者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())
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP
no_ack=false,设置为手动应答
开启后需要进行消息的消费确认后才会进行移除,否者该消息会一直存在消息队列中
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

消费端代码
D:\phpstudy_pro\WWW\rabbitmq\worker\worker1.php

<?php require_once "../vendor/autoload.php";

use PhpAmqpLib\Connection\AMQPStreamConnection;


//建立connction
$connection = new AMQPStreamConnection(&#39;192.168.10.105&#39;, 5672, &#39;root&#39;, &#39;root&#39;, &#39;order&#39;);
//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();
Copier après la connexion

fanout模式生产者推送到交换器

发布/订阅模式
是要是公用一个交换机的消费端都能收到同样的消息,类似广播的功能

文档:rabbitmq Publish/Subscribe
https://www.rabbitmq.com/tutorials/tutorial-three-php.html
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

rabbitmq Exchange类型

交换器、路由键、绑定

    Exchange:交换器。发送消息的AMQP实体。交换器拿到一个消息之后将它路由给一个或几个队列。它使用哪种路由算法是由交换机类型和被称作绑定(Binding)的规则所决定的。RabbitMQ有四种类型。

    RoutingKey:路由键。生产者将消息发送给交换器。一般会指定一个RoutingKey,用来指定这个消息的路由规则,而这个RoutingKey需要与交换器类型和绑定键(BindingKey)联合使用才能最终失效。

    Binding:绑定。绑定(Binding)是交换机(Exchange)将消息(Message)路由给队列(Queue)所需遵循的规则。

# 四种模式
Direct  定向 消息与一个特定的路由键完全匹配

Topic  通配符 路由键和某模式进行匹配

Fanout  广播 发送到该类型交换机的消息都会被广播到与该交换机绑定的所有队列
Headers 不处理路由键,而是根据发送的消息内容中的headers属性进行匹配
Copier après la connexion

exchange_declare($exchange, $type, $passive = false, $durable = false, $auto_delete = true, $internal = false, $nowait = false, $arguments = array(), $ticket = null) 。试探性申请一个交换器,若该交换器不存在,则创建;若存在,则跳过。

Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP
生产者代码
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(&#39;192.168.10.105&#39;, 5672, &#39;root&#39;, &#39;root&#39;, &#39;order&#39;);
//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();
Copier après la connexion

Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

fanout模式消费者消费消息
是要是公用一个交换机的消费端都能收到同样的消息,类似广播的功能

当消费端运行时才会显示该队列
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP
消费端:
D:\phpstudy_pro\WWW\rabbitmq\ps\worker1.php

<?php require_once "../vendor/autoload.php";

use PhpAmqpLib\Connection\AMQPStreamConnection;

//建立connction
$connection = new AMQPStreamConnection(&#39;192.168.10.105&#39;, 5672, &#39;root&#39;, &#39;root&#39;, &#39;order&#39;);
//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();
Copier après la connexion

direct模式消息队列使用

文档:
https://www.rabbitmq.com/tutorials/tutorial-four-php.html

用来指定不同的交换机和指定routing_key,在消费端进行消费
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP
生产者代码:
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(&#39;192.168.10.105&#39;, 5672, &#39;root&#39;, &#39;root&#39;, &#39;order&#39;);
//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();
Copier après la connexion

消费者代码
D:\phpstudy_pro\WWW\rabbitmq\routing\info.php

<?php require_once "../vendor/autoload.php";

use PhpAmqpLib\Connection\AMQPStreamConnection;

//建立connction
$connection = new AMQPStreamConnection(&#39;192.168.10.105&#39;, 5672, &#39;root&#39;, &#39;root&#39;, &#39;order&#39;);
//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();
Copier après la connexion

topic模式消息队列使用

通配符的匹配模式

如消费端中routing_key = ‘user.*’;
Premiers pas avec la file dattente de messages RabbitMQ et explication détaillée des exemples PHP

生产者:
指定routing_key= ‘user.top’

<?php require_once "../vendor/autoload.php";

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

//建立connction
$connection = new AMQPStreamConnection(&#39;192.168.10.105&#39;, 5672, &#39;root&#39;, &#39;root&#39;, &#39;order&#39;);
//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();
Copier après la connexion

消费者
消费端中routing_key = ‘user.*’;

<?php require_once "../vendor/autoload.php";

use PhpAmqpLib\Connection\AMQPStreamConnection;

//建立connction
$connection = new AMQPStreamConnection(&#39;192.168.10.105&#39;, 5672, &#39;root&#39;, &#39;root&#39;, &#39;order&#39;);
//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();
Copier après la connexion

推荐学习:《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!

Étiquettes associées:
php
source:csdn.net
Déclaration de ce site Web
Le contenu de cet article est volontairement contribué par les internautes et les droits d'auteur appartiennent à l'auteur original. Ce site n'assume aucune responsabilité légale correspondante. Si vous trouvez un contenu suspecté de plagiat ou de contrefaçon, veuillez contacter admin@php.cn
Tutoriels populaires
Plus>
Derniers téléchargements
Plus>
effets Web
Code source du site Web
Matériel du site Web
Modèle frontal