Phpamqplib Messenger Laravel Package

Getting Started

1. Install and register the bundle. This is a pure-PHP alternative to the ext-amqp transport — it uses php-amqplib over a socket, so no C extension is needed (which matters for Docker images and shared hosting).

composer require jwage/phpamqplib-messenger

If Flex doesn't register it, add it to config/bundles.php:

Jwage\PhpAmqpLibMessengerBundle\PhpAmqpLibMessengerBundle::class => ['all' => true],

2. Point a Messenger transport at it. The DSN scheme is amqp-lib://; the last path segment is the exchange name and %2f is the vhost.

MESSENGER_AMQP_DSN=amqp-lib://guest:guest@rabbitmq:5672/%2f/messages

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            amqp_lib:
                dsn: '%env(MESSENGER_AMQP_DSN)%'
                options:
                    auto_setup: true
                    prefetch_count: 10
                    heartbeat: 15
                    exchange:
                        name: messages
                        type: direct
                    queues:
                        messages:
                            binding_keys: [messages]
        routing:
            'App\Message\SendInvoice': amqp_lib

3. Verify the topology before writing handlers.

bin/console messenger:setup-transports amqp_lib
bin/console debug:config framework messenger   # confirms options parsed as you expect

Then check RabbitMQ (rabbitmqctl list_queues name messages consumers or the management UI): you should see the exchange, the queue, and the binding.

4. First use case — the classic "make the web request fast" move. Take an existing synchronous service call (PDF generation, email, third-party API push), wrap its arguments in a message class, dispatch it from the controller, and consume it in a worker. This is where the package earns its keep: a two-line change replaces a multi-second request with a publish, and DelayStamp gives you retries without any extra infrastructure.

$this->bus->dispatch(new SendInvoice($invoiceId));                    // async now
$this->bus->dispatch(new SendInvoice($invoiceId), [new DelayStamp(5000)]); // delayed

First place to look when something breaks: bin/console messenger:consume amqp_lib -vv output plus the RabbitMQ management UI's "Queues → Messages ready/unacked" counters.

Implementation Patterns

One exchange per bounded context, one queue per consumer role. Publish to an exchange; bind a queue per worker type so a slow consumer doesn't starve a fast one.

options:
    exchange:
        name: orders
        type: topic
        default_publish_routing_key: order.created
    queues:
        orders_email:
            binding_keys: ['order.created', 'order.refunded']
        orders_erp_sync:
            binding_keys: ['order.*']

Delayed retries via DelayStamp + retry strategy. Symfony's retry mechanism re-publishes the message with a DelayStamp, so delay support is not optional — configure the delay exchange explicitly:

options:
    delay:
        exchange:
            name: delay
            type: direct
            default_publish_routing_key: delay
        queue_name_pattern: 'delay_%delay%'
retry_strategy:
    max_retries: 3
    delay: 1000
    multiplier: 2
    max_delay: 10000

Batch consumption for throughput-sensitive workers. Set batch_size and make prefetch_count >= batch_size, otherwise the batch never fills and you sit on a timer instead of acking.

options:
    prefetch_count: 50
    batch_size: 50

class ImportRowHandler implements BatchHandlerInterface
{
    use BatchHandlerTrait;

    public function __invoke(ImportRow $message, AckStamp $ack, Context $context): ?self
    {
        $this->buffer[] = $message;

        return $this->shouldFlush($context) ? $this : null;
    }

    private function process(array $jobs): void
    {
        $this->importer->bulkInsert(array_map(fn ($j) => $j[0]->row, $jobs));
    }
}

Deploy-time topology, not runtime topology. Keep auto_setup: true locally for fast iteration, but in production run messenger:setup-transports as a deploy step and set auto_setup: false so publishers don't pay a queue.declare round trip on cold start.

Supervisor + bounded workers. Always run with limits so memory leaks in handlers (Doctrine identity map, etc.) don't accumulate:

command=php bin/console messenger:consume amqp_lib --time-limit=3600 --memory-limit=128M -vv

Chaining async work. Publishing from inside a handler is fine and idiomatic — just keep each hop's exchange/queue separate so a poison message in step 3 doesn't wedge steps 1 and 2. Avoid sync:// in the middle of a chain; it silently serialises everything back into one process.

Failure transport elsewhere. Use Doctrine or Redis as failure_transport — messenger:failed:show against AMQP is awkward and you lose easy redelivery.

