Используйте Messenger для чтения сообщений в очереди, отправленных не с помощью Messenger

Райан Уивер ответил на аналогичный вопрос в Slack Symfony.:

You will need a custom serializer for messenger if the messages do not originate from messenger :)

1) You create a custom serialize (implements SerializerInterface from Messenger) and configure it under the messenger config

2) Somehow in that serializer, you take JSON and turn it into some "message" object you have in your code. How you do that is up to you - you need to somehow be able to look at your JSON and figure out which message class it should be mapped to. You could then create that object manually and populate the data, or use Symfony's serializer. Wrap this in an Envelope before returning it

3) Because your serializer is now returning a "message" object if some sort, Messenger uses its normal logic to find the handler(s) for that Message and execute them

Я сделал быструю реализацию для своих нужд, чтобы вы соответствовали вашей бизнес-логике. :

1 - Создайте Serializer, который реализует SerializerInterface:

   // I keeped the default serializer, and just override his decode method.

   /**
     * {@inheritdoc}
     */
    public function decode(array $encodedEnvelope): Envelope
    {
        if (empty($encodedEnvelope['body']) || empty($encodedEnvelope['headers'])) {
            throw new InvalidArgumentException('Encoded envelope should have at least a "body" and some "headers".');
        }

        if (empty($encodedEnvelope['headers']['action'])) {
            throw new InvalidArgumentException('Encoded envelope does not have an "action" header.');
        }

        // Call a factory to return the Message Class associate with the action
        if (!$messageClass = $this->messageFactory->getMessageClass($encodedEnvelope['headers']['action'])) {
            throw new InvalidArgumentException(sprintf('"%s" is not a valid action.', $encodedEnvelope['headers']['action']));
        }

        // ... keep the default Serializer logic

        return new Envelope($message, ...$stamps);
    }

2 - Получить право Message с помощью фабрики:

class MessageFactory
{
    /**
     * @param string $action
     * @return string|null
     */
    public function getMessageClass(string $action)
    {
        switch($action){
            case ActionConstants::POST_MESSAGE :
                return PostMessage::class ;
            default:
                return null;
        }
    }
}

3) Настройте свой новый пользовательский сериализатор для мессенджера:

framework:
  messenger:
    serializer: 'app.my_custom_serializer'

Я попытаюсь пойти немного дальше и найти способ «подключить» очередь напрямую, дам вам знать.