This article brings you relevant knowledge about PHP, which mainly introduces the introduction of message queue RabbitMQ and some practical details. The message queue is a communication method between applications. Let’s talk about it together. Take a look, hope it helps everyone.

Recommended study: "PHP Video Tutorial"

Introduction to message queue and message queue application scenarios

RabbitMQ

Description
MQ (Message Queue) is a message queue, which is a communication method between applications. The message can be returned immediately after it is sent. , the message system ensures reliable delivery of messages. "Message queue" is a container that saves messages during their transmission. It is typical: producer, consumer model. Producers continue to produce messages into the message queue, and consumers continue to obtain messages from the queue. Because the production and consumption of messages are asynchronous, and only care about the sending and receiving of messages, there is no intrusion of business logic, thus achieving the decoupling of producers and consumers.

Why use message middleware?
The message queue is an important component in the distributed system. It solves problems such as application decoupling, asynchronous messaging, and traffic peak shaving, and achieves high concurrency, high availability, scalability, and eventual consistency architecture

Asynchronous processing
Users need to send emails and registration text messages after registering information
1. After the user registration information is written into the database, even if the registration is successful, information about successful registration is returned
2. Sending emails and registration text messages Executed asynchronously through the message queue, the user does not need to wait for these two operations

#Application decoupling
After the user places an order, the order system needs to notify the inventory system. The traditional approach is that the order system calls the interface of the inventory system to increase or decrease inventory
1. The user places an order for production and returns a success prompt
2. The queue consumption inventory system increases or decreases inventory

Traffic peak shaving
Traffic peak shaving is also a common scenario in message queues. It is generally used extensively in flash sales or group grab activities
1. When a group of users Requests come and enter the queue, and the number of queues is controlled. If the number exceeds a certain number, the flash sale will end.
2. Then the queues will be consumed one by one according to the first-in-first-out method

Rabbitmq Features

Reliability RabbitMQ uses some mechanisms to ensure reliability, such as persistence, transmission confirmation, and release confirmation.
Flexible Routing The message is routed through Exchange before the message enters the queue. For typical routing functionality, RabbitMQ already provides some built-in Exchange implementations. For more complex routing functions, multiple Exchanges can be bound together, and your own Exchange can also be implemented through the plug-in mechanism.
Message Clustering Multiple RabbitMQ servers can form a cluster to form a logical Broker.
Highly Available Queues Queues can be mirrored on machines in the cluster, so that the queues are still available if some nodes have problems.
Multi-protocol RabbitMQ supports multiple message queue protocols, such as STOMP, MQTT, etc.
Multi-language clients (Many Clients) RabbitMQ supports almost all commonly used languages, such as PHP Java, .NET, Ruby, etc.
Management UI (Management UI) RabbitMQ provides an easy-to-use user interface that allows users to monitor and manage many aspects of the message Broker.
Tracing mechanism (Tracing) If the message is abnormal, RabbitMQ provides a message tracking mechanism so that users can find out what happened.
Plugin System RabbitMQ provides many plug-ins to expand in many aspects, and you can also write your own plug-ins.

How RabbitMQ works

Broker: An application that receives and distributes messages, RabbitMQ Server is the Message Broker.

Virtual host: Similar to mysql database, when multiple different users use the services provided by the same RabbitMQ server, multiple vhosts can be divided, and each user can use the service provided by the same RabbitMQ server. vhost creates exchange/queue, etc.

Connection: TCP connection between publisher/consumer and broker.

Channel: If a Connection is established every time RabbitMQ is accessed, the overhead of establishing a TCP Connection will be huge and the efficiency will be low when the message volume is large. Channel is a logical connection established inside the connection. As a lightweight Connection, Channel greatly reduces the cost of establishing a TCP connection by the operating system.

Exchange: The message reaches the first stop of the broker. According to the distribution rules, it matches the routing key in the query table and distributes the message to the queue. Commonly used types are: direct (point-to-point), topic (publish-subscribe) and fanout (multicast).

Queue: The message is finally sent here to be picked up by the consumer. A message can be copied to multiple queues at the same time.

rabbitmq installation startup

RabbitMQ official address: http://www.rabbitmq.com
To install rabbitmq, you need to install erlang first

The first step: Erlang installation
To install rabbitmq, you need to install erlang first. Centos7 does not support the installation of erlang 24 version

Download:

##

#?系統(tǒng)??centos?7#?下載erlang包，手動下載后上傳至服務器，我在使用wget下載后無法安裝，這里沒明白


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

