Message Queue mit Redis konfigurieren

Sofern mehrere Prozesse zur Abarbeitung von Nachrichten aus der Message Queue von Shopware 6 eingesetzt werden sollen, muss sichergestellt werden, dass identische Nachrichten nicht von denselben Consumern abgearbeitet werden. Hierzu ist die Verwendung einer Technologie notwendig, die das atomare Sperren von Nachrichten unterstützt. Hierfür eignen sich Redis oder RabbitMQ.

Voraussetzungen

• Redis-Server
• Shopware-Server

Vorbereitung in Shopware

Zunächst muss der Admin Worker deaktiviert werden, damit die Message Queue über einen Hintergrundprozess abgearbeitet werden kann. Mehr Informationen zum Deaktivieren des Admin Workers und der initialen Einrichtung über einen Supervisor-Prozess erfahren Sie hier: Shopware 6 Frontend Message Queue Worker deaktivieren.

Konfiguration der Shopware 6 Background Queue Worker

Die Konfiguration der Supervisor-Prozesse wird in folgenden Hilfecenter-Artikeln behandelt:

• Shopware 6 Background Queue Worker (Managed)
• Shopware 6 Background Queue Worker (Unmanaged)

Konfiguration in Redis

Da Message-Queue-Jobs standardmäßig keine TTL wie in Redis besitzen, empfiehlt sich der Einsatz geeigneter Key-Eviction-Policies, deren Konfiguration im verlinkten Hilfecenter-Artikel beschrieben wird.

Wird die Message-Queue auf einem dedizierten Server betrieben, der ausschließlich diesem Zweck dient, kann problemlos die Standard-Policy noeviction verwendet werden.

Alternativ besteht die Möglichkeit, die Message-Queue gemeinsam mit dem App-Cache auf demselben Redis-Server zu betreiben. In diesem Fall sollte jedoch die Key-Eviction-Policy volatile-lru gewählt werden. Dadurch wird sichergestellt, dass bei Erreichen des maxmemory-Limits ausschließlich Cache-Einträge entfernt werden, da nur diese über eine TTL verfügen. Die Jobs der Message-Queue bleiben somit erhalten und werden nicht unbeabsichtigt gelöscht.

Konfiguration in Shopware

Erstellen Sie eine neue Konfigurationsdatei unter dem Pfad: config/packages/messenger.yaml. Mithilfe dieser Konfigurationsdatei kann der Shopware Messenger Transport-Dienst konfiguriert werden.

Fügen Sie anschließend folgenden Inhalt in die Datei ein:

Bis Shopware 6.4

parameters:
  env(MESSENGER_CONSUMER_NAME): 'consumer'

framework:
  messenger:
    transports:
      default:
        dsn: "redis://%env(REDIS_HOST_MESSENGER)%:%env(REDIS_PORT_MESSENGER)%/messages/symfony/consumer-%env(MESSENGER_CONSUMER_NAME)%/?delete_after_ack=true&delete_after_reject=true&dbindex=%env(REDIS_DB_MESSENGER)%"

Ab Shopware 6.5

# File: config/packages/messenger.yaml

parameters:
  env(MESSENGER_CONSUMER_NAME): 'consumer'

framework:
  messenger:
    transports:
      # shopware default transports
      async:
        dsn: "redis://%env(REDIS_HOST_MESSENGER)%:%env(REDIS_PORT_MESSENGER)%/messages_async?dbindex=%env(REDIS_DB_MESSENGER)%"
        options:
          consumer: "%env(MESSENGER_CONSUMER_NAME)%"
      failed:
        dsn: "redis://%env(REDIS_HOST_MESSENGER)%:%env(REDIS_PORT_MESSENGER)%/messages_failed?dbindex=%env(REDIS_DB_MESSENGER)%"
        options:
          consumer: "%env(MESSENGER_CONSUMER_NAME)%"
      low_priority:
        dsn: "redis://%env(REDIS_HOST_MESSENGER)%:%env(REDIS_PORT_MESSENGER)%/messages_low_priority?dbindex=%env(REDIS_DB_MESSENGER)%"
        options:
          consumer: "%env(MESSENGER_CONSUMER_NAME)%"

Beschreibung eines Transports

Die von uns verwendete Transport-DSN Variante für Redis ist wie folgt aufgebaut:

<transport-name>:
  dsn: "redis://<hostname>:<port>/<stream_key_name>?<options>"
  options:
    consumer: "custom-consumer-name-defined-in-environment-variable"    

Beispiel für async

async:
  dsn: "redis://%env(REDIS_HOST_MESSENGER)%:%env(REDIS_PORT_MESSENGER)%/messages_async?dbindex=%env(REDIS_DB_MESSENGER)%"
  options:
    consumer: "%env(MESSENGER_CONSUMER_NAME)%"

Die hier verwendeten Variablen müssen anschließend in der Umgebungskonfiguration .env.local von Shopware hinzugefügt werden. Öffnen Sie hierzu die .env.local Datei im Shopware Root-Verzeichnis und ergänzen Sie folgenden Teil in der bestehenden Konfiguration:

# [...]

REDIS_HOST_MESSENGER=127.0.0.1
REDIS_PORT_MESSENGER=6379
REDIS_DB_MESSENGER=1

Speichern Sie die Datei anschließend und leeren Sie den Shopware Cache über die Konsole:

bin/console cache:clear

Anschließend wird die Message Queue über Redis ausgeführt.

