SymfonyのMessengerコンポーネントを使ってメール送信の非同期処理をEC-CUBE4.1で実装する方法

この記事はSymfonyのMessengerコンポーネントを使ってメール送信の非同期処理をEC-CUBE4.1で実装する方法を調べていたのでその備忘録です。

Messengerコンポーネントの概念を正しく理解していない可能性がありますので間違いがありましたらご指摘いただけると嬉しいです。

はじめに

以下のサンプルコードはSymfonyのドキュメントを参考にして書いています。詳細はドキュメントを参照してください。

Messengerコンポーネントをインストール

まずはMessengerコンポーネントをインストールします。

composer require symfony/messenger

インストールが完了すると、app/config/eccube/packagesディレクトリにmessenger.yamlが生成されます。

framework:
    messenger:
        # Uncomment this (and the failed transport below) to send failed messages to this transport for later handling.
        # failure_transport: failed

        transports:
            # https://symfony.com/doc/current/messenger.html#transport-configuration
            # async: '%env(MESSENGER_TRANSPORT_DSN)%'
            # failed: 'doctrine://default?queue_name=failed'
            # sync: 'sync://'

        routing:
            # Route your messages to the transports
            # 'App\Message\YourMessage': async

そして.envに以下の内容が追記されます。

###> symfony/messenger ###
# Choose one of the transports below
# MESSENGER_TRANSPORT_DSN=doctrine://default
# MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages
# MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages
###< symfony/messenger ###

メッセージクラスの作成

データを保持するためのメッセージクラスを作成します。メッセージクラスのオブジェクトはシリアライズされてキューに保存されます。

<?php

namespace Customize\Message;

class CustomerNotification
{
    private $customerId;

    private $content;

    public function __construct(int $customerId, string $content)
    {
        $this->customerId = $customerId;
        $this->content = $content;
    }

    public function getCustomerId(): int
    {
        return $this->customerId;
    }

    public function getContent(): string
    {
        return $this->content;
    }
}

メッセージハンドラークラスの作成

メッセージハンドラークラスはメッセージがディスパッチされたときに呼び出されます。MessageHandlerInterfaceを実装して__invokeメソッドを持たせる必要があります。

<?php

namespace Customize\MessageHandler;

use Customize\Message\CustomerNotification;
use Eccube\Entity\Customer;
use Eccube\Repository\CustomerRepository;
use Symfony\Component\Messenger\Handler\MessageHandlerInterface;

class CustomerNotificationHandler implements MessageHandlerInterface
{
    private $mailer;

    private $customerRepository;

    public function __construct(\Swift_Mailer $mailer, CustomerRepository $customerRepository)
    {
        $this->mailer = $mailer;
        $this->customerRepository = $customerRepository;
    }

    public function __invoke(CustomerNotification $notification)
    {
        /** @var Customer $customer */
        $customer = $this->customerRepository->find($notification->getCustomerId());

        $message = (new \Swift_Message())
            ->setSubject('お知らせ')
            ->setTo([$customer->getEmail()])
            ->setBody($notification->getContent());

        $result = $this->mailer->send($message);
        if ($result < 1) {
            throw new \Exception('送信に失敗しました');
        }
    }
}

メッセージをディスパッチするコントローラー

メッセージをディスパッチするコントローラーを用意します。以下のコードはすべての会員を取得して、各会員にメッセージ付きでディスパッチしています。

<?php

namespace Customize\Controller;

use Customize\Message\CustomerNotification;
use Eccube\Controller\AbstractController;
use Eccube\Entity\Customer;
use Eccube\Repository\CustomerRepository;
use Symfony\Component\HttpFoundation\JsonResponse;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Routing\Annotation\Route;

/**
 * @Route("/%eccube_admin_route%")
 */
class CustomerNotificationController extends AbstractController
{
    private $customerRepository;

    public function __construct(CustomerRepository $customerRepository)
    {
        $this->customerRepository = $customerRepository;
    }

    /**
     * @Route("/customer/notification", name="admin_customer_notification")
     */
    public function index(MessageBusInterface $bus): JsonResponse
    {
        $customers = $this->customerRepository->findAll();

        /** @var Customer $customer */
        foreach ($customers as $customer) {
            $bus->dispatch(new CustomerNotification($customer->getId(), 'お知らせです'));
        }

        return $this->json(['success' => true]);
    }
}

Doctrineトランスポートの設定

今回は非同期処理するためのトランスポートをDoctrineにします。

まずは.envでトランスポートをDoctrineに設定してください。

###> symfony/messenger ###
# Choose one of the transports below
MESSENGER_TRANSPORT_DSN=doctrine://default
# MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages
# MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages
###< symfony/messenger ###

次にmessenger.yamlを以下のように編集してください。

framework:
    messenger:
        # Uncomment this (and the failed transport below) to send failed messages to this transport for later handling.
        failure_transport: failed

        transports:
            # https://symfony.com/doc/current/messenger.html#transport-configuration
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                retry_strategy:
                    max_retries: 3 # 送信に失敗したらリトライする回数

            failed: 'doctrine://default?queue_name=failed' # 送信に失敗したらキューに残す
            # sync: 'sync://'

        routing:
            # Route your messages to the transports
            'Customize\Message\CustomerNotification': async

上記設定は送信に失敗したら3回リトライして、リトライしても送信に失敗したらキューに残すようにしています。

messenger:consumeコマンドを実行

バックグランドで動作させるために、messenger:consumeコマンドを実行する必要があります。このコマンドを実行することで監視用の別のプロセスが立ち上がります。

bin/console messenger:consume -vv

上記コマンドを実行するとデータベースにmessenger_messagesテーブルが生成されます。このテーブルにデータが登録されます。

実際に実行してみる

以下のURLをブラウザで開いて実行できます。

https://127.0.0.1:8000/admin/customer/notification

実行すると即時ページが表示され、バックグラウンドでメール送信処理が行われます。

登録されたデータは即時処理されるので送信に失敗していなかったらmessenger_messagesテーブルは空のままだと思います。

実際に処理前のデータを確認したい場合はmessenger:consumeコマンドを停止してメール送信用のページにアクセスしてみてください。

本番環境での運用について

本番環境で運用する場合、何かしらの理由で落ちてしまったり、メモリーリークが発生して正しく動作しない場合があります。

なので、バックグラウンドプロセスが常に実行されるようにSupervisorを使うことをおすすめします。

詳しい設定方法はSymfonyドキュメントを参考にしてください。