#?驗證安裝是否成功
erl

Second Step: Install rabbitmq

#?系統(tǒng)??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

php message queue rabbitmq various Mode usage

rabbitmq management interface and command line usage

4369: epmd (Erlang Port Mapper Daemon), erlang service port

5672: client communication port

15672: HTTP API client, management UI (only if management plugin is enabled) does not necessarily start

25672: Used for inter-node communication (Erlang distribution server port)

rabbitmq Management Command Start 15672: HTTP API client, management UI (only when the management plug-in is enabled)

#?啟動rabbitmq_management插件
rabbitmq-plugins??enable??rabbitmq_management

#?查看所有插件
rabbitmq-plugins??list

Test access to the UI interface: (not at this time The localhost address cannot be logged in)

http://192.168.10.105:15672/

rabbitmq configuration management interface

#?新增一個用戶??
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，或其他自定義名稱)。

Command line to create vhost and PHP extension installation Similar to mysql database, when multiple different users use the services provided by the same RabbitMQ server, multiple vhosts can be divided, and each user creates exchange/queue, etc. in their own vhost. .

1) View vhosts of different users

Create vhost and assign permissions

#?新增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) Install rabbitmq extension installation for php

https://github.com/php-amqplib/php-amqplib Extension installation

Modify Alibaba Cloud Image

composer?config?-g?repo.packagist?composer?https://mirrors.aliyun.com/composer/

Start downloading – Sometimes the lower version of 2.8 will be downloaded here. You need to specify the version

. If the download is unsuccessful, upgrade composer, php.ini, open sockets, expand and switch domestic mirrors

#?升級composer
composer?self-update

#php.ini?打開?sockets?擴展

#下載指定版本
composer?require?php-amqplib/php-amqplib=^3.0

Simple mode producer messages are pushed to the message queue Documentation:
https://www.rabbitmq.com/tutorials/tutorial-one-php.html

Simple producer and messager

Producer code
http://localhost/rabbitmq/simple/pro.php

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

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

//生產(chǎn)者
//Connection: publisher／consumer和broker之間的TCP連接
//Channel: 如果每一次訪問RabbitMQ都建立一個Connection，在消息量大的時候建立TCP Connection的開銷將是巨大的，效率也較低。Channel是在connection內(nèi)部建立的邏輯連接Channel作為輕量級的Connection極大減少了操作系統(tǒng)建立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);

//生產(chǎn)數(shù)據(jù)
$data?=?'this?is?messge';
//創(chuàng)建消息
$msg?=?new?AMQPMessage($data,?['delivery_mode'?=>?AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]);
//發(fā)布消息
$channel->basic_publish($msg,?$exchange?=?'',?$queue_name);
//關閉連接
$channel->close();
$connection->close();