Abarbeitung alter Jobs in der Datenbank (nicht Redis, sondern MariaDB/MySQL)

Es besteht die Möglichkeit, dass nach der Umstellung der Message Queue auf Redis alte Jobs in der Datenbank vergessen werden, deren Ausführung u.a. in ferner Zukunft liegen kann.

Wie prüfe ich, ob alte Jobs abgearbeitet werden müssen?

Um diese Frage beantworten zu können, empfiehlt sich ein Blick in die Shopware Datenbank-Tabelle messenger_messages.

Plesk

Rufen Sie Ihr Plesk Control Panel auf, navigieren Sie in die Datenbank-Übersicht des entsprechenden Plesk-Abonnements (Webseiten & Domains → Domain → Dashboard → Datenbanken) und rufen Sie die Schaltfläche phpMyAdmin für Ihre Shopware-Datenbank auf.

Geben Sie als Nächstes im Suchen-Textfeld oben links messenger_messages ein, bestätigen Sie die Eingabe mit Enter und wählen Sie die Tabelle in der linken Spalte wie im Screenshot mit der Maus aus:

Kommandozeile

Melden Sie sich an Ihrem Datenbank-Server via SSH an und führen Sie folgenden Befehl aus:

MariaDB

mariadb -u <datenbank_benutzer> -p -h 127.0.0.1 --port 3306 <datenbank_name>

MySQL

mysql -u <datenbank_benutzer> -p -h 127.0.0.1 --port 3306 <datenbank_name>

Innerhalb der MySQL- oder MariaDB-Shell können Sie die Jobs wie folgt ausgeben:

SELECT * FROM messenger_messages;

Wie arbeite ich die alten Jobs ab?

Wie in unserem Beispiel zu erkennen, liegen dort einige Jobs, welche nach der Umstellung noch in der Datenbank vorhanden sind. Um die Umstellung auf Redis abzuschließen, sollten diese Jobs noch zukünftig abgearbeitet werden, gerade, wenn es sich um Jobs handelt, wie beispielsweise Zahlungserinnerungen nach 14 Tagen.

Hierzu müssen eigene Transports definiert werden, durch die der bin/console messenger:consume Befehl bei der Abarbeitung auf die Datenbank und nicht auf Redis zugreift.

In der Datenbank-Tabelle ist hier die Spalte queue_name essenziell. In unserem Beispiel-Screenshot finden wir einmal default und low_priority vor.

Demnach müssen wir in der messenger.yaml zwei benutzerdefinierte Transports wie folgt definieren:

# File: config/packages/messenger.yaml
# [...]
      # benutzerdefinierter Transport 1
      doctrine_low_priority:  # Name des Transports, sichtbar in messenger:consume (frei wählbarer)
        dsn: "%env(MESSENGER_TRANSPORT_DSN)%" # URL für die Abarbeitung
        options:    # Options-Definition
          queue_name: low_priority # Definition des Queue-Namens

      # benutzerdefinierter Transport 2
      doctrine_default:
        dsn: "%env(MESSENGER_TRANSPORT_DSN)%"
        options:
          queue_name: default
# [...]

Speichern Sie die Datei anschließend und leeren Sie den Shopware Cache über die Konsole:

bin/console cache:clear

Anschließend müssen die Transports doctrine_low_priority und doctrine_default in die bestehende Supervisor-Konfiguration aufgenommen werden:

# Shopware Consumer
[program:shopware-consumer-doctrine-old]
process_name=%(program_name)s_%(process_num)02d
numprocs=1
command= /var/www/vhosts/<ihre-domain.de>/.phpenv/shims/php /var/www/vhosts/<ihre-domain.de>/httpdocs/bin/console messenger:consume doctrine_default doctrine_low_priority --time-limit=120 --memory-limit=512M
environment=MESSENGER_CONSUMER_NAME="%(program_name)s_%(process_num)02d"
user=<ssh-user>
autostart=true
autorestart=true
startsecs=5
startretries=10
stopwaitsecs=3600
redirect_stderr=true
stdout_logfile=/var/log/supervisor/%(program_name)s.log

Weitere Informationen zur Einrichtung der den Background Queue Workern finden Sie in diesem Hilfecenter-Artikel.

Wie lange muss ich die benutzerdefinierten Transports mit dem Messenger Consumer abarbeiten?

Der folgende Befehl gibt die Zeile mit dem ältesten Datum in der Spalte available_at aus, anhand dessen Sie sehen können, zu wann der letzte Job ausgeführt werden soll.

SELECT * FROM messenger_messages ORDER BY available_at DESC LIMIT 1;

Plesk

Wählen Sie in der oberen Leiste die Schaltfläche SQL aus, fügen Sie obigen Befehl in das Textfeld ein und bestätigen Sie die Eingabe mit der OK-Schaltfläche unten rechts.

SQL

Öffnen Sie die MySQL- oder MariaDB-Shell wie bei der Überprüfung beschrieben und führen Sie obigen Befehl in dieser aus.

Ergebnis

Sie erhalten als Ausgabe eine Zeile, deren Spalte available_at das Datum enthält, zu dem der letzte Job aus der Message-Queue in der Datenbank abgearbeitet wird.

Sollten keine Jobs mehr in der Datenbank nach Datum des letzten auftauchen, ist die Umstellung erfolgreich umgesetzt worden und Sie können die Receiver wieder aus den Background Queue Workern entfernen. Entfernen Sie dazu einfach die obige Supervisor-Konfiguration aus der entsprechenden Datei.