Dieser Artikel bringt Ihnen relevantes Wissen über PHP. Er stellt hauptsächlich die Nachrichtenwarteschlange vor und zeigt einige praktische Details. Ich hoffe, dass dies hilfreich ist Jeder. Hilfreich.
Empfohlenes Lernen: „PHP-Video-Tutorial“
Erklärung
MQ (Message. Queue ) ist eine Nachrichtenwarteschlange, die ist zwischen Anwendungen Kommunikationsmethode, die Nachricht kann sofort nach dem Senden zurückgegeben werden und das Nachrichtensystem stellt die zuverlässige Zustellung der Nachricht sicher. „Nachrichtenwarteschlange“ ist ein Container, der Nachrichten während ihrer Übertragung speichert. Es ist typisch: Produzenten-Konsumenten-Modell. Produzenten produzieren weiterhin Nachrichten in der Nachrichtenwarteschlange und Verbraucher erhalten weiterhin Nachrichten aus der Warteschlange. Da sowohl die Produktion als auch der Konsum von Nachrichten asynchron sind und sich nur um das Senden und Empfangen von Nachrichten kümmert, erfolgt kein Eingriff in die Geschäftslogik, wodurch eine Entkopplung von Produzenten und Konsumenten erreicht wird.
Warum Nachrichten-Middleware verwenden?
Die Nachrichtenwarteschlange ist eine wichtige Komponente in verteilten Systemen. Sie löst Probleme wie Anwendungsentkopplung, asynchrone Nachrichten, Reduzierung von Verkehrsspitzen usw. und erreicht eine hohe Parallelität, hohe Verfügbarkeit, Skalierbarkeit und letztendliche Konsistenzarchitektur.
Asynchrone Verarbeitung
Benutzer Registrierung Nach den Informationen müssen Sie E-Mails senden und Textnachrichten registrieren
1. Nachdem die Benutzerregistrierungsinformationen in die Datenbank geschrieben wurden, werden die Informationen auch bei erfolgreicher Registrierung zurückgegeben
2. Das Senden von E-Mails und die Registrierung von Textnachrichten werden ausgeführt asynchron durch die Nachrichtenwarteschlange, und der Benutzer muss nicht auf diese beiden Vorgänge warten
Anwendungsentkopplung
Nachdem der Benutzer eine Bestellung aufgegeben hat, muss das Bestellsystem das Inventarsystem benachrichtigen. Der traditionelle Ansatz besteht darin, dass das Bestellsystem die Schnittstelle des Inventarsystems aufruft, um den Lagerbestand zu erhöhen oder zu verringern
1. Der Benutzer gibt einen Produktionsauftrag auf und gibt eine Erfolgsmeldung zurück
2. Das Warteschlangenverbrauchsinventarsystem erhöht oder verringert den Lagerbestand 
Verkehrsspitzenreduzierung
Verkehrsspitzenreduzierung ist auch ein häufiges Szenario in Nachrichtenwarteschlangen. Es wird im Allgemeinen häufig bei Flash-Verkäufen oder Gruppenaktivitäten eingesetzt.
1. Wenn eine Gruppe von Benutzern den Eintritt in die Warteschlange anfordert, kontrollieren Sie die Anzahl der Warteschlangen . Wenn die Anzahl eine bestimmte Anzahl überschreitet, endet der Flash-Sale
2. Dann werden die Warteschlangen nacheinander nach dem First-In-First-Out-Prinzip verbraucht
Rabbitmq-Funktionen
Zuverlässigkeit (Zuverlässigkeit) RabbitMQ verwendet einige Mechanismen, um die Zuverlässigkeit sicherzustellen, z. B. Persistenz, Übertragungsbestätigung und Freigabebestätigung.
Flexibles Routing Die Nachricht wird über Exchange weitergeleitet, bevor die Nachricht in die Warteschlange gelangt. Für typische Routing-Funktionen bietet RabbitMQ bereits einige integrierte Exchange-Implementierungen. Für komplexere Routing-Funktionen können mehrere Exchanges miteinander verbunden werden, und auch Ihr eigener Exchange kann über den Plug-in-Mechanismus implementiert werden.
Message Clustering Mehrere RabbitMQ-Server können einen Cluster bilden, um einen logischen Broker zu bilden.
Hochverfügbare Warteschlangen Warteschlangen können auf Maschinen im Cluster gespiegelt werden, sodass die Warteschlangen weiterhin verfügbar sind, wenn einige Knoten ausfallen.
Multiprotokoll RabbitMQ unterstützt mehrere Nachrichtenwarteschlangenprotokolle wie STOMP, MQTT usw.
Viele Clients RabbitMQ unterstützt fast alle gängigen Sprachen, wie PHP Java, .NET, Ruby usw.
Verwaltungsoberfläche RabbitMQ bietet eine benutzerfreundliche Benutzeroberfläche, die es Benutzern ermöglicht, viele Aspekte des Message Brokers zu überwachen und zu verwalten.
Verfolgung Wenn die Nachricht abnormal ist, stellt RabbitMQ einen Nachrichtenverfolgungsmechanismus bereit, damit Benutzer herausfinden können, was passiert ist.
Plugin-System RabbitMQ bietet viele Plug-Ins, um es in vielerlei Hinsicht zu erweitern, und Sie können auch Ihre eigenen Plug-Ins schreiben.
So funktioniert RabbitMQ
Broker: RabbitMQ Server ist eine Anwendung, die Nachrichten empfängt und verteilt und der Nachrichtenbroker ist.
Virtueller Host: Ähnlich wie bei einer MySQL-Datenbank können mehrere Vhosts aufgeteilt werden, wenn mehrere verschiedene Benutzer die vom selben RabbitMQ-Server bereitgestellten Dienste nutzen, und jeder Benutzer erstellt Austausch/Warteschlange usw. in seinem eigenen Vhost.
Verbindung: TCP-Verbindung zwischen Publisher/Consumer und Broker.
Kanal: Wenn bei jedem Zugriff auf RabbitMQ eine Verbindung hergestellt wird, ist der Aufwand für den Aufbau einer TCP-Verbindung enorm und die Effizienz gering, wenn das Nachrichtenvolumen groß ist. Channel ist eine logische Verbindung, die innerhalb der Verbindung hergestellt wird. Als Lightweight-Verbindung reduziert Channel die Kosten für den Aufbau einer TCP-Verbindung durch das Betriebssystem erheblich.
Exchange: Die Nachricht erreicht den ersten Stopp des Brokers. Gemäß den Verteilungsregeln stimmt er mit dem Routing-Schlüssel in der Abfragetabelle überein und verteilt die Nachricht an die Warteschlange. Häufig verwendete Typen sind: Direct (Punkt-zu-Punkt), Topic (Publish-Subscribe) und Fanout (Multicast).
Warteschlange: Die Nachricht wird schließlich hierher gesendet, damit sie vom Verbraucher abgeholt werden kann. Eine Nachricht kann gleichzeitig in mehrere Warteschlangen kopiert werden.
Offizielle RabbitMQ-Adresse: http://www.rabbitmq.com
Um Rabbitmq zu installieren, müssen Sie zuerst Erlang installieren
Schritt 1: Erlang-Installation
Um Rabbitmq zu installieren, müssen Sie Erlang installieren Erstens unterstützt Centos7 die Installation der Erlang 24-Version nicht. 2: Rabbitmq installieren
# 系统 centos 7# 下载erlang包，手动下载后上传至服务器，我在使用wget下载后无法安装，这里没明白 # 安装 yum install erlang-23.3.4.4-1.el7.x86_64.rpm # 验证安装是否成功 erl
PHP-Nachrichtenwarteschlange Rabbitmq wird in verschiedenen Modi verwendet 
Rabbitmq-Verwaltungsschnittstelle und Befehlszeilenverwendung
4369: epmd (Erlang Port Mapper Daemon), Erlang-Service-Port 
15672: HTTP-API-Client, Verwaltung Die Benutzeroberfläche (nur wenn das Verwaltungs-Plugin aktiviert ist) startet nicht unbedingt
25672: Wird für die Kommunikation zwischen Knoten verwendet (Erlang-Verteilungsserver-Port)
rabbitmq-Verwaltungsbefehle 
# 系统 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
Testen Sie den Zugriff auf die UI-Schnittstelle: (Derzeit können sich nicht-lokale Host-Adressen nicht anmelden) 
rabbitmq Konfigurieren Sie die Verwaltungsschnittstelle 
# 启动rabbitmq_management插件 rabbitmq-plugins enable rabbitmq_management # 查看所有插件 rabbitmq-plugins list
# 新增一个用户 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，或其他自定义名称)。
https://github.com/php-amqplib/php- Installation der amqplib-Erweiterung
# 新增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 ".*" ".*" ".*" ".*" ".*" ".*"后边三个.*分别代表：配置权限、写权限、读权限
composer config -g repo.packagist composer https://mirrors.aliyun.com/composer/
Einfachmodus-Erzeuger-Nachrichten-Push an die Nachrichtenwarteschlange
Dokumentation:
https://www.rabbitmq.com/tutorials/tutorial-one-php.html
Einfacher Produzent und Messager 
Produzentencode
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队列可以进入到消息详情
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();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();发布/订阅模式
是要是公用一个交换机的消费端都能收到同样的消息，类似广播的功能
文档：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();文档：
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();通配符的匹配模式
如消费端中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视频教程》
Das obige ist der detaillierte Inhalt vonErste Schritte mit der Nachrichtenwarteschlange RabbitMQ und detaillierte Erläuterung von PHP-Beispielen. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
So öffnen Sie eine PHP-Datei
So entfernen Sie die ersten paar Elemente eines Arrays in PHP
Was tun, wenn die PHP-Deserialisierung fehlschlägt?
So verbinden Sie PHP mit der MSSQL-Datenbank
So verbinden Sie PHP mit der MSSQL-Datenbank
So laden Sie HTML hoch
So lösen Sie verstümmelte Zeichen in PHP
So öffnen Sie PHP-Dateien auf einem Mobiltelefon
