In Teil 1 haben wir die Theorie und einen einfachen Anwendungsfall des AMQP -Protokolls in PHP mit Rabbitmq als Broker behandelt. Lassen Sie uns nun in einige fortgeschrittenere Beispiele eintauchen.

Key Takeaways

• PHP und Rabbitmq verwenden, um Daten asynchron zwischen mehreren Arbeitnehmern zu verarbeiten und die Effizienz in Umgebungen mit hoher Transaktion zu verbessern.
• Implementieren Sie persistente Nachrichten in RabbitMQ, um den Datenverlust w?hrend der Serverabstürze zu verhindern, indem Sie die Nachricht "Delivery_Mode" auf 2 festlegen und Warteschlangen als dauerhaft deklarieren.
• Verwenden Sie die Einstellungen für Qualit?t des Dienstes (QoS) in Rabbitmq, um die Nachrichtenverteilung unter den Arbeitnehmern zu kontrollieren, und stellen Sie sicher, dass kein einzelner Arbeiter überw?ltigt ist.
• Rabbitmq für RPC (Remote Procedure Call) verwenden, indem Nachrichten gesendet werden, die eine Antwort erfordern, die für Aufgaben wie Benutzerauthentifizierung nützlich ist.
• exklusive, tempor?re Warteschlangen für RPC -Antworten einrichten, um sicherzustellen, dass Nachrichten korrekt und sicher zwischen Client und Server gerichtet sind.
• RPC -Antworten effektiv handhaben, indem Sie "correlation_id" in der Antwort mit der Anforderung übereinstimmen, um die korrekte Handhabung und Verarbeitung von Antworten sicherzustellen.

Beispiel 1: Senden Sie die Anforderung, Daten asynchron unter mehreren Arbeitnehmern zu verarbeiten

Im Beispiel des vorherigen Teils hatten wir einen Produzenten, einen Verbraucher. Wenn der Verbraucher starb, würden sich die Nachrichten weiter in der Warteschlange stapeln, bis der Verbraucher erneut begann. Es würde dann alle Nachrichten nacheinander verarbeiten.

Dies kann in einer gleichzeitigen Benutzerumgebung mit einer angemessenen Anzahl von Anfragen pro Minute weniger als ideal sein. Glücklicherweise ist die Skalierung der Verbraucher super einfach, aber lassen Sie uns ein weiteres Beispiel implementieren.

Nehmen wir an, wir haben einen Rechnungsgenerierungsdienst, in dem die Benutzer nur die Rechnungsnummer bereitstellen müssen, und das System generiert automatisch eine PDF -Datei und eine E -Mail an den Benutzer. Das Generieren und Senden der E -Mail kann noch einige Sekunden dauern, wenn der Server, auf dem der Erzeugungsprozess ausgeführt wird, Ressource Limited ist. Nehmen wir nun an, wir müssen mehrere Transaktionen pro Sekunde unterstützen, wie wir dies erreichen, ohne den Server zu überw?ltigen?

Wir müssen das folgende Muster implementieren:

Schauen wir uns unsere Produzentenklasse an:

