Comment utiliser le package de composants RabbitMQ simple et fiable en php

Lorsque Rabbitmq a été largement utilisé dans le projet, voici un simple résumé des fonctions régulières de Rabbitmq, et encapsulé dans un package composer, l'adresse du package composer, l'adresse github sont les bienvenues. En raison du niveau limité, les bugs sont inévitables. . Bienvenue Donnez de précieux conseils.

Introduction au package easy-rabbitmq

Une encapsulation secondaire du package php-amqplib/php-amqplib, fournissant un ensemble de solutions de production prêtes à l'emploi pour les fonctions courantes. La liste des fonctions actuellement prises en charge est la suivante :

• Messages push vers les commutateurs directs (y compris les messages retardés)

• Messages push vers les commutateurs de secteur (y compris les messages retardés)

• Messages push vers les commutateurs de sujet (y compris les messages retardés) messages)

• Consommation fiable sous le modèle d'abonnement, les consommateurs tenteront de continuer à consommer après un échec, jusqu'à 5 tentatives.

• Consommation fiable en mode pull, les consommateurs tenteront de continuer à consommer après un échec, jusqu'à 5 tentatives.

S'il y a d'autres scènes, veuillez continuer à en ajouter et à répéter plus tard ! !

Exigences

Les exigences de la version PHP du package d'installation dépendent principalement des exigences du package php-amqplib/php-amqplib lui-même. Afin de prendre en compte les utilisateurs de php5.0, nous utilisons le php-amqplib/. Paquet php-amqplib V2.

Veuillez vous référer ici pour les exigences spécifiques.

Cependant, l'auteur recommande d'utiliser php7.0 et supérieur. Ce package de développement est également développé sur la version 7.0 !

Installation

      composer require maweibinguo/easyrabbitmq

Utilisation

Ici, nous recommandons la combinaison script php + superviseur pour assurer la fiabilité du processus de consommation et améliorer la capacité de consommation des travailleurs ! Si vous n'avez pas entendu parler de superviseur, vous pouvez cliquer ici pour en savoir plus

1. Messages push

1-1 Messages push vers les commutateurs directement connectés

      $config = [
          "host" => "127.0.0.1",
            "port" => "5672",
            "user" => "guest",
            "password" => "guest",
            "vhost" => "/",
            "channel_max_num" => 10,
      ];    
      $instance = RabbitMq::getInstance($config);
      
      //延迟消息,30 秒中后才会到达指定的交换机
      $instance->pushToDirect(
                        $msg = time(), //消息体内容
                        $exchange = "easy_direct_exchange", //交换机名称
                        $routingKey = "direct_test_queue", //消息的routingKey，consume(get) 方法到bingdingKey 要和routingKey保持一致
                        $delaySec = 30 //延迟秒数
      );

      //无延迟，推入到指定到直链交换机
      $instance->pushToDirect(
                        $msg = time(), //消息体内容
                        $exchange = "easy_direct_exchange", //交换机名称
                        $routingKey = "direct_test_queue", //消息的routingKey，consume(get) 方法到bingdingKey 要和routingKey保持一致
      );

1-2.

1-3, envoyez des messages vers l'échange de sujets

      $config = [
          "host" => "127.0.0.1",
            "port" => "5672",
            "user" => "guest",
            "password" => "guest",
            "vhost" => "/",
            "channel_max_num" => 10,
      ];    
      $instance = RabbitMq::getInstance($config);
      
      //延迟消息,30 秒中后才会到达指定的交换机
      $instance->pushToFanout(
                        $msg = time(), //消息体内容
                        $exchange = "easy_fanout_exchange", //交换机名称
                        $delaySec = 30 //延迟秒数
      );

      //无延迟，推入到指定到直链交换机
      $instance->pushToFanout(
                        $msg = time(), //消息体内容
                        $exchange = "easy_fanout_exchange" //交换机名称
      );

2. Consommer des messages

La consommation prend en charge les nouvelles tentatives automatiques, avec un maximum de 5 tentatives après chaque échec de consommation, le message sera remis dans la file d'attente de consommation. Le temps de nouvelle tentative passera progressivement à mesure que le nombre d'échecs augmente. La stratégie de retard prise en charge par ce client est la suivante :

Échec une fois (sera à nouveau délivré après 1 seconde), échec 2 fois (sera à nouveau délivré après 2 secondes) Livraison), échoué 3 fois (sera à nouveau livré après 4 secondes), échoué 4 fois (sera à nouveau livré après 8 secondes), échoué 5 fois (sera à nouveau livré après 16 secondes)

2-1, Mode abonnement

Consommation fiable en mode abonnement

      $config = [
          "host" => "127.0.0.1",
            "port" => "5672",
            "user" => "guest",
            "password" => "guest",
            "vhost" => "/",
            "channel_max_num" => 10,
      ];    
      $instance = RabbitMq::getInstance($config);
      
      //延迟消息,30 秒中后才会到达指定的交换机
      $instance->pushToTopic(
                        $msg = time(), //消息体内容
                        $exchange = "easy_topic_exchange", //交换机名称
                        /**
                         * routingKey 要同consum(get)方法的bindingKey相匹配
                         * bindingKey支持两种特殊的字符"*"、“#”，用作模糊匹配, 其中"*"用于匹配一个单词、“#”用于匹配多个单词(也可以是0个)
                         * 无论是bindingKey还是routingKey, 被"."分隔开的每一段独立的字符串就是一个单词, easy.topic.queue, 包含三个单词easy、topic、queue
                         */
                        $routingKey = "easy.topic.queue",
                        $delaySec = 30 //延迟秒数
      );

      //无延迟，推入到指定到直链交换机
      $instance->pushToTopic(
                        $msg = time(), //消息体内容
                        $exchange = "easy_topic_exchange", //交换机名称
                        $routingKey = "easy.topic.queue"    
      );

2-2, mode pull

Consommation fiable en mode pull

      $config = [
          "host" => "127.0.0.1",
            "port" => "5672",
            "user" => "guest",
            "password" => "guest",
            "vhost" => "/",
            "channel_max_num" => 10,
      ];    
      $instance = RabbitMq::getInstance($config);
      $instance->consume(
            $queueName = "direct_test_queue",//订阅的队列名称
            $consumerTag = "c1",//消费标记
            $exchange = "easy_direct_exchange",//交换机名称
            $bindingKey = "direct_test_queue",//bindingkey，如果是直链交换机需要同routingKey保持一致
            $callback = function($msg){
                $body = $msg->body;
                file_put_contents("./test.log", "time => " . time() . "\t" . " body => " . $body . PHP_EOL , FILE_APPEND);
                //如果返回结果不绝对等于(===)true,那么将触发消息重试机制
                return false;
            },
            //5次消费消费失败后，失败消息将会投递到的队列名称
            $failedQueue = "easymq@failed"
      );

Apprentissage recommandé :

Tutoriel vidéo php

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!