運行生產(chǎn)者腳本：
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(&#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);

//不斷的循環(huán)進行消費
while?($channel->is_open())?{
????$channel->wait();
}

//關閉連接
$channel->close();
$connection->close();

worker模式生產(chǎn)消費消息

rabbitmq Work Queues
一個生產(chǎn)者對應多個消費者，消費特別慢時增加幾個消費分發(fā)

生產(chǎn)者，和上文生產(chǎn)者不變

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

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

//生產(chǎn)者
//Connection: publisher／consumer和broker之間的TCP連接
//Channel: 如果每一次訪問RabbitMQ都建立一個Connection，在消息量大的時候建立TCP Connection的開銷將是巨大的，效率也較低。Channel是在connection內(nèi)部建立的邏輯連接Channel作為輕量級的Connection極大減少了操作系統(tǒng)建立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]);
//發(fā)布消息
????$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(&#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);

//不斷的循環(huán)進行消費
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(&#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";
????//確認消息已被消費，從生產(chǎn)隊列中移除
????$msg->ack();
};

//設置消費成功后才能繼續(xù)進行下一個消費
$channel->basic_qos(null,?1,?null);

//開啟消費no_ack=false,設置為手動應答
$channel->basic_consume($queue_name,?'',?false,?false,?false,?false,?$callback);


//不斷的循環(huán)進行消費
while?($channel->is_open())?{
????$channel->wait();
}

//關閉連接
$channel->close();
$connection->close();

fanout模式生產(chǎn)者推送到交換器

發(fā)布/訂閱模式
是要是公用一個交換機的消費端都能收到同樣的消息，類似廣播的功能

文檔：rabbitmq Publish/Subscribe
https://www.rabbitmq.com/tutorials/tutorial-three-php.html

rabbitmq Exchange類型

交換器、路由鍵、綁定

????Exchange：交換器。發(fā)送消息的AMQP實體。交換器拿到一個消息之后將它路由給一個或幾個隊列。它使用哪種路由算法是由交換機類型和被稱作綁定（Binding）的規(guī)則所決定的。RabbitMQ有四種類型。

????RoutingKey：路由鍵。生產(chǎn)者將消息發(fā)送給交換器。一般會指定一個RoutingKey,用來指定這個消息的路由規(guī)則，而這個RoutingKey需要與交換器類型和綁定鍵（BindingKey）聯(lián)合使用才能最終失效。

????Binding：綁定。綁定（Binding）是交換機（Exchange）將消息（Message）路由給隊列（Queue）所需遵循的規(guī)則。

#?四種模式
Direct??定向?消息與一個特定的路由鍵完全匹配

Topic??通配符?路由鍵和某模式進行匹配

Fanout??廣播?發(fā)送到該類型交換機的消息都會被廣播到與該交換機綁定的所有隊列
Headers?不處理路由鍵，而是根據(jù)發(fā)送的消息內(nèi)容中的headers屬性進行匹配

exchange_declare($exchange, $type, $passive = false, $durable = false, $auto_delete = true, $internal = false, $nowait = false, $arguments = array(), $ticket = null) 。試探性申請一個交換器，若該交換器不存在，則創(chuàng)建；若存在，則跳過。

生產(chǎn)者代碼
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);

//聲明數(shù)據(jù)
$data?=?'this?is?fanout?message';
//創(chuàng)建消息
$msg?=?new?AMQPMessage($data,?['delivery_mode'?=>?AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]);

//發(fā)布消息
$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(&#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);

//獲取系統(tǒng)生成的消息隊列名稱
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";
????//確認消息已被消費，從生產(chǎn)隊列中移除
????$msg->ack();
};

//設置消費成功后才能繼續(xù)進行下一個消費
$channel->basic_qos(null,?1,?null);

//開啟消費no_ack=false,設置為手動應答
$channel->basic_consume($queue_name,?'',?false,?false,?false,?false,?$callback);

//不斷的循環(huán)進行消費
while?($channel->is_open())?{
????$channel->wait();
}

//關閉連接
$channel->close();
$connection->close();

direct模式消息隊列使用

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

用來指定不同的交換機和指定routing_key，在消費端進行消費

生產(chǎn)者代碼：
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);

//聲明數(shù)據(jù)
$data?=?'this?is?'?.?$routing_key?.?'?message';
//創(chuàng)建消息
$msg?=?new?AMQPMessage($data,?['delivery_mode'?=>?AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]);

//發(fā)布消息
//指定使用的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(&#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);

//獲取系統(tǒng)生成的消息隊列名稱
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";
????//確認消息已被消費，從生產(chǎn)隊列中移除
????$msg->ack();
};

//設置消費成功后才能繼續(xù)進行下一個消費
$channel->basic_qos(null,?1,?null);

//開啟消費no_ack=false,設置為手動應答
$channel->basic_consume($queue_name,?'',?false,?false,?false,?false,?$callback);

//不斷的循環(huán)進行消費
while?($channel->is_open())?{
????$channel->wait();
}

//關閉連接
$channel->close();
$connection->close();

topic模式消息隊列使用

通配符的匹配模式

如消費端中routing_key = ‘user.*’;

生產(chǎn)者：
指定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);

//聲明數(shù)據(jù)
$data?=?'this?is?'?.?$routing_key?.?'?message';
//創(chuàng)建消息
$msg?=?new?AMQPMessage($data,?['delivery_mode'?=>?AMQPMessage::DELIVERY_MODE_NON_PERSISTENT]);

//發(fā)布消息
//指定使用的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(&#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);

//獲取系統(tǒng)生成的消息隊列名稱
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";
????//確認消息已被消費，從生產(chǎn)隊列中移除
????$msg->ack();
};

//設置消費成功后才能繼續(xù)進行下一個消費
$channel->basic_qos(null,?1,?null);

//開啟消費no_ack=false,設置為手動應答
$channel->basic_consume($queue_name,?'',?false,?false,?false,?false,?$callback);

//不斷的循環(huán)進行消費
while?($channel->is_open())?{
????$channel->wait();
}

//關閉連接
$channel->close();
$connection->close();

推薦學習：《PHP視頻教程》

The above is the detailed content of Getting started with message queue RabbitMQ and detailed explanation of PHP examples. For more information, please follow other related articles on the PHP Chinese website!