Symfony + RabbitMQクイックスタートフォーヤング

こんにちは、友達。

今日は、SymfonyでRabbitMQを使用する方法についてお話ししたいと思いましたが、水中コームについてはほとんど説明しませんでした。 最後に、完全に戦車に乗っている人のために、ウサギについての興味深い瞬間（ロシア語の「ウサギ」の翻訳）をいくつか書きます。



RabbitMQ自体については説明しませんので、これをまだ知らない場合は、次の翻訳をお読みください。



記事1

セクション2

セクション3

セクション4

セクション5



真珠やpythonの例を恐れてはいけません-これは恐ろしいことではありません。すべてはソースコードから十分に明らかです。



+やがてそれを読んだとき、すべてが十分に詳細に記述されているので、コードを精神的に解釈して、どのように、そしてなぜかを理解するのに十分でした。



消費者が何であるか、そしてその中に$ em-> clear（）+ gc_collect_cyclesを実行し、データベース接続を閉じる必要がある理由をすでに知っている場合、ほとんどの場合、新しいことは学習しません。 この記事は、AMQPプロトコルに対処したくないが、今すぐキューを適用する必要がある人や、何らかの理由でRabbitMQでの選択が気にせず、同じ軽量のbeanstalkdである可能性が高いです。

マイクロサービスアーキテクチャがあり、AMQPを介してコンポーネント間の通信を溶接する方法、RPCを美しくする方法を教えてくれると期待している場合、私自身はこのようなことを非常に長い間待っていました...



RabbitMQを使用してキュー内のメールにメッセージを送信し、フォールトトレランスを提供するタスクがあります：メールサーバーがタイムアウトまたは他の何かで応答した場合、30秒後にタスクを再度実行する必要があります。



したがって、 バンドルをインストールします 。



AppKernelのcomposer requireコマンドと行をコピーする方法を説明するのが面倒です。



あなた自身がこれを実行し、バンドルの構成を開始する準備ができていることを本当に願っています。







バンドル構成：

 old_sound_rabbit_mq: connections: default: host: 'localhost' port: 5672 user: 'guest' password: 'guest' vhost: '/' lazy: false connection_timeout: 3 read_write_timeout: 3 keepalive: false heartbeat: 0 use_socket: true producers: send_email: connection: default exchange_options: { name: 'notification.v1.send_email', type: direct } consumers: send_email: connection: default exchange_options: { name: 'notification.v1.send_email', type: direct } queue_options: { name: 'notification.v1.send_email' } callback: app.consumer.mail_sender
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ここでは、生産者と消費者に大きな注意を払う必要があります。 非常に短く単純な場合：プロデューサーはRabbitMQを介してコンシューマーにメッセージを送信し、コンシューマーはこれらのメッセージを受信して​​処理します。 ここで、exchange_optionsは交換機のオプションです（記事の冒頭にあるrabbitmqに関する記事を読みましたか？）。Queue_optionsはキューのオプションです（同様に）。 コンシューマーのコールバックにも注意を払う価値があります。ConsumerInterface（メッセージ引数を指定した実行メソッド）を拡張するサービスIDは次のとおりです。



なぜなら これまでのところ、あなたはそれを持っていません。アプリケーションを起動するか、コンテナをコンパイルすると、サービスが見つからなかったといういくつかのDI例外を受け取りますが、それを要求します。 それでは、サービスを作成しましょう。

 #app/config/services.yml services: app.consumer.mail_sender: class: AppBundle\Consumer\MailSenderConsumer
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そして、クラス自体：

 namespace AppBundle\Consumer; use OldSound\RabbitMqBundle\RabbitMq\ConsumerInterface; use PhpAmqpLib\Message\AMQPMessage; /** * Class NotificationConsumer */ class MailSenderConsumer implements ConsumerInterface { /** * @var AMQPMessage $msg * @return void */ public function execute(AMQPMessage $msg) { echo '     : '.$msg->getBody().PHP_EOL; echo ' !...'; } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

さて、あなたは私が記事にSwiftMailerを使用する方法を含めなかったことに腹を立てませんでしたか？ :)ここでは、メッセージキューを介して非同期的に行が配信されることが重要です。この行の処理方法はビジネスです。 メールはケースの一例です。



