Php Amqplib Laravel Package

Getting Started

Install and connect. The package name on Packagist is php-amqplib/php-amqplib (the old videlalvaro/php-amqplib name is the historical one):

composer require php-amqplib/php-amqplib

There is no framework integration or config file — you construct a connection yourself:

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$connection = new AMQPStreamConnection('rabbitmq', 5672, 'guest', 'guest', '/');
$channel    = $connection->channel();

$channel->queue_declare('orders', false, true, false, false);
$channel->basic_publish(
    new AMQPMessage(json_encode(['id' => 1]), [
        'content_type'  => 'application/json',
        'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
    ]),
    '',
    'orders'
);

$channel->close();
$connection->close();

Where to look first.

1. demo/ in the repository — this is the real documentation. Start with amqp_publisher.php / amqp_consumer.php, then connection_recovery_consume.php, amqp_connect_multiple_hosts.php, basic_get.php, amqp_consumer_pcntl_heartbeat.php.
2. The official RabbitMQ PHP tutorials — this library is the reference client they are written against, so the six tutorials map one-to-one onto the demos.
3. The generated API docs at php-amqplib.github.io/php-amqplib/ for method signatures (AMQPChannel::basic_consume, basic_qos, confirm_select, etc.).

First real use case. A single producer publishing JSON job payloads to a durable queue, and a long-running CLI consumer (Artisan command or plain PHP script under Supervisor) that processes them with manual acknowledgement. Get that working end-to-end before touching exchanges, confirms, or dead-lettering — the failure modes only make sense once you've seen a consumer survive a restart.

Implementation Patterns

Wrap the connection in a factory, not a global singleton. A connection is a socket; a channel is a logical session on top of it. In long-running processes, bind a factory that returns a fresh connection per worker, and keep channels scoped to the unit of work:

final class AmqpConnectionFactory
{
    public function __construct(private array $config) {}

    public function connect(): AMQPStreamConnection
    {
        return new AMQPStreamConnection(
            $this->config['host'], $this->config['port'],
            $this->config['user'], $this->config['password'],
            $this->config['vhost'],
            false, 'AMQPLAIN', null, 'en_US',
            3.0,   // connection timeout
            130.0, // read_write_timeout — keep > 2x heartbeat
            null, true, 0, 60  // heartbeat = 60
        );
    }
}

In Laravel, register the factory in a service provider and keep the connection out of the container as a singleton; resolve it only inside the worker command. If you want a full queue-driver abstraction instead of raw channels, vladimir-yuldashev/laravel-queue-rabbitmq sits on top of this library.

Declare topology once, idempotently, from one place. Put exchange/queue/binding declarations in a dedicated Topology class or a deploy-time command, not scattered through producers and consumers. Redeclaring with different arguments raises a channel-level PRECONDITION_FAILED (406) and closes the channel. In consumers, prefer queue_declare($name, false, true, false, false, true /* passive */) when the infrastructure team owns the broker configuration.

Producer with confirms for anything that must not be lost.

$channel->confirm_select();
foreach ($jobs as $job) {
    $channel->batch_basic_publish($this->makeMessage($job), 'orders', $job['type']);
}
$channel->publish_batch();
$channel->wait_for_pending_acks(5.0);

confirm_select() + wait_for_pending_acks() gives you at-least-once semantics; batch_basic_publish/publish_batch amortises socket writes, and reusing a single AMQPMessage instance ($msg->setBody($newBody)) avoids re-encoding properties on every publish. Both are worth it above a few hundred messages per second.

Consumer with bounded prefetch and explicit ack.

$channel->basic_qos(0, 10, false); // 10 unacked per worker
$channel->basic_consume('orders', '', false, false, false, false,
    function (AMQPMessage $msg) {
        try {
            $this->handler->handle(json_decode($msg->getBody(), true));
            $msg->ack();
        } catch (\Throwable $e) {
            $msg->nack(false, false); // route to DLX, don't hot-loop
        }
    });

while ($channel->is_consuming()) {
    $channel->wait(null, false, 5.0); // timeout lets you poll a stop flag
}

Ack after the business transaction commits, not before. Use nack($requeue = false) for poison messages and let a dead-letter exchange own retries (delay queues built from queue TTL + DLX are the standard trick, since AMQP has no native delay). Keep prefetch modest — 10–50 — because unacked messages are held in the consumer's memory.

Own the worker lifecycle. Run consumers under Supervisor or systemd, register pcntl_signal handlers for SIGTERM/SIGINT (the library dispatches them between messages automatically unless AMQP_WITHOUT_SIGNALS is defined), and make shutdown graceful: set a $shouldStop flag, then basic_cancel($tag) before $channel->close(). If handlers can run longer than the heartbeat interval, register PCNTLHeartbeatSender so the broker doesn't drop the connection mid-job.

