Detaillierte Erläuterung der PHP-Nachrichtenwarteschlange

Dieser Artikel gibt Ihnen hauptsächlich eine detaillierte Erklärung der PHP-Nachrichtenwarteschlange. Ich hoffe, er kann Ihnen helfen. Lassen Sie uns zunächst verstehen, was eine Nachrichtenwarteschlange ist.

1. Was ist eine Nachrichtenwarteschlange

Nachrichtenwarteschlange (englisch: Message queue) ist eine Methode der prozessübergreifenden Kommunikation oder der Kommunikation zwischen verschiedenen Threads desselben Prozesses

2. Warum Nachrichtenwarteschlangen verwenden

Die Nachrichtenwarteschlangentechnologie ist eine Technologie zum Austausch von Informationen zwischen verteilten Anwendungen. Nachrichtenwarteschlangen können sich im Speicher oder auf der Festplatte befinden und die Warteschlange speichert Nachrichten, bis sie von einer Anwendung gelesen werden. Nachrichtenwarteschlangen ermöglichen es Anwendungen, unabhängig voneinander auszuführen, ohne den Standort der anderen zu kennen oder darauf zu warten, dass das empfangende Programm die Nachricht empfängt, bevor es fortfährt.

3. Wann man die Nachrichtenwarteschlange verwendet

Sie müssen zuerst den Unterschied zwischen Nachrichtenwarteschlange und Remote-Prozeduraufruf herausfinden. Als viele Leser mich befragten, stellte ich fest, dass sie RPC benötigen (. Remote Procedure Call) statt Nachrichtenwarteschlange.

Nachrichtenwarteschlangen können synchron oder asynchron implementiert werden. Normalerweise verwenden wir Nachrichtenwarteschlangen asynchron, und Remote-Prozeduraufrufe verwenden meist synchrone Methoden.

Was ist der Unterschied zwischen MQ und RPC? MQ liefert normalerweise ein unregelmäßiges Protokoll, das vom Benutzer definiert wird und die Speicherung und Weiterleitung implementiert, während RPC normalerweise ein dediziertes Protokoll ist und der aufrufende Prozess Ergebnisse zurückgibt.

4. Wann Sie die Nachrichtenwarteschlange verwenden sollten

Für Synchronisierungsanforderungen ist Remote Procedure Call (PRC) besser geeignet.

Für asynchrone Anforderungen ist die Nachrichtenwarteschlange besser geeignet.

Derzeit unterstützen viele Message-Queue-Software auch RPC-Funktionen, und viele RPC-Systeme können auch asynchron aufgerufen werden.

Nachrichtenwarteschlange wird verwendet, um die folgenden Anforderungen umzusetzen

Speichern und weiterleiten

Verteilte Transaktionen

Veröffentlichen und abonnieren

Inhaltsbasiert Routing

Punkt-zu-Punkt-Verbindung

5. Wer ist für die Verarbeitung der Nachrichtenwarteschlange verantwortlich

Wenn ein kleines Projektteam normalerweise eine Person umsetzen kann, einschließlich Push- und Empfangsverarbeitung von Nachrichten. Wenn das Team groß ist, definieren sie normalerweise das Nachrichtenprotokoll und entwickeln dann jeweils ihre eigenen Teile. Beispielsweise ist ein Team für das Schreiben des Push-Protokollteils und ein anderes Team für das Schreiben des Empfangs- und Verarbeitungsteils verantwortlich.

Warum sprechen wir also nicht über das Framing von Nachrichtenwarteschlangen?

Framework hat mehrere Vorteile:

Entwickler müssen die Nachrichtenwarteschlangenschnittstelle nicht erlernen

Entwickler müssen sich nicht um das Senden und Empfangen von Nachrichten kümmern

Entwickler übergeben Unified API-Push-Nachrichten

Der Fokus der Entwickler liegt auf der Implementierung von Geschäftslogikfunktionen

6. So implementieren Sie das Nachrichtenwarteschlangen-Framework

Im Folgenden wird ein SOA-Framework entwickelt Vom Autor werden drei Schnittstellen bereitgestellt, nämlich SOAP, RESTful und AMQP (RabbitMQ). Sobald Sie die Framework-Idee verstanden haben, können Sie sie problemlos weiter erweitern, z. B. durch das Hinzufügen von Unterstützung für XML-RPC, ZeroMQ usw.

https://github.com/netkiller/SOA

In diesem Artikel geht es nur um den Teil des Nachrichtenwarteschlangen-Frameworks.

6.1. Daemon-Prozess

Das Nachrichtenwarteschlangen-Framework ist eine lokale Anwendung (Befehlszeilenprogramm). Damit es im Hintergrund ausgeführt werden kann, müssen wir einen Daemon-Prozess implementieren.

https://github.com/netkiller/SOA/blob/master/bin/rabbitmq.php

Jede Instanz verarbeitet eine Reihe von Warteschlangen für die Instanziierung. $queueName = 'Warteschlangenname', $exchangeName = 'Exchange-Name', $routeKey = 'route'

$daemon = neues FrameworkRabbitDaemon($queueName = 'email', $exchangeName = 'email', $routeKey = 'email');

Der Daemon-Prozess muss als Root-Benutzer ausgeführt werden. Nach der Ausführung wechselt er zum normalen Benutzer und erstellt eine Prozess-ID-Datei zur Verwendung, wenn der Prozess stoppt.

Daemon-Kerncode https://github.com/netkiller/SOA/blob/master/system/rabbitdaemon.class.php

6.2. Message Queuing Protocol

Das Das Nachrichtenprotokoll ist ein Array, das serialisiert oder in JSON konvertiert und an den Nachrichtenwarteschlangenserver übertragen wird. Hier wird das JSON-Formatprotokoll verwendet.

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

Serialisiertes Protokoll

{"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() {
}
}

相关推荐：

PHP实现消息队列

php实现消息队列类实例分享

什么是消息队列？在Linux中使用消息队列

Das obige ist der detaillierte Inhalt vonDetaillierte Erläuterung der PHP-Nachrichtenwarteschlange. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!