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
)
);
Nach dem Login kopieren
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) .'');
});Nach dem Login kopieren
public function loader($msg = null) 负责拆解协议，然后载入对应的类文件，传递参数，运行方法，反馈结果。
Time 可以输出程序运行所花费的时间，对于后期优化十分有用。
提示
loader() 可以进一步优化，使用多线程每次调用loader将任务提交到线程池中，这样便可以多线程处理消息队列。
6.4. 测试
测试代码 https://github.com/netkiller/SOA/blob/master/test/queue/email.php
'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();Nach dem Login kopieren
这里只给出了少量测试与演示程序，如有疑问请到渎者群，或者公众号询问。
7. 多线程
上面消息队列 核心代码如下
$this->queue->consume(function($envelope, $queue) {
$msg = $envelope->getBody();
$result = $this->loader($msg);
$queue->ack($envelope->getDeliveryTag());
});Nach dem Login kopieren
这段代码生产环境使用了半年，发现效率比较低。有些业务场入队非常快，但处理起来所花的时间就比较长，容易出现队列堆积现象。
增加多线程可能更有效利用硬件资源，提高业务处理能力。代码如下
<?php
namespace framework;
require_once( __DIR__.'/autoload.class.php' );
class RabbitThread extends \Threaded {
private $queue;
public $classspath;
protected $msg;
public function __construct($queue, $logging, $msg) {
$this->classspath = __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() {
}
}Nach dem Login kopieren
相关推荐：
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!
