🌡️ 🥖 👅 Usando RabbitMQ con MonsterMQ Parte 4 🥢 ✅ 👦🏻

En un artículo anterior, escribimos un sistema de registro simple. En el que enviamos mensajes a varios destinatarios a la vez. En este artículo, agregaremos una nueva función que permitirá a los destinatarios recibir solo un subconjunto específico de mensajes. Debido a esto, podemos, por ejemplo, enviar solo mensajes críticos a un archivo de registro, al mismo tiempo que se muestran todos los mensajes enviados en una ventana de terminal.

Enlaces entre la cola y el intercambiador

En el artículo anterior, aprendimos cómo conectar una cola con un intercambiador. Sobre este código:

$producer->queue('queue-1')->bind('logs');

Además, el método de enlace puede aceptar una segunda clave de enrutamiento de argumento (clave de enrutamiento). Para no confundirlo con la clave de enrutamiento, que adopta el método Producer :: Publish () como segundo parámetro, llamémosla clave de enlace (clave de enlace). Así es como podemos vincular el intercambiador con la cola usando la tecla de vinculación:

$producer->queue('queue-2')->bind('logs', 'binding-key');

El valor de la clave de enlace depende del tipo de intercambiador. El intercambiador fanout de la lección anterior simplemente ignora este valor.

Intercambiador directo

Nuestro sistema de registro del último artículo envía todos los mensajes a todos los destinatarios.

Cambiémoslo ligeramente para que filtre los mensajes según el nivel de gravedad. Por ejemplo, para que solo escriba mensajes críticos en el archivo de registro y no desperdicie espacio en disco en mensajes de advertencia e información.

Fanout intercambiador no nos da esa oportunidad.

Pero podemos usar un intercambiador directo que envía mensajes solo a aquellas colas cuya clave de enlace coincide con la clave de enrutamiento del mensaje enviado (el segundo argumento del método Publish ()). Este mecanismo de acción se ilustra en la siguiente imagen:

imagen

(imagen tomada del sitio oficial de RabbitMQ )

Esta imagen muestra el intercambiador Xde tipo directo y dos colas, una con una tecla de enlace naranja y la segunda con dos teclas de enlace negras y verdes .

En esta instalación, los mensajes con la clave de enrutamiento naranja se enviarán primero, y los mensajes con las teclas de enrutamiento negras o verdes se enviarán a la segunda. Cualquier otro mensaje será descartado.

Enlace múltiple

(imagen tomada del sitio web oficial de RabbitMQ )

Es perfectamente aceptable vincular el intercambiador con varias colas con una clave de vinculación. En nuestro ejemplo, podemos agregar una conexión entre el intercambiador X y la cola Q1 usando la tecla de enlace negra . En este caso, nuestro intercambiador de tipo directo se comportará como un intercambiador de tipo fanout, enviará mensajes a todas las colas con la clave de enlace correspondiente. Se enviará un mensaje con la clave de enrutamiento negra a Q1 y Q2 .

Lanzamiento de mensaje

Utilizaremos el modelo descrito anteriormente para nuestro sistema de registro. En lugar de enviar mensajes al intercambiador fanout, los enviaremos al intercambiador directo. Indicaremos el nivel de gravedad del mensaje como clave de enrutamiento. Por lo tanto, el script del destinatario podrá seleccionar mensajes del nivel de gravedad deseado. Pero primero, centrémonos en lanzar mensajes. Como siempre, para esto primero necesitamos crear un intercambiador.


$producer->newDirectExchange('my-logs');

Ahora podemos enviar mensajes:


$producer->newDirectExchange('my-logs');
$producer->publish($message, $severity, 'my-logs');

Para no complicar las cosas, supongamos que el nivel de gravedad ($ severidad) solo puede ser 'información', 'advertencia', 'error'.

Suscripción

Como en la parte anterior, crearemos dos trabajadores que anunciarán las colas y las asociarán con el intercambiador de la siguiente manera:


// Worker 1
foreach ($severities as $severity) {
   $consumer->queue('queue-1')->bind('my-logs', $severity);
}

Poniendolo todo junto

(imagen tomada del sitio web oficial de RabbitMQ )

código send.php :


try {
   $producer = \MonsterMQ\Client\Producer();

   $producer->connect('127.0.0.1', 5672);
   $producer->logIn('guest', 'guest');

   $producer->newDirectExchange('my-logs');

   $severity = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'info';

   $message = implode(' ', array_slice($argv, 2));
   $message = empty($message) ? "Hello World!" : $message;

   $producer->publish($message, $severity, 'my-logs');

   echo "\n Sent {$message} \n";
} catch(\Exception $e) {
   var_dump($e);
}

Código trabajador-1.php :


try {
   $consumer = \MonsterMQ\Client\Consumer();

   $consumer->connect('127.0.0.1', 5672);
   $consumer->logIn('guest', 'guest');

   $producer->queue('queue-1')->setExclusive()->declare();

   $severities = array_slice($argv, 1);
   if (empty($severities)) {
      file_put_contents('php://stderr', "Usage: $argv[0] [info] [warning] [error]\n");
      exit(1);
   }

   foreach ($severities as $severity) {
      $producer->queue('queue-1')->bind('my-logs', $severity);
   }

   $consumer->consume('queue-1');

  echo " \n Waiting for logs. To exit press CTRL+C\n";

   $consumer->wait(function ($message, $channelNumber) use ($consumer){
      echo "\n $message \n";
   });
} catch(\Exception $e) {
   var_dump($e);
}

El código de trabajador-2.php diferirá de trabajador-1.php solo en el nombre de la cola asociada con el intercambiador y pasado al método consume () . Reemplace 'queue-1' con 'queue-2'.

Ahora, si desea que el primer trabajador guarde los mensajes del nivel de 'error' y 'advertencia' en un archivo, ejecute el siguiente comando en el terminal:

php worker-1.php warning error > logs_from_rabbit.log

Y si desea que el segundo trabajador muestre todos los mensajes en la ventana del terminal, haga lo siguiente:

php worker-2.php info warning error
# => Waiting for logs. To exit press CTRL+C

Y para emitir un mensaje de error, escriba lo siguiente:

php send.php error "Run. Run. Or it will explode."

Eso es todo. En la siguiente parte, aprenderemos cómo recibir mensajes basados en un patrón específico.