Detailed explanation of PHP message queue

$msg = array(
'Namespace'=>'namespace',
"Class"=>"Email",
"Method"=>"smtp",
"Param" => array(
$mail, $subject, $message, null
)
);

$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) .'');
});

 '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();

$this->queue->consume(function($envelope, $queue) {
$msg = $envelope->getBody();
$result = $this->loader($msg);
$queue->ack($envelope->getDeliveryTag());
});

<?php
namespace framework;
require_once( __DIR__.&#39;/autoload.class.php&#39; );
class RabbitThread extends \Threaded {
private $queue;
public $classspath;
protected $msg;
public function __construct($queue, $logging, $msg) {
$this->classspath = __DIR__.&#39;/../queue&#39;;
$this->msg = $msg;
$this->logging = $logging;
$this->queue = $queue;
}
public function run() {
$speed = microtime(true);
$result = $this->loader($this->msg);
$this->logging->debug(&#39;Result: &#39;. $result.&#39; &#39;);
$this->logging->debug(&#39;Time: &#39;. (microtime(true) - $speed) .&#39;&#39;);
}
// private
public  function loader($msg = null){
$protocol = json_decode($msg,true);
$namespace= $protocol[&#39;Namespace&#39;];
$class = $protocol[&#39;Class&#39;];
$method = $protocol[&#39;Method&#39;];
$param = $protocol[&#39;Param&#39;];
$result = null;
$classspath = $this->classspath.&#39;/&#39;.$this->queue.&#39;/&#39;.$namespace.&#39;/&#39;.strtolower($class)  . &#39;.class.php&#39;;
if( is_file($classspath) ){
require_once($classspath);
//$class = ucfirst(substr($request_uri, strrpos($request_uri, &#39;/&#39;)+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.&#39;->&#39;.$method.&#39;()&#39;);
}else{
$tmp = call_user_func_array(array($obj, $method), $param);
$result = (json_encode($tmp));
$this->logging->info($class.&#39;->&#39;.$method.&#39;("&#39;.implode(&#39;","&#39;, $param).&#39;")&#39;);
}
}else{
$this->logging->error(&#39;Object &#39;. $class. &#39;->&#39; . $method. &#39; is not exist.&#39;);
}
}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 = &#39;&#39;, $exchangeName = &#39;&#39;, $routeKey = &#39;&#39;) {
$this->config = new \framework\Config(&#39;rabbitmq.ini&#39;);
$this->logfile = __DIR__.&#39;/../log/rabbitmq.%s.log&#39;;
$this->logqueue = __DIR__.&#39;/../log/queue.%s.log&#39;;
$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(&#39;pool&#39;)[&#39;thread&#39;]);
}
public function main(){
$connection = new \AMQPConnection($this->config->get(&#39;rabbitmq&#39;));
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(&#39;Protocol: &#39;.$msg.&#39; &#39;);
//$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.&#39;: &#39;.$msg);
}
public function __destruct() {
}
}