Testing. Swap the transport for in-memory:// in config/packages/test/messenger.yaml and assert on TransportInterface::get(), or use sync:// when you want the handler exercised inline. Only hit a real broker in an integration suite with Testcontainers/docker-compose.

Gotchas and Tips

requeue_on_failure can create a hot loop. If nack-and-requeue is enabled and Symfony's retry strategy is also active, a permanently failing message can bounce back onto the queue immediately and burn CPU. Set requeue_on_failure: false and let the retry strategy + delay exchange + failure transport own the lifecycle.

Delay queue setup now retries on stale connections (0.10.1+). When a delayed message is published, the transport has to declare the per-delay queue before publishing. On a long-lived worker or an idle PHP-FPM process, that declaration could hit a connection the broker had already closed, and the publish failed — a DelayStamp dispatch blowing up on the first message after an idle period. 0.10.1 retries the delay queue setup when the underlying AMQP connection is stale, so the declaration goes through on a fresh connection instead of surfacing as a publish exception. Practical effect: the "first delayed message after a quiet spell throws" class of intermittent bug goes away; you still want a bounded retry strategy and a heartbeat setting that matches your traffic pattern (see below), and if you were working around it with a try/catch-and-redispatch on DelayStamp publishes, remove that wrapper — it just doubles publishes now.

Delay queues multiply. queue_name_pattern: 'delay_%delay%' creates a queue per distinct delay value. A multiplier-based retry strategy produces a small, predictable set (1s, 2s, 4s…); a random or per-message delay produces thousands of queues that never get cleaned up. Cap with max_delay and clean up stale delay queues.

Heartbeats vs. long handlers. With heartbeat: 15 and a handler that blocks for a minute, the broker drops the connection mid-ack. Either keep handlers short, raise the heartbeat well beyond your p99 handler duration, or set heartbeat: 0 on transports with genuinely slow work (and accept the dead-connection detection tradeoff). This is the single most common "works locally, dies in prod" failure with this package.

Prefetch vs. batch vs. worker count. Total unacked messages = prefetch_count × worker count. Ten workers at prefetch 50 is 500 messages invisible to other consumers; a deploy or a crash then stalls far more work than you expect. Tune prefetch_count down as worker count goes up.

Bindings are not re-declared for running consumers. Change binding_keys in config and running workers keep their old binding until restarted; a queue.declare with different arguments also fails hard (PRECONDITION_FAILED). During schema changes, either bump the queue name (cheap and safe) or delete the queue explicitly.

ext-amqp stamps don't apply. AmqpStamp / AmqpReceivedStamp are from the extension transport; handlers that type-hint them won't get them here. Everything routing-related lives in transport options and the DSN — if you need two different routing keys per message type, define two transports pointing at the same exchange with different default_publish_routing_key.

Interop with non-PHP producers. Symfony's serializer reads the message class from the type header. A Go/Python publisher must set type (and Content-Type) correctly, or switch the transport to symfony/messenger's JSON serializer and keep payloads flat. Test this before you promise a polyglot queue.

Message size. RabbitMQ's default max_message_size and the frame_max negotiate a hard ceiling; large payloads also blow prefetch_count memory budgets in PHP. Use the claim-check pattern (store the blob in S3/DB, send the ID) once payloads pass a few hundred KB.

Version pinning matters here. The package is still pre-1.0 (0.10.x at time of writing), maintained by a single author with a small but active install base; releases are mostly patch-level bugfixes on the connection/delay-queue plumbing, and the bundle/factory wiring has changed across minor versions. Pin a caret version (^0.10), read the release notes before bumping — a "patch" here can touch connection handling — and confirm the bundle FQCN in vendor/ after install rather than assuming the docs match.

Debugging checklist, in order: messenger:setup-transports run cleanly → queue exists and is bound (rabbitmqctl list_bindings) → consumers > 0 → messages_ready drops when you publish → only then look at handler code. For delayed publishes specifically, confirm the delay_* queue exists and had recent activity; if delayed messages silently vanish, the declaration path is where to look first. Enable Messenger's -vv output plus a PSR-3 logger wired into the connection and you'll usually see the exception on the publish path rather than guess at it.

Extension point. If you need custom exchange types (headers, consistent-hash) or per-message properties the transport options don't expose, subclass the transport factory and register it with the messenger.transport_factory tag — that's a far smaller change than forking the connection layer.