Recovery belongs in your loop, not the library. There is no automatic reconnect. Wrap the consume loop in a while (true) with try/catch for AMQPConnectionClosedException, AMQPIOException, AMQPRuntimeException, \RuntimeException, and \ErrorException; on catch, close/null the old connection, usleep() with backoff, and re-establish the connection, channel, QoS, and consumers. Reconnect logic must rebuild everything — consumers are registered on a channel, and a channel dies with its connection.

Multiple hosts for HA clusters. AMQPStreamConnection::create_connection([$node1, $node2], $options) tries in order and throws the last failure if all fail. Combine this with your own retry/backoff loop; it does not do periodic rebalancing.

Protect memory on untrusted payloads. $channel->setBodySizeLimit(1_048_576) truncates oversized bodies before your callback ever sees them, setting $msg->is_truncated and leaving $msg->body_size accurate. Since the excess bytes are discarded from the wire, nack such messages so the broker (which still has the full copy) can dead-letter them.

Gotchas and Tips

Channels are not thread-safe and not fork-safe. One channel per process, used by one coroutine/thread at a time. If you pcntl_fork() after connecting, the child inherits a socket neither process owns correctly — open a new connection in the child. For high fan-out, AMQProxy pools connections and channels server-side and meaningfully cuts RabbitMQ CPU.

A persistent message is not automatically safe. DELIVERY_MODE_PERSISTENT only survives a broker restart if the exchange and queue are durable, and if auto_ack = true you lose in-flight messages on worker crash. Manual ack is the default you want; auto_ack is only for genuinely disposable data.

read_write_timeout must exceed the heartbeat. A common production incident is connections silently dying because heartbeat is 60s and read_write_timeout is 30s (or the default). Set read_write_timeout to at least 2× heartbeat, and remember heartbeats are only serviced inside wait() — a 10-minute handler means a dead connection unless you use PCNTLHeartbeatSender.

AMQP_DEBUG is a constant, not config. define('AMQP_DEBUG', true) must run before the first connection is used; it dumps raw protocol frames to output. Invaluable for "why does the broker close my channel," catastrophic in production logs and in any process whose stdout matters.

Watch for channel-level exceptions masquerading as application bugs. A 406 from a redeclare mismatch, a 403 from a bad vhost/permission, or a basic_publish to a nonexistent exchange with mandatory set will close the channel, not just fail one call. Check $channel->is_open() in long-lived code, and always close in finally:

finally {
    try { $channel?->close(); } catch (\Throwable) {}
    try { $connection?->close(); } catch (\Throwable) {}
}

Close order and double-close. Close the channel before the connection; swallow AMQPChannelClosedException on teardown since the broker may already have closed it.

Default exchange quirks. Publishing with exchange = '' routes by queue name directly — convenient for simple setups, but it means a typo silently drops the message (unless you set mandatory and handle basic_return). guest/guest only works from localhost; remote connections are refused by the broker, not the library.

Long-running consumers leak in userland, not in the library. The usual culprits are the framework's own state: a stale PDO connection, an ever-growing log context, EntityManager identity maps, accumulated debug bars. Reconnect the database every N messages, unset() large payload variables, and run gc_collect_cycles() periodically. Memory limits should be enforced by Supervisor restarting the worker anyway.

Signals are dispatched between messages. A SIGTERM arriving during a 30-second handler won't be handled until the handler returns — Supervisor's stopwaitsecs must exceed your longest job, or you'll get SIGKILL mid-transaction and a redelivered message.

Truncation is silent data loss. setBodySizeLimit() reads and discards; there is no way to retrieve the tail in the callback. If you enable it, always have a DLX wired up and nack truncated messages, otherwise large messages vanish.

Publisher confirms are a throughput trade. Confirm every message individually and you'll halve your publish rate; batch them and call wait_for_pending_acks() once per batch, with a timeout, and treat a timeout as an unknown delivery state (log it and make consumers idempotent rather than blindly republishing).

Extension points. The library is composition-friendly: subclass or wrap AMQPStreamConnection / AMQPChannel behind your own interfaces so tests can substitute an in-memory implementation; enqueue/amqp-lib exists if you need the queue-interop abstraction for swappable transports. AMQPLazyConnection defers the actual TCP connect until first use — useful in DI graphs where a connection may go unused.

Licensing. LGPL-2.1 means you can use it in closed-source applications as a Composer dependency; the obligation bites if you modify and redistribute the library itself. Keep patches upstreamed or in a separate fork package rather than editing vendor/, and note any modifications in your docs.