Getting Started

pdezwart/php-amqp is not a Composer library — it's the C extension that provides the AMQPConnection / AMQPChannel / AMQPExchange / AMQPQueue / AMQPEnvelope classes. Your first step is packaging, not composer require.

1. Install the extension where PHP runs (not in vendor/).

# Debian/Ubuntu
apt-get install -y librabbitmq-dev
pecl install amqp          # or: apt-get install php8.2-amqp

# Dockerfile
RUN apt-get update && apt-get install -y librabbitmq-dev \
 && pecl install amqp && docker-php-ext-enable amqp

2. Fail fast in CI and deploys by declaring the dependency in composer.json:

"require": { "ext-amqp": "*" }

3. Verify what you actually installed. This matters more than usual, because the API changed between the pdezwart-era releases and the current php-amqp/php-amqp fork that ships as the PECL amqp package today:

php -m | grep amqp
php --ri amqp            # extension version + librabbitmq version

Where to look first: the stub files in the repo (*.stub.php / stubs/) are the authoritative API surface — more reliable than blog posts, which mix the old and new APIs freely. Then the PHP manual's "AMQP" section for the class list.

First use case — publish, then consume one message:

$conn = new AMQPConnection([
    'host'  => 'rabbitmq',
    'port'  => 5672,
    'vhost' => '/',
    'login' => 'guest',
    'password' => 'guest',
    'read_timeout' => 2,      // seconds
    'write_timeout' => 2,
    'connect_timeout' => 2,
]);
$conn->connect();

$ch = new AMQPChannel($conn);

$ex = new AMQPExchange($ch);
$ex->setName('jobs');
$ex->setType(AMQP_EX_TYPE_DIRECT);
$ex->setFlags(AMQP_DURABLE);
$ex->declareExchange();

$q = new AMQPQueue($ch);
$q->setName('jobs.email');
$q->setFlags(AMQP_DURABLE);
$q->declareQueue();
$q->bind('jobs', 'email');

$ex->publish(
    json_encode(['to' => '[email protected]']),
    'email',
    AMQP_NOPARAM,
    ['delivery_mode' => AMQP_DURABLE, 'content_type' => 'application/json']
);

$q->consume(function (AMQPEnvelope $env, AMQPQueue $queue) {
    echo $env->getBody(), PHP_EOL;
    $queue->ack($env->getDeliveryTag());
    return true;              // false stops the loop
}, AMQP_NOPARAM);

Implementation Patterns

Connection per process, channel per unit of work. Connections are expensive and not fork-safe; channels are cheap and are where errors surface. Build one AMQPConnection as a singleton for the lifetime of the worker, then open a fresh AMQPChannel per job batch so a channel-level failure doesn't take down the connection.

Configure from an options array, not setters. The constructor array (host, port, vhost, login, password, read_timeout, write_timeout, connect_timeout, and — depending on build — heartbeat, channel_max, frame_max) is the cleanest way to feed env config:

new AMQPConnection([
    'host'     => config('amqp.host'),
    'port'     => (int) config('amqp.port'),
    'vhost'    => config('amqp.vhost'),
    'login'    => config('amqp.user'),
    'password' => config('amqp.pass'),
    'read_timeout' => 5,
]);

Declare topology idempotently at boot. Exchange/queue/bind declarations are safe to repeat as long as the arguments match exactly (see Gotchas). Wrap them so producers and consumers can both call ensureTopology().

Publisher pattern — attach the metadata you'll need to debug and route:

$exchange->publish($payload, $routingKey, AMQP_NOPARAM, [
    'delivery_mode' => AMQP_DURABLE,       // 2 — survives broker restart
    'content_type'  => 'application/json',
    'message_id'    => $uuid,
    'correlation_id'=> $requestId,
    'app_id'        => 'billing',
    'timestamp'     => time(),
    'headers'       => ['retry' => 0, 'schema' => 3],
]);

Consumer pattern — prefetch one, ack explicitly, nack poison messages:

$ch->setPrefetchCount(1);                 // qos; otherwise the broker floods you

$q->consume(function (AMQPEnvelope $env, AMQPQueue $queue) {
    try {
        $this->handler->handle($env);
        $queue->ack($env->getDeliveryTag());
    } catch (\Throwable $e) {
        $attempt = (int) ($env->getHeader('retry') ?? 0) + 1;
        if ($attempt >= 3) {
            $queue->nack($env->getDeliveryTag(), AMQP_NOPARAM); // requeue=false → DLX
        } else {
            $queue->nack($env->getDeliveryTag(), AMQP_REQUEUE);
        }
    }
    return true;
});

