Home > Article > Backend Development > Detailed explanation of PHP message queue
Detailed explanation of PHP message queue
- 小云云Original
- 2018-03-27 09:29:463954browse
This article mainly shares with you a detailed explanation of PHP message queue. I hope it can help you. First, let’s understand what a message queue is.
1. What is a message queue
Message queue (English: Message queue) is a method of inter-process communication or communication between different threads of the same process
2. Why use message queue
Message queue technology is a technology for exchanging information between distributed applications. Message queues can reside in memory or on disk, and the queue stores messages until they are read by the application. Message queues allow applications to execute independently without knowing each other's locations or waiting for the receiving program to receive the message before continuing.
3. When to use message queue
You first need to figure out the difference between message queue and remote procedure call. When many readers consulted me, I found that what they need is RPC( Remote Procedure Call) instead of message queue.
Message queues can be implemented synchronously or asynchronously. Usually we use message queues asynchronously, and remote procedure calls mostly use synchronous methods.
What is the difference between MQ and RPC? MQ usually delivers an irregular protocol, which is defined by the user and implements store and forwarding; while RPC is usually a dedicated protocol, and the calling process returns results.
4. When to use message queue
For synchronization needs, remote procedure call (PRC) is more suitable for you.
For asynchronous needs, message queue is more suitable for you.
Currently many message queue software also supports RPC functions, and many RPC systems can also be called asynchronously.
Message queue is used to implement the following requirements
Store and forward
Distributed transactions
Publish and subscribe
Content-based routing
Point-to-point connection
5. Who is responsible for processing the message queue
The usual practice is that if a small project team can have one person implement it, including message push and receive processing. If the team is large, they usually define the message protocol and then each develop their own parts. For example, one team is responsible for writing the push protocol part, and another team is responsible for writing the receiving and processing part.
So why don’t we talk about message queue framing?
Framing has several benefits:
Developers do not need to learn the message queue interface
Developers do not need to care about message push and reception
Developers pass Unified API push messages
The focus of developers is to implement business logic functions
6. How to implement the message queue framework
The following is an SOA framework developed by the author. The framework Three interfaces are provided, namely SOAP, RESTful, and AMQP (RabbitMQ). Once you understand the framework idea, you can easily expand it further, such as adding support for XML-RPC, ZeroMQ, etc.
https://github.com/netkiller/SOA
This article only talks about the message queue framework part.
6.1. Daemon process
The message queue framework is a local application (command line program). In order to let it run in the background, we need to implement a daemon process.
https://github.com/netkiller/SOA/blob/master/bin/rabbitmq.php
Each instance handles a set of queues. Three parameters need to be provided for instantiation. $queueName = 'queue name', $exchangeName = 'exchange name', $routeKey = 'route'
$daemon = new \framework\RabbitDaemon($queueName = 'email', $exchangeName = 'email' , $routeKey = 'email');
The daemon process needs to be run as the root user. After running, it will switch to the ordinary user and create a process ID file for use when the process stops.
Daemon core code https://github.com/netkiller/SOA/blob/master/system/rabbitdaemon.class.php
6.2. Message queue protocol
The message protocol is an array. The array is serialized or converted into JSON and pushed to the message queue server. The json format protocol is used here.
$msg = array( 'Namespace'=>'namespace', "Class"=>"Email", "Method"=>"smtp", "Param" => array( $mail, $subject, $message, null ) );
Serialized protocol
{"Namespace":"single","Class":"Email","Method":"smtp","Param":["netkiller@msn.com","Hello"," TestHelloWorld",null]}
使用json格式是考虑到通用性，这样推送端可以使用任何语言。如果不考虑兼容，建议使用二进制序列化，例如msgpack效率更好。
6.3. 消息队列处理
消息队列处理核心代码
https://github.com/netkiller/SOA/blob/master/system/rabbitmq.class.php
所以消息的处理在下面一段代码中进行
$this->queue->consume(function($envelope, $queue) {
$speed = microtime(true);
$msg = $envelope->getBody();
$result = $this->loader($msg);
$queue->ack($envelope->getDeliveryTag()); //手动发送ACK应答
//$this->logging->info(''.$msg.' '.$result)
$this->logging->debug('Protocol: '.$msg.' ');
$this->logging->debug('Result: '. $result.' ');
$this->logging->debug('Time: '. (microtime(true) - $speed) .'');
});public function loader($msg = null) 负责拆解协议，然后载入对应的类文件，传递参数，运行方法，反馈结果。
Time 可以输出程序运行所花费的时间，对于后期优化十分有用。
提示
loader() 可以进一步优化，使用多线程每次调用loader将任务提交到线程池中，这样便可以多线程处理消息队列。
6.4. 测试
测试代码 https://github.com/netkiller/SOA/blob/master/test/queue/email.php
d59d185a8fe5ed4100dde982382642ee '192.168.4.1',
'port' => '5672',
'vhost' => '/',
'login' => 'guest',
'password' => 'guest'
));
$connection->connect() or die("Cannot connect to the broker!\n");
$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$queue = new AMQPQueue($channel);
$queue->setName($queueName);
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue();
$msg = array(
'Namespace'=>'namespace',
"Class"=>"Email",
"Method"=>"smtp",
"Param" => array(
$mail, $subject, $message, null
)
);
$exchange->publish(json_encode($msg), $routeKey);
printf("[x] Sent %s \r\n", json_encode($msg));
$connection->disconnect();
这里只给出了少量测试与演示程序，如有疑问请到渎者群，或者公众号询问。
7. 多线程
上面消息队列 核心代码如下
$this->queue->consume(function($envelope, $queue) {
$msg = $envelope->getBody();
$result = $this->loader($msg);
$queue->ack($envelope->getDeliveryTag());
});
这段代码生产环境使用了半年，发现效率比较低。有些业务场入队非常快，但处理起来所花的时间就比较长，容易出现队列堆积现象。
增加多线程可能更有效利用硬件资源，提高业务处理能力。代码如下
d973597657c05dfc9f24b5259e680b04classspath = __DIR__.'/../queue';
$this->msg = $msg;
$this->logging = $logging;
$this->queue = $queue;
}
public function run() {
$speed = microtime(true);
$result = $this->loader($this->msg);
$this->logging->debug('Result: '. $result.' ');
$this->logging->debug('Time: '. (microtime(true) - $speed) .'');
}
// private
public function loader($msg = null){
$protocol = json_decode($msg,true);
$namespace= $protocol['Namespace'];
$class = $protocol['Class'];
$method = $protocol['Method'];
$param = $protocol['Param'];
$result = null;
$classspath = $this->classspath.'/'.$this->queue.'/'.$namespace.'/'.strtolower($class) . '.class.php';
if( is_file($classspath) ){
require_once($classspath);
//$class = ucfirst(substr($request_uri, strrpos($request_uri, '/')+1));
if (class_exists($class)) {
if(method_exists($class, $method)){
$obj = new $class;
if (!$param){
$tmp = $obj->$method();
$result = json_encode($tmp);
$this->logging->info($class.'->'.$method.'()');
}else{
$tmp = call_user_func_array(array($obj, $method), $param);
$result = (json_encode($tmp));
$this->logging->info($class.'->'.$method.'("'.implode('","', $param).'")');
}
}else{
$this->logging->error('Object '. $class. '->' . $method. ' is not exist.');
}
}else{
$msg = sprintf("Object is not exist. (%s)", $class);
$this->logging->error($msg);
}
}else{
$msg = sprintf("Cannot loading interface! (%s)", $classspath);
$this->logging->error($msg);
}
return $result;
}
}
class RabbitMQ {
const loop = 10;
protected $queue;
protected $pool;
public function __construct($queueName = '', $exchangeName = '', $routeKey = '') {
$this->config = new \framework\Config('rabbitmq.ini');
$this->logfile = __DIR__.'/../log/rabbitmq.%s.log';
$this->logqueue = __DIR__.'/../log/queue.%s.log';
$this->logging = new \framework\log\Logging($this->logfile, $debug=true);
//.H:i:s
$this->queueName= $queueName;
$this->exchangeName= $exchangeName;
$this->routeKey= $routeKey;
$this->pool = new \Pool($this->config->get('pool')['thread']);
}
public function main(){
$connection = new \AMQPConnection($this->config->get('rabbitmq'));
try {
$connection->connect();
if (!$connection->isConnected()) {
$this->logging->exception("Cannot connect to the broker!"
.PHP_EOL);
}
$this->channel = new \AMQPChannel($connection);
$this->exchange = new \AMQPExchange($this->channel);
$this->exchange->setName($this->exchangeName);
$this->exchange->setType(AMQP_EX_TYPE_DIRECT); //direct类型
$this->exchange->setFlags(AMQP_DURABLE); //持久�?
$this->exchange->declareExchange();
$this->queue = new \AMQPQueue($this->channel);
$this->queue->setName($this->queueName);
$this->queue->setFlags(AMQP_DURABLE); //持久�?
$this->queue->declareQueue();
$this->queue->bind($this->exchangeName, $this->routeKey);
$this->queue->consume(function($envelope, $queue) {
$msg = $envelope->getBody();
$this->logging->debug('Protocol: '.$msg.' ');
//$result = $this->loader($msg);
$this->pool->submit(new RabbitThread($this->queueName,
new \framework\log\Logging($this->logqueue, $debug=true), $msg));
$queue->ack($envelope->getDeliveryTag());
});
$this->channel->qos(0,1);
}
catch(\AMQPConnectionException $e){
$this->logging->exception($e->__toString());
}
catch(\Exception $e){
$this->logging->exception($e->__toString());
$connection->disconnect();
$this->pool->shutdown();
}
}
private function fault($tag, $msg){
$this->logging->exception($msg);
throw new \Exception($tag.': '.$msg);
}
public function __destruct() {
}
}
相关推荐：
The above is the detailed content of Detailed explanation of PHP message queue. For more information, please follow other related articles on the PHP Chinese website!