消費者に文字列を渡す方法は？ これを行うには、テストコマンドを作成しましょう。

 namespace AppBundle\Command; use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand; use Symfony\Component\Console\Input\InputInterface; use Symfony\Component\Console\Output\OutputInterface; class TestConsumerCommand extends ContainerAwareCommand { /** * {@inheritdoc} */ protected function configure() { $this ->setName('app:test-consumer') ->setDescription('Hello PhpStorm'); } /** * {@inheritdoc} */ protected function execute(InputInterface $input, OutputInterface $output) { $this->getContainer()->get('old_sound_rabbit_mq.send_email_producer')->publish('    ...'); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

繰り返しますが、私はあなたのために美しい型でコントローラーを作成しなかったことをおizeびします-私はこれが面倒です。 はい、多すぎます。 しかし、私は経済的に悪いので、描くのが好きです。理論とアプリケーションアーキテクチャを少し夢見ています。 気が散る。



次に、コンシューマを起動し、RabbitMQからのメッセージを待つように彼に命令します。

 bin/console rabbitmq:consumer send_email -vvv
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そして、テストチームからメッセージを送信します。

 bin/console app:test-consumer
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そして今、rabbitmq：消費者のプロセスで、私たちのメッセージを見ることができます！ そして、その疑似ディスパッチは成功しました。



次に、エラーが発生した場合に遅延メッセージ処理を実装する方法を見てみましょう。 保留中のメッセージにRabbitMQプラグインを使用しません。 これを実現するには、新しいキューを作成します。このキューでは、メッセージの有効期間を30秒に指定し、次の設定を設定します。死後-メインキューに転送します。



新しいプロデューサーを追加するだけです：

 producers: send_email: connection: default exchange_options: { name: 'notification.v1.send_email', type: direct } delayed_send_email: connection: default exchange_options: name: 'notification.v1.send_email_delayed_30000' type: direct queue_options: name: 'notification.v1.send_email_delayed_30000' arguments: x-message-ttl: ['I', 30000] x-dead-letter-exchange: ['S', 'notification.v1.send_email']
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

それでは、コンシューマのロジックを変更しましょう。

 namespace AppBundle\Consumer; use OldSound\RabbitMqBundle\RabbitMq\ConsumerInterface; use OldSound\RabbitMqBundle\RabbitMq\ProducerInterface; use PhpAmqpLib\Message\AMQPMessage; /** * Class NotificationConsumer */ class MailSenderConsumer implements ConsumerInterface { private $delayedProducer; /** * MailSenderConsumer constructor. * @param ProducerInterface $delayedProducer */ public function __construct(ProducerInterface $delayedProducer) { $this->delayedProducer = $delayedProducer; } /** * @var AMQPMessage $msg * @return void */ public function execute(AMQPMessage $msg) { $body = $msg->getBody(); echo '     '.$body.' ...'.PHP_EOL; try { if ($body == 'bad') { throw new \Exception(); } echo ' ...'.PHP_EOL; } catch (\Exception $exception) { echo 'ERROR'.PHP_EOL; $this->delayedProducer->publish($body); } } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

一般に、出力にLoggerInterfaceを使用すると便利です。また、美しくてスケーラブルです。

しかし、私たちは怠けすぎで、追加の「思考」を作りたくありませんよね？ ただ知っている。



ここで、遅延キューのプロデューサーをスキップする必要があります。

 #app/config/services.yml services: app.consumer.mail_sender: class: AppBundle\Consumer\MailSenderConsumer arguments: ['@old_sound_rabbit_mq.delayed_send_email_producer']
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そして、コマンドを変更します。

 namespace AppBundle\Command; use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand; use Symfony\Component\Console\Input\InputInterface; use Symfony\Component\Console\Output\OutputInterface; class TestConsumerCommand extends ContainerAwareCommand { /** * {@inheritdoc} */ protected function configure() { $this ->setName('app:test-consumer') ->setDescription('Hello PhpStorm'); } /** * {@inheritdoc} */ protected function execute(InputInterface $input, OutputInterface $output) { $this->getContainer()->get('old_sound_rabbit_mq.send_email_producer')->publish(', ...'); $this->getContainer()->get('old_sound_rabbit_mq.send_email_producer')->publish('bad'); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

今、彼女は通常のメッセージとともに、悪いメッセージを送信します。



実行すると、次の出力が表示されます。

      , ...  ...      bad... ERROR
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

30秒後、処理メッセージが再び表示されます。

      bad... ERROR
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そして無限に。 最大試行回数などのロジック 自分で考えてください。 次に、販売といくつかの機能に関するヒントをいくつか紹介します。



今あなたの販売のためのヒント：



1）処理の試行回数を最大にしてトピックから逸​​脱することなく、作業中のコンテキストのすべての例外の102％に注意してください！ 再処理が必要な場合とそうでない場合を想像してください。そうでない場合は、ログからゴミになり、何が起こっているのか理解できません。 RabbitMQで実際のデータの通常のタスクを使用して、beatられたタスクが回転している場合、コンシューマーのコードを更新して再起動することなく、破損したタスクを松葉杖なしで破棄することはできません。 すぐに考え直してください。 この場合、一部のSMTPTimeOutExceptionのみをキャッチするのが正しいでしょう。

また、このようなモデルでは、次のことを理解することが重要です。最初の段階で-1つの「何かの状態を変更するためのグローバルな責任」。 危険なタスクをワーカーに与えすぎないでください。 1Cを使用するオプションを検討する場合、問題は次のようになります：たとえば、製品を1Cに正常にまたは正常に変更しない場合、または製品を1Cに追加する場合、データベースに何かを書き込みます（たとえば、最後に成功した同期の日付または失敗した日付など）。 すなわち ここでは、1Cデータベースとアプリケーションデータベースの2つのデータベースが同時に更新されます。 すべてが1Cで正常に作成された場合、「最後に成功した同期の日付」フィールドがデータベースで更新されたと仮定します-ホップ、エラーがポップアップし、データベースサーバーが応答しません-タスクが「後で」延期され、データベースが応答し始めるまで繰り返されます。 同時に、1Cでのエンティティの作成に関連する「サブタスク」が正常に実行されるたびに、サイトデータベースへの書き込み試行が失敗するたびに、これは間違っています。



2）RabbitMQを使用しているため、耐久性についてお読みください。 PS：バンドル構成、特にexchange_optionsおよびqueue_optionsの「耐久性のある」フラグをtrueまたはfalseとして起動します



3）プログラムが作業を完了した後、あなたの人生はすべてデータベースへの接続を閉じます。 また、EMクリーンアップを実行し、ガベージコレクターの後にリンクをクリーンアップします。 すなわち 最終的に、消費者は次のようになります。

 class MailSenderConsumer implements ConsumerInterface { private $delayedProducer; private $entityManager; /** * MailSenderConsumer constructor. * @param ProducerInterface $delayedProducer * @param EntityManagerInterface $entityManager */ public function __construct(ProducerInterface $delayedProducer, EntityManagerInterface $entityManager) { $this->delayedProducer = $delayedProducer; $this->entityManager = $entityManager; gc_enable(); } /** * @var AMQPMessage $msg * @return void */ public function execute(AMQPMessage $msg) { $body = $msg->getBody(); echo '     '.$body.' ...'.PHP_EOL; try { if ($body == 'bad') { throw new \Exception(); } echo ' ...'.PHP_EOL; } catch (\Exception $exception) { echo 'ERROR'.PHP_EOL; $this->delayedProducer->publish($body); } $this->entityManager->clear(); $this->entityManager->getConnection()->close(); gc_collect_cycles(); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

コンシューマーはデーモンのように動作するため、リンクを絶えず格納し、データベースへの接続を維持するのは悪いことです。 MySQLの場合、MySQLサーバーがなくなってしまいます。



4）保留中のメッセージモデルが予想外にビジネスを殺す可能性がある理由をよく考えてください。 たとえば、管理パネルで製品を変更するときに、これらの変更を1Cのキューに流し込むメカニズムがあります。 ここで状況を想像してください。管理者が製品を変更します->タスク1が作成され、1Cデータベースの同じデータを変更しようとします。 サーバー1Cが応答しないため、すべてが機能するまでタスクは単純にシフトします。 この間、管理者は同じ製品で他の何かを修正することを決定しました。 タスク＃2が記録されます。



ここで、タスク＃1と＃2が1つずつ実行され、延期される状況を想像してください。



タスク2が完了するまでに1Cが動作している場合はどうなりますか？ タスクが完了し、最新の変更があふれます。 次に、タスク＃1が使用され、安定した変更が消去されます:)

終了：タイムスタンプをバージョンとして送信し、タスクが「過去から」の場合、それを破棄します。



5）非同期状態になります-競合状態、さまざまなマシンでの消費者の不一致など、多くのアーキテクチャ上の問題についてお読みください。



6）キューにバージョンを書き込みます...うわー、それが実際の製品でどのように役立つか。 原則として、この例ではそれを行いました。



7）おそらく、RabbitMQとAMQPプロトコル全体は必要ありません。 beanstalkdに注目してください。



8）スーパーバイザーを介してphpでコンシューマーおよびその他すべての悪魔を実行し、そのプロセスのフォールの完全なロギングを接続します。 彼はまた、このすべてのビジネスを管理するためのWebインターフェースを持っています。 常に問題があります。