<span><span><?php
</span></span><span><span>namespace Acme<span>\AmqpWrapper</span>;
</span></span><span>
</span><span><span>use PhpAmqpLib<span>\Connection\AMQPConnection</span>;
</span></span><span><span>use PhpAmqpLib<span>\Message\AMQPMessage</span>;
</span></span><span>
</span><span><span>class WorkerSender
</span></span><span><span>{
</span></span><span>    <span>/* ... SOME OTHER CODE HERE ... */
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * Sends an invoice generation task to the workers
</span></span><span><span>     * 
</span></span><span><span>     * <span>@param <span>int</span> $invoiceNum
</span></span></span><span><span>     */ 
</span></span><span>    <span>public function execute($invoiceNum)
</span></span><span>    <span>{
</span></span><span>        <span>$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
</span></span><span>        <span>$channel = $connection->channel();
</span></span><span>        
</span><span>        <span>$channel->queue_declare(
</span></span><span>            <span>'invoice_queue',    #queue - Queue names may be up to 255 bytes of UTF-8 characters
</span></span><span>            <span>false,              #passive - can use this to check whether an exchange exists without modifying the server state
</span></span><span>            <span>true,               #durable, make sure that RabbitMQ will never lose our queue if a crash occurs - the queue will survive a broker restart
</span></span><span>            <span>false,              #exclusive - used by only one connection and the queue will be deleted when that connection closes
</span></span><span>            <span>false               #auto delete - queue is deleted when last consumer unsubscribes
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->basic_publish(
</span></span><span>            <span>$msg,               #message 
</span></span><span>            <span>'',                 #exchange
</span></span><span>            <span>'invoice_queue'     #routing key (queue)
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->close();
</span></span><span>        <span>$connection->close();
</span></span><span>    <span>}
</span></span><span><span>}</span></span>

Die MOTORELTER :: Execute () erh?lt eine Rechnungsnummer. Als n?chstes erstellen wir wie gewohnt eine Verbindung, Kanal und Warteschlange.

<span><span><?php
</span></span><span><span>/* ... SOME CODE HERE ... */
</span></span><span>
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>
</span><span><span>/* ... SOME CODE HERE ... */</span></span>

Bitte beachten Sie, dass der Konstruktor dieses Mal beim Erstellen des Nachrichtenobjekts einen zweiten Parameter erh?lt: Array ('lieferung_mode' => 2). In diesem Fall m?chten wir angeben, dass die Nachricht nicht verloren gehen sollte, wenn der Rabbitmq -Server abstürzt. Bitte beachten Sie, dass die Warteschlange auch für dauerhaft erkl?rt werden muss.

Der folgende Code kann verwendet werden, um die Formulardaten zu empfangen und den Produzenten auszuführen:

<span><span><?php
</span></span><span><span>namespace Acme<span>\AmqpWrapper</span>;
</span></span><span>
</span><span><span>use PhpAmqpLib<span>\Connection\AMQPConnection</span>;
</span></span><span><span>use PhpAmqpLib<span>\Message\AMQPMessage</span>;
</span></span><span>
</span><span><span>class WorkerSender
</span></span><span><span>{
</span></span><span>    <span>/* ... SOME OTHER CODE HERE ... */
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * Sends an invoice generation task to the workers
</span></span><span><span>     * 
</span></span><span><span>     * <span>@param <span>int</span> $invoiceNum
</span></span></span><span><span>     */ 
</span></span><span>    <span>public function execute($invoiceNum)
</span></span><span>    <span>{
</span></span><span>        <span>$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
</span></span><span>        <span>$channel = $connection->channel();
</span></span><span>        
</span><span>        <span>$channel->queue_declare(
</span></span><span>            <span>'invoice_queue',    #queue - Queue names may be up to 255 bytes of UTF-8 characters
</span></span><span>            <span>false,              #passive - can use this to check whether an exchange exists without modifying the server state
</span></span><span>            <span>true,               #durable, make sure that RabbitMQ will never lose our queue if a crash occurs - the queue will survive a broker restart
</span></span><span>            <span>false,              #exclusive - used by only one connection and the queue will be deleted when that connection closes
</span></span><span>            <span>false               #auto delete - queue is deleted when last consumer unsubscribes
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->basic_publish(
</span></span><span>            <span>$msg,               #message 
</span></span><span>            <span>'',                 #exchange
</span></span><span>            <span>'invoice_queue'     #routing key (queue)
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->close();
</span></span><span>        <span>$connection->close();
</span></span><span>    <span>}
</span></span><span><span>}</span></span>

Bitte verwenden Sie die Eingabeeinstellungen/Validierung, mit denen Sie sich wohl fühlen.

Die Dinge werden auf der Verbraucherseite ein bisschen interessant:

<span><span><?php
</span></span><span><span>/* ... SOME CODE HERE ... */
</span></span><span>
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>
</span><span><span>/* ... SOME CODE HERE ... */</span></span>

Wie gewohnt müssen wir eine Verbindung herstellen, einen Kanal ableiten und eine Warteschlange deklarieren (die Parameter der Warteschlange müssen wie der Produzent gleich sein).

<span><span><?php
</span></span><span><span>chdir(dirname(__DIR__));
</span></span><span><span>require_once('vendor/autoload.php');
</span></span><span>
</span><span><span>use Acme<span>\AmqpWrapper\WorkerSender</span>;
</span></span><span>
</span><span><span>$inputFilters = array(
</span></span><span>    <span>'invoiceNo' => FILTER_SANITIZE_NUMBER_INT,
</span></span><span><span>);
</span></span><span><span>$input = filter_input_array(INPUT_POST, $inputFilters);
</span></span><span><span>$sender = new WorkerSender();
</span></span><span><span>$sender->execute($input['invoiceNo']);</span></span>

Um das Verhalten von Work-Verhaltensweisen (Versandnachrichten unter mehreren Prozessen) zu haben, müssen wir die Parameter der Qualit?t des Dienstes (QoS) mit $ Channel-> Basic_qos () deklarieren:

• Vorabklagegr??e : Keine spezifische Grenze, wir k?nnten so viele Arbeitnehmer haben, wie wir
brauchen
• Bez?hlen Sie die Anzahl : Wie viele Nachrichten k?nnen Sie pro Arbeiter abrufen, bevor Sie eine Best?tigung zurücksenden. Dadurch wird der Arbeiter dazu verarbeitet, 1 Nachricht gleichzeitig zu verarbeiten.
• global : null bedeutet, dass die obigen Einstellungen nur für diesen Verbraucher gelten

Als n?chstes werden wir mit einem wichtigen Unterschied in den Parametern beginnen. Wir werden die automatischen ACKs ausschalten, da wir dem Rabbitmq -Server mitteilen werden, wenn wir die Meldung beendet haben und bereit sind, eine neue zu erhalten.

Wie schicken wir das ACK? Bitte werfen Sie einen Blick auf die METHODERRECEIVER :: Process () (die als Rückrufmethode deklariert wird, wenn eine Nachricht empfangen wird). Die Aufrufe von generierten Methoden fürPdf () und Sendemail () sind nur Dummy -Methoden, die die Zeit simulieren, die für die Erfüllung beider Aufgaben aufgewendet wird. Der $ msg -Parameter enth?lt nicht nur die vom Produzenten gesendete Nutzlast, sondern auch Informationen zu den vom Produzenten verwendeten Objekten. Wir k?nnen Informationen über den vom Produzent verwendeten Kanal mit $ msg-> lieferung_info ['Kanal'] extrahieren (was der gleiche Objekttyp wie der Kanal ist, den wir für den Verbraucher mit $ Connection-> Kanal () ge?ffnet haben;). Da wir dem Kanal des Produzenten eine Best?tigung, dass wir den Prozess abgeschlossen haben ordnungsgem?? zu assoziieren, zu welcher Nachricht der ACK geh?rt.

Wie steuern wir die Arbeiter an? Erstellen Sie einfach eine Datei wie Folgendes und rufen Sie die WorkerReceiver :: listen () auf:

<span><span><?php
</span></span><span><span>namespace Acme<span>\AmqpWrapper</span>;
</span></span><span>
</span><span><span>use PhpAmqpLib<span>\Connection\AMQPConnection</span>;
</span></span><span><span>use PhpAmqpLib<span>\Message\AMQPMessage</span>;
</span></span><span>
</span><span><span>class WorkerReceiver
</span></span><span><span>{
</span></span><span>    <span>/* ... SOME OTHER CODE HERE ... */
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * Process incoming request to generate pdf invoices and send them through 
</span></span><span><span>     * email.
</span></span><span><span>     */ 
</span></span><span>    <span>public function listen()
</span></span><span>    <span>{
</span></span><span>        <span>$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
</span></span><span>        <span>$channel = $connection->channel();
</span></span><span>        
</span><span>        <span>$channel->queue_declare(
</span></span><span>            <span>'invoice_queue',    #queue
</span></span><span>            <span>false,              #passive
</span></span><span>            <span>true,               #durable, make sure that RabbitMQ will never lose our queue if a crash occurs
</span></span><span>            <span>false,              #exclusive - queues may only be accessed by the current connection
</span></span><span>            <span>false               #auto delete - the queue is deleted when all consumers have finished using it
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>/**
</span></span><span><span>         * don't dispatch a new message to a worker until it has processed and 
</span></span><span><span>         * acknowledged the previous one. Instead, it will dispatch it to the 
</span></span><span><span>         * next worker that is not still busy.
</span></span><span><span>         */
</span></span><span>        <span>$channel->basic_qos(
</span></span><span>            <span>null,   #prefetch size - prefetch window size in octets, null meaning "no specific limit"
</span></span><span>            <span>1,      #prefetch count - prefetch window in terms of whole messages
</span></span><span>            <span>null    #global - global=null to mean that the QoS settings should apply per-consumer, global=true to mean that the QoS settings should apply per-channel
</span></span><span>            <span>);
</span></span><span>        
</span><span>        <span>/**
</span></span><span><span>         * indicate interest in consuming messages from a particular queue. When they do 
</span></span><span><span>         * so, we say that they register a consumer or, simply put, subscribe to a queue.
</span></span><span><span>         * Each consumer (subscription) has an identifier called a consumer tag
</span></span><span><span>         */ 
</span></span><span>        <span>$channel->basic_consume(
</span></span><span>            <span>'invoice_queue',        #queue
</span></span><span>            <span>'',                     #consumer tag - Identifier for the consumer, valid within the current channel. just string
</span></span><span>            <span>false,                  #no local - TRUE: the server will not send messages to the connection that published them
</span></span><span>            <span>false,                  #no ack, false - acks turned on, true - off.  send a proper acknowledgment from the worker, once we're done with a task
</span></span><span>            <span>false,                  #exclusive - queues may only be accessed by the current connection
</span></span><span>            <span>false,                  #no wait - TRUE: the server will not respond to the method. The client should not wait for a reply method
</span></span><span>            <span>array($this, 'process') #callback
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>while(count($channel->callbacks)) {
</span></span><span>            <span>$this->log->addInfo('Waiting for incoming messages');
</span></span><span>            <span>$channel->wait();
</span></span><span>        <span>}
</span></span><span>        
</span><span>        <span>$channel->close();
</span></span><span>        <span>$connection->close();
</span></span><span>    <span>}
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * process received request
</span></span><span><span>     * 
</span></span><span><span>     * <span>@param AMQPMessage $msg
</span></span></span><span><span>     */ 
</span></span><span>    <span>public function process(AMQPMessage $msg)
</span></span><span>    <span>{
</span></span><span>        <span>$this->generatePdf()->sendEmail();
</span></span><span>        
</span><span>        <span>/**
</span></span><span><span>         * If a consumer dies without sending an acknowledgement the AMQP broker 
</span></span><span><span>         * will redeliver it to another consumer or, if none are available at the 
</span></span><span><span>         * time, the broker will wait until at least one consumer is registered 
</span></span><span><span>         * for the same queue before attempting redelivery
</span></span><span><span>         */ 
</span></span><span>        <span>$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
</span></span><span>    <span>}
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * Generates invoice's pdf
</span></span><span><span>     * 
</span></span><span><span>     * <span>@return WorkerReceiver
</span></span></span><span><span>     */ 
</span></span><span>    <span>private function generatePdf()
</span></span><span>    <span>{
</span></span><span>        <span>/**
</span></span><span><span>         * Mocking a pdf generation processing time.  This will take between
</span></span><span><span>         * 2 and 5 seconds
</span></span><span><span>         */ 
</span></span><span>        <span>sleep(mt_rand(2, 5));
</span></span><span>        <span>return $this;
</span></span><span>    <span>}
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * Sends email
</span></span><span><span>     * 
</span></span><span><span>     * <span>@return WorkerReceiver
</span></span></span><span><span>     */ 
</span></span><span>    <span>private function sendEmail()
</span></span><span>    <span>{
</span></span><span>        <span>/**
</span></span><span><span>         * Mocking email sending time.  This will take between 1 and 3 seconds
</span></span><span><span>         */ 
</span></span><span>        <span>sleep(mt_rand(1,3));
</span></span><span>        <span>return $this;
</span></span><span>    <span>}
</span></span><span><span>}</span></span>

Verwenden Sie nun den PHP -Befehl (z. B. PHP Worker.php oder welcher Namen, den Sie der oben genannten Datei gegeben haben), um den Arbeiter abzubauen. Aber warte, der Zweck war, zwei oder mehr Arbeiter zu haben, nicht wahr? Kein Problem, starten mehr Arbeitnehmer auf die gleiche Weise, um mehrere Prozesse derselben Datei zu haben, und Rabbitmq registriert die Verbraucher und vertreibt die Arbeiten gem?? den QoS -Parametern.

Beispiel 2: RPC -Anfragen senden und erwarten Sie eine Antwort

Bisher haben wir Nachrichten an den Rabbitmq -Server gesendet, ohne dass der Benutzer auf eine Antwort warten muss. Dies ist in Ordnung für asynchrone Prozesse, die m?glicherweise mehr Zeit in Anspruch nehmen, als der Benutzer bereit ist, nur eine "OK" -Meldung zu sehen. Aber was ist, wenn wir tats?chlich eine Antwort brauchen? Nehmen wir einige Ergebnisse aus einer komplexen Berechnung an, damit wir es dem Benutzer zeigen k?nnen?

Nehmen wir an, wir haben einen zentralisierten Anmelderver (einzelnes Zeichen), der als Authentifizierungsmechanismus aus den übrigen Anwendungen isoliert wird. Die einzige M?glichkeit, diesen Server zu erreichen, ist über Rabbitmq. Wir müssen eine M?glichkeit implementieren, die Anmeldeinformationen an diesen Server zu senden und auf eine Zugriffsantwort zu warten/zu verweigern.

Wir müssen das folgende Muster implementieren:

Schauen wir uns zuerst den Produzenten an:

<span><span><?php
</span></span><span><span>namespace Acme<span>\AmqpWrapper</span>;
</span></span><span>
</span><span><span>use PhpAmqpLib<span>\Connection\AMQPConnection</span>;
</span></span><span><span>use PhpAmqpLib<span>\Message\AMQPMessage</span>;
</span></span><span>
</span><span><span>class WorkerSender
</span></span><span><span>{
</span></span><span>    <span>/* ... SOME OTHER CODE HERE ... */
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * Sends an invoice generation task to the workers
</span></span><span><span>     * 
</span></span><span><span>     * <span>@param <span>int</span> $invoiceNum
</span></span></span><span><span>     */ 
</span></span><span>    <span>public function execute($invoiceNum)
</span></span><span>    <span>{
</span></span><span>        <span>$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
</span></span><span>        <span>$channel = $connection->channel();
</span></span><span>        
</span><span>        <span>$channel->queue_declare(
</span></span><span>            <span>'invoice_queue',    #queue - Queue names may be up to 255 bytes of UTF-8 characters
</span></span><span>            <span>false,              #passive - can use this to check whether an exchange exists without modifying the server state
</span></span><span>            <span>true,               #durable, make sure that RabbitMQ will never lose our queue if a crash occurs - the queue will survive a broker restart
</span></span><span>            <span>false,              #exclusive - used by only one connection and the queue will be deleted when that connection closes
</span></span><span>            <span>false               #auto delete - queue is deleted when last consumer unsubscribes
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->basic_publish(
</span></span><span>            <span>$msg,               #message 
</span></span><span>            <span>'',                 #exchange
</span></span><span>            <span>'invoice_queue'     #routing key (queue)
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->close();
</span></span><span>        <span>$connection->close();
</span></span><span>    <span>}
</span></span><span><span>}</span></span>

Betrachten Sie die Methode von RPCSender :: Execute ausführen, bitte beachten Sie, dass der Parameter von $ condentInicals ein Array in Form von ['Benutzername' => 'x', 'Passwort' => 'y'] ist. Auch hier ?ffnen wir eine neue Verbindung und erstellen wie gewohnt einen Kanal.

<span><span><?php
</span></span><span><span>/* ... SOME CODE HERE ... */
</span></span><span>
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>
</span><span><span>/* ... SOME CODE HERE ... */</span></span>

Der erste Unterschied ergibt sich aus der Erkl?rung einer Warteschlange. Beachten Sie zun?chst, dass wir das List () -Konstrukt verwenden, um das Ergebnis von $ Channel-> queue_declare () zu fangen. Dies liegt daran, dass wir einen Warteschlangennamen nicht explizit senden, w?hrend wir ihn erkl?ren. Deshalb müssen wir herausfinden, wie diese Warteschlange identifiziert wird. Wir interessieren uns nur für das erste Element des Ergebnisarrays, das eine eindeutige Kennung der Warteschlange sein wird (so etwas wie amq.gen-_u0kjvm8HelfZQK9p0z9gg). Die zweite ?nderung ist, dass wir diese Warteschlange als exklusiv deklarieren müssen, sodass die Ergebnisse anderer gleichzeitiger Prozesse nicht vermischt werden.

.

Eine weitere gro?e ?nderung ist, dass der Produzent auch ein Verbraucher einer Warteschlange sein wird, wenn $ Channel-> Basic_consume () beachten, dass wir den Wert $ callback_queue bereitstellen, den wir w?hrend der Deklaration der Warteschlange erhalten haben. Und wie jeder Verbraucher werden wir einen Rückruf erkl?ren, der ausgeführt wird, wenn der Prozess eine Antwort erh?lt.

<span><span><?php
</span></span><span><span>chdir(dirname(__DIR__));
</span></span><span><span>require_once('vendor/autoload.php');
</span></span><span>
</span><span><span>use Acme<span>\AmqpWrapper\WorkerSender</span>;
</span></span><span>
</span><span><span>$inputFilters = array(
</span></span><span>    <span>'invoiceNo' => FILTER_SANITIZE_NUMBER_INT,
</span></span><span><span>);
</span></span><span><span>$input = filter_input_array(INPUT_POST, $inputFilters);
</span></span><span><span>$sender = new WorkerSender();
</span></span><span><span>$sender->execute($input['invoiceNo']);</span></span>

Als n?chstes müssen wir eine Korrelations -ID für die Nachricht erstellen, dies ist nichts weiter als eine eindeutige Bezeichnung für jede Nachricht. Im Beispiel verwenden wir die Ausgabe von Uniqid (), aber Sie k?nnen den Mechanismus verwenden, den Sie bevorzugen (solange er keine Rassenbedingung erzeugt, muss keine starke, kryptosichere RNG sein).

<span><span><?php
</span></span><span><span>namespace Acme<span>\AmqpWrapper</span>;
</span></span><span>
</span><span><span>use PhpAmqpLib<span>\Connection\AMQPConnection</span>;
</span></span><span><span>use PhpAmqpLib<span>\Message\AMQPMessage</span>;
</span></span><span>
</span><span><span>class WorkerReceiver
</span></span><span><span>{
</span></span><span>    <span>/* ... SOME OTHER CODE HERE ... */
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * Process incoming request to generate pdf invoices and send them through 
</span></span><span><span>     * email.
</span></span><span><span>     */ 
</span></span><span>    <span>public function listen()
</span></span><span>    <span>{
</span></span><span>        <span>$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
</span></span><span>        <span>$channel = $connection->channel();
</span></span><span>        
</span><span>        <span>$channel->queue_declare(
</span></span><span>            <span>'invoice_queue',    #queue
</span></span><span>            <span>false,              #passive
</span></span><span>            <span>true,               #durable, make sure that RabbitMQ will never lose our queue if a crash occurs
</span></span><span>            <span>false,              #exclusive - queues may only be accessed by the current connection
</span></span><span>            <span>false               #auto delete - the queue is deleted when all consumers have finished using it
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>/**
</span></span><span><span>         * don't dispatch a new message to a worker until it has processed and 
</span></span><span><span>         * acknowledged the previous one. Instead, it will dispatch it to the 
</span></span><span><span>         * next worker that is not still busy.
</span></span><span><span>         */
</span></span><span>        <span>$channel->basic_qos(
</span></span><span>            <span>null,   #prefetch size - prefetch window size in octets, null meaning "no specific limit"
</span></span><span>            <span>1,      #prefetch count - prefetch window in terms of whole messages
</span></span><span>            <span>null    #global - global=null to mean that the QoS settings should apply per-consumer, global=true to mean that the QoS settings should apply per-channel
</span></span><span>            <span>);
</span></span><span>        
</span><span>        <span>/**
</span></span><span><span>         * indicate interest in consuming messages from a particular queue. When they do 
</span></span><span><span>         * so, we say that they register a consumer or, simply put, subscribe to a queue.
</span></span><span><span>         * Each consumer (subscription) has an identifier called a consumer tag
</span></span><span><span>         */ 
</span></span><span>        <span>$channel->basic_consume(
</span></span><span>            <span>'invoice_queue',        #queue
</span></span><span>            <span>'',                     #consumer tag - Identifier for the consumer, valid within the current channel. just string
</span></span><span>            <span>false,                  #no local - TRUE: the server will not send messages to the connection that published them
</span></span><span>            <span>false,                  #no ack, false - acks turned on, true - off.  send a proper acknowledgment from the worker, once we're done with a task
</span></span><span>            <span>false,                  #exclusive - queues may only be accessed by the current connection
</span></span><span>            <span>false,                  #no wait - TRUE: the server will not respond to the method. The client should not wait for a reply method
</span></span><span>            <span>array($this, 'process') #callback
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>while(count($channel->callbacks)) {
</span></span><span>            <span>$this->log->addInfo('Waiting for incoming messages');
</span></span><span>            <span>$channel->wait();
</span></span><span>        <span>}
</span></span><span>        
</span><span>        <span>$channel->close();
</span></span><span>        <span>$connection->close();
</span></span><span>    <span>}
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * process received request
</span></span><span><span>     * 
</span></span><span><span>     * <span>@param AMQPMessage $msg
</span></span></span><span><span>     */ 
</span></span><span>    <span>public function process(AMQPMessage $msg)
</span></span><span>    <span>{
</span></span><span>        <span>$this->generatePdf()->sendEmail();
</span></span><span>        
</span><span>        <span>/**
</span></span><span><span>         * If a consumer dies without sending an acknowledgement the AMQP broker 
</span></span><span><span>         * will redeliver it to another consumer or, if none are available at the 
</span></span><span><span>         * time, the broker will wait until at least one consumer is registered 
</span></span><span><span>         * for the same queue before attempting redelivery
</span></span><span><span>         */ 
</span></span><span>        <span>$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
</span></span><span>    <span>}
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * Generates invoice's pdf
</span></span><span><span>     * 
</span></span><span><span>     * <span>@return WorkerReceiver
</span></span></span><span><span>     */ 
</span></span><span>    <span>private function generatePdf()
</span></span><span>    <span>{
</span></span><span>        <span>/**
</span></span><span><span>         * Mocking a pdf generation processing time.  This will take between
</span></span><span><span>         * 2 and 5 seconds
</span></span><span><span>         */ 
</span></span><span>        <span>sleep(mt_rand(2, 5));
</span></span><span>        <span>return $this;
</span></span><span>    <span>}
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * Sends email
</span></span><span><span>     * 
</span></span><span><span>     * <span>@return WorkerReceiver
</span></span></span><span><span>     */ 
</span></span><span>    <span>private function sendEmail()
</span></span><span>    <span>{
</span></span><span>        <span>/**
</span></span><span><span>         * Mocking email sending time.  This will take between 1 and 3 seconds
</span></span><span><span>         */ 
</span></span><span>        <span>sleep(mt_rand(1,3));
</span></span><span>        <span>return $this;
</span></span><span>    <span>}
</span></span><span><span>}</span></span>

Lassen Sie uns nun eine Nachricht erstellen, die wichtige ?nderungen im Vergleich zu dem hat, was wir in den ersten beiden Beispielen gewohnt waren. Abgesehen von der Zuweisung einer JSON-kodierten Zeichenfolge, die die Anmeldeinformationen enth?lt, die wir authentifizieren m?chten, müssen wir dem AMQPMessage Constructor ein Array mit zwei definierten Eigenschaften zur Verfügung stellen:

• correlation_id : Ein Tag für die Nachricht
• Reply_to : Die Warteschlangenkennung erzeugt, w?hrend sie es
deklariert hat

Nach der Ver?ffentlichung der Nachricht werden wir die Antwort bewerten, die zu Beginn leer ist. W?hrend der Antwortwert leer bleibt, warten wir auf eine Antwort aus dem Kanal mit $ Channel-> Wait ();.

Sobald wir eine Antwort vom Kanal erhalten haben, wird die Rückrufmethode aufgerufen (RPCSender :: OnResponse ()). Diese Methode entspricht der empfangenen Korrelations -ID gegen die erzeugte. Wenn sie gleich sind, setzen Sie den Antwortk?rper ein, wodurch die While -Schleife gebrochen wird.

Was ist mit dem RPC -Verbraucher? Hier ist es:

<span><span><?php
</span></span><span><span>namespace Acme<span>\AmqpWrapper</span>;
</span></span><span>
</span><span><span>use PhpAmqpLib<span>\Connection\AMQPConnection</span>;
</span></span><span><span>use PhpAmqpLib<span>\Message\AMQPMessage</span>;
</span></span><span>
</span><span><span>class WorkerSender
</span></span><span><span>{
</span></span><span>    <span>/* ... SOME OTHER CODE HERE ... */
</span></span><span>    
</span><span>    <span>/**
</span></span><span><span>     * Sends an invoice generation task to the workers
</span></span><span><span>     * 
</span></span><span><span>     * <span>@param <span>int</span> $invoiceNum
</span></span></span><span><span>     */ 
</span></span><span>    <span>public function execute($invoiceNum)
</span></span><span>    <span>{
</span></span><span>        <span>$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
</span></span><span>        <span>$channel = $connection->channel();
</span></span><span>        
</span><span>        <span>$channel->queue_declare(
</span></span><span>            <span>'invoice_queue',    #queue - Queue names may be up to 255 bytes of UTF-8 characters
</span></span><span>            <span>false,              #passive - can use this to check whether an exchange exists without modifying the server state
</span></span><span>            <span>true,               #durable, make sure that RabbitMQ will never lose our queue if a crash occurs - the queue will survive a broker restart
</span></span><span>            <span>false,              #exclusive - used by only one connection and the queue will be deleted when that connection closes
</span></span><span>            <span>false               #auto delete - queue is deleted when last consumer unsubscribes
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->basic_publish(
</span></span><span>            <span>$msg,               #message 
</span></span><span>            <span>'',                 #exchange
</span></span><span>            <span>'invoice_queue'     #routing key (queue)
</span></span><span>            <span>);
</span></span><span>            
</span><span>        <span>$channel->close();
</span></span><span>        <span>$connection->close();
</span></span><span>    <span>}
</span></span><span><span>}</span></span>

gleiche alte Verbindung und Kanalerstellung :)

Wie bei der Deklaration der Warteschlange hat diese Warteschlange jedoch einen vordefinierten Namen (‘ rpc_queue '). Wir werden die QoS -Parameter definieren, da wir automatische ACKs deaktivieren werden, damit wir mitteilen k?nnen, wenn wir die Anmeldeinformationen überprüfen und ein Ergebnis haben.

<span><span><?php
</span></span><span><span>/* ... SOME CODE HERE ... */
</span></span><span>
</span><span>        <span>$msg = new AMQPMessage(
</span></span><span>            <span>$invoiceNum,
</span></span><span>            <span>array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
</span></span><span>            <span>);
</span></span><span>
</span><span><span>/* ... SOME CODE HERE ... */</span></span>

Die Magie kommt aus dem deklarierten Rückruf. Sobald wir die Anmeldeinformationen authentifiziert haben (ja, ich wei?, dass der Prozess gegen statische Benutzername/Kennwortwerte durchgeführt wird, geht es in diesem Tutorial nicht darum, wie Anmeldeinformationen authentifiziert werden k?nnen;)), sondern die Rückgabenachricht mit derselben Korrelations -ID des Produzenten erstellen erstellt. Wir k?nnen dies aus der Anforderungsnachricht mit $ req-> get ('correlation_id') extrahieren und diesen Wert genauso übergeben wie im Produzenten.

<span><span><?php
</span></span><span><span>chdir(dirname(__DIR__));
</span></span><span><span>require_once('vendor/autoload.php');
</span></span><span>
</span><span><span>use Acme<span>\AmqpWrapper\WorkerSender</span>;
</span></span><span>
</span><span><span>$inputFilters = array(
</span></span><span>    <span>'invoiceNo' => FILTER_SANITIZE_NUMBER_INT,
</span></span><span><span>);
</span></span><span><span>$input = filter_input_array(INPUT_POST, $inputFilters);
</span></span><span><span>$sender = new WorkerSender();
</span></span><span><span>$sender->execute($input['invoiceNo']);</span></span>

Jetzt müssen wir diese Nachricht an die gleiche Warteschlange ver?ffentlichen, die im Produzenten erstellt wurde (die mit dem "zuf?lligen" Namen). Wir extrahieren den Namen der Warteschlange mit $ req-> get ('Reply_to') und verwenden ihn als Routing-Schlüssel in Basic_publish ().

Nachdem wir die Nachricht ver?ffentlicht haben, müssen wir den ACK-Hinweis mit $ req-> lieferung_info ['Kanal']-> basic_ack () an den Liefertag in $ req-> lieferung_info ['lieferung_tag' senden. ] so kann der Produzent aufh?ren zu warten.

starten Sie erneut einen H?rprozess und Sie sind bereit zu gehen. Sie k?nnen sogar Beispiele 2 und 3 kombinieren, um einen Multi-Arbeiter-RPC-Prozess zu haben, um die Authentifizierungsanfragen auszuführen, als nur durch das Aufrühren mehrerer Arbeiter skaliert werden kann.

Es gibt weit mehr zu sagen über Rabbitmq und AMQP, wie virtuelle Hosts, Exchange -Typen, Serververwaltung usw. Sie k?nnen hier und auf der Dokumentationsseite weitere Anwendungsmuster (wie Routing, Themen) finden. Es gibt auch ein Befehlszeilen -Tool zum Verwalten von Rabbitmq sowie eine webbasierte Schnittstelle.

Wenn Ihnen diese Tutorial -Serie gefallen hat und mehr über MQ und reale Anwendungsf?lle erfahren m?chten, teilen Sie uns bitte in den Kommentaren unten mit!

h?ufig gestellte Fragen (FAQs) zu PHP Rabbitmq Erweiterte Beispiele

Welche Rolle spielt Rabbitmq in PHP? Es spielt eine entscheidende Rolle in PHP -Anwendungen, indem sie es erm?glicht, hohe Lasten und komplexe Aufgaben effizienter zu behandeln. RabbitMQ verwendet das erweiterte Message Queuing Protocol (AMQP), um den Austausch von Nachrichten zwischen verschiedenen Teilen einer Anwendung zu erleichtern. Dies erm?glicht die Entkopplung von Prozessen, wodurch die Anwendung skalierbarer und widerstandsf?higer wird.

Wie installiere ich Rabbitmq für PHP? Ihre Maschine. Dies kann über die offizielle Rabbitmq -Website erfolgen. Nach der Installation des Servers k?nnen Sie die PHP AMQP -Erweiterung installieren, die die erforderlichen Funktionen für die Interaktion mit Rabbitmq bietet. Dies kann mit dem PECL -Installationsprogramm mit dem Befehl Pecl install AMQP durchgeführt werden. der AMQPChannel -Klasse. Diese Methode nimmt mehrere Parameter an, einschlie?lich des Namens des Austauschs, des Typs des Austauschs (Direkte, Thema, Fanout oder Header) und optionale Parameter wie Passiv, langlebig, auto_delete und Argumente.

Wie Sende ich eine Nachricht an eine Rabbitmq -Warteschlange in PHP? der AMQPMessage -Klasse mit dem Nachrichteninhalt. Anschlie?end k?nnen Sie die Basic_Publish -Methode der AMQPChannel -Klasse verwenden, um die Nachricht an die Warteschlange zu senden. Die Basic_Publish -Methode nimmt die Nachricht, den Austausch und den Routing -Schlüssel als Parameter an. Rabbitmq -Warteschlange unter Verwendung der Basic_consume -Methode der AMQPChannel -Klasse. Diese Methode enth?lt mehrere Parameter, einschlie?lich des Warteschlangennamens, des Verbraucher -Tags, der NO_Local, der NO_ACK, des exklusiven und einer Rückruffunktion, die ausgeführt wird, wenn eine Nachricht empfangen wird. ? Die PHP AMQP -Erweiterung l?st Ausnahmen der AMQPException -Klasse aus, wenn ein Fehler auftritt. Sie k?nnen diese Ausnahmen erfassen und sie entsprechend den Anforderungen Ihrer Anwendung bew?ltigen. Delivery_Mode -Eigenschaft der AMQPMessage -Klasse auf 2.. Dadurch wird Rabbitmq die Nachricht auf der Festplatte speichern, um sicherzustellen, dass sie nicht verloren geht, selbst wenn der Rabbitmq -Server nicht mehr Abstürze oder Neustart.

Wie kann ich vorrangige Warteschlangen in Rabbitmq mit PHP? Wenn Sie dann eine Nachricht senden, k?nnen Sie die Priorit?tseigenschaft der AMQPMessage -Klasse auf einen Wert zwischen 0 und die maximale Priorit?t festlegen, die Sie angegeben haben. > RabbitMQ kann für den Remote Procedure Call (RPC) in PHP verwendet werden, indem eine Nachricht mit einer Antwort-zu-Eigenschaft an eine Rückrufwarteschlange gesendet wird. Der Server kann dann die Antwort an die Rückrufwarteschlange senden und der Client kann die Antwort von dort aus konsumieren. Das RabbitMQ-Management-Plugin, das eine webbasierte Schnittstelle zur überwachung und Verwaltung Ihres RabbitMQ-Servers bietet. Sie k?nnen auch die Methoden der AMQPChannel -Klasse verwenden, um Informationen über den Zustand des Kanals zu erhalten, z. B. die Anzahl der nicht anerkannten Nachrichten.

Das obige ist der detaillierte Inhalt vonPHP und Rabbitmq: Fortgeschrittene Beispiele. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!