Polling instead of blocking consume() for low-volume or interleaved workloads — AMQPQueue::get() returns an AMQPEnvelope or false, letting you check a shutdown flag between polls:

while (!$this->shouldStop()) {
    $env = $queue->get(AMQP_NOPARAM);
    if ($env) { $this->handle($env); } else { usleep(200_000); }
}

RPC / request-reply with an exclusive reply queue and correlation_id matching — publish() on the reply queue's name via the default exchange ('') is the idiomatic shortcut.

Laravel wiring. Laravel's queue system is built on php-amqplib, so this extension does not plug into QUEUE_CONNECTION by itself. Two sensible approaches:

Keep ext-amqp for direct pub/sub and events, and use vladimir-yuldashev/laravel-queue-rabbitmq for anything you need dispatch()ed onto a queue.
Bind in a service provider and run your own worker command under Supervisor:

$this->app->singleton(AMQPConnection::class, function () {
    $c = new AMQPConnection(config('amqp.connection'));
    $c->connect();
    return $c;
});

Whatever you choose, put a thin interface in front of it (Publisher::publish(string $routingKey, array $payload), Consumer::run()) and implement it with the extension classes. Internal classes are mockable but brittle; an interface lets you swap in php-amqplib later without touching call sites, and unit tests stay broker-free.

Gotchas and Tips

Version drift is the number one trap. The pdezwart repo is the historical home; the actively maintained extension is php-amqp/php-amqp and ships as PECL amqp. Method names changed across that boundary (declareExchange() vs declare(), ack-by-tag vs ack-by-envelope, added heartbeat/confirm support). Always check php --ri amqp before copying a snippet, and pin the extension version in your image.
Forking breaks connections. Sharing an AMQPConnection across pcntl_fork / Supervisor workers produces silent stalls. Reconnect in the child, or open the connection after the fork.
Half-open connections with no heartbeat. Older builds lack heartbeat support, so a dead TCP peer leaves you blocked in consume() forever. Set read_timeout/connect_timeout, add a Supervisor stopwaitsecs + restart policy, and prefer a version with heartbeats if uptime matters.
AMQP_AUTOACK is a data-loss footgun. If the worker dies mid-callback the message is already gone. Ack manually unless you genuinely can tolerate loss.
Flags are bitmasks. AMQP_DURABLE, AMQP_PASSIVE, AMQP_AUTODELETE, AMQP_EXCLUSIVE, AMQP_REQUEUE are integer flags, not booleans. Forgetting AMQP_DURABLE on the exchange, the queue, or delivery_mode in the message means messages vanish on broker restart — you need all three.
PRECONDITION_FAILED on re-declare means an argument mismatch (x-message-ttl, x-dead-letter-exchange, x-max-priority, type). RabbitMQ won't let you mutate a live queue. Fix by versioning queue names (jobs.email.v2) or declaring passively (AMQP_PASSIVE) and managing topology out-of-band.
Errors are channel-scoped. A bad declare or publish throws AMQPQueueException / AMQPExchangeException / AMQPChannelException and closes the channel; AMQPConnectionException closes the connection. Catch, decide, and rebuild the channel — don't assume you can keep using it. Log $e->getMessage(), which usually carries the broker's reply text.
Prefetch defaults to unlimited, which distributes work unevenly and lets one slow worker buffer thousands of messages. setPrefetchCount(1..N).
consume() blocks with no timeout in the pdezwart-era API, so graceful shutdown is awkward — either use get() polling, or install a pcntl_signal handler and rely on read timeouts to break the loop.
Publisher confirms / transactions. If your correctness model needs confirms, verify the extension version supports confirmSelect(); otherwise you're limited and should consider php-amqplib (which does confirms well) for that specific producer.
Memory in long-running workers. Envelope objects and large bodies accumulate if you hold references. unset($env) at the end of each loop and sanity-check with memory_get_usage() in a soak test.
Large payloads get fragmented against frame_max and can hit broker limits. Keep messages small and store blobs in S3/Redis with a pointer in the body.
Debugging tools: rabbitmqctl list_queues name messages consumers and list_consumers (or the management UI) tell you instantly whether the problem is your consumer or the topology. Always check consumers before blaming the code.
Testing: integration tests need a real broker — run RabbitMQ in Docker, use a per-run vhost or queue-name suffix, and purge in setUp. Never call consume() in tests; use get() with a short read_timeout.

Php Amqp Laravel Package

Getting Started

Implementation Patterns

Gotchas and Tips