prolic/humus-amqp
HumusAmqp is an asynchronous AMQP 0-9-1 client for PHP built on ReactPHP. It provides producers, consumers, and a fluent API for RabbitMQ messaging, enabling non-blocking publishing and consuming without blocking your event loop.
prolic/humus-amqp is a thin, object-oriented wrapper around the ext-amqp PHP extension. It is not a Laravel queue driver — there is no Queue::push() integration, no job serialization, no worker lifecycle. What it gives you is a small set of abstractions (Connection → Channel → Exchange/Queue → Producer/Consumer) plus JSON convenience wrappers, so you can build your own messaging layer. In a Laravel app you consume it as a library, not as a framework module.
Requirements first, code second:
# ext-amqp is mandatory; there is no pure-PHP fallback
apt-get install -y librabbitmq-dev
pecl install amqp
docker-php-ext-enable amqp
composer require prolic/humus-amqp
Check the composer.json PHP constraint against your runtime before committing — the packagist description still advertises PHP 7.4, and the last release was September 2023. Verify it installs cleanly on 8.2/8.3 rather than assuming.
Where to look first (the src/ tree is small, read it in this order):
Connection / Channel interfaces — the two objects you must own the lifecycle of.Driver/AmqpExtension/ — the concrete implementations; this is where ext-amqp behaviour leaks through.Producer / Consumer and their JsonProducer / JsonConsumer subclasses — the day-to-day API.Envelope, Message, DeliveryMode — payload and delivery semantics.Serializer/Json — what JsonProducer/JsonConsumer actually do.Your first use case — publish a domain event from a controller, consume it in an Artisan command:
use Humus\Amqp\Driver\AmqpExtension\Connection;
use Humus\Amqp\JsonProducer;
use Humus\Amqp\DeliveryMode;
$connection = new Connection([
'host' => config('amqp.host'),
'port' => (int) config('amqp.port'),
'login' => config('amqp.user'),
'password' => config('amqp.password'),
'vhost' => config('amqp.vhost', '/'),
'heartbeat' => 30,
'read_timeout' => 5,
]);
$connection->connect();
$channel = $connection->newChannel();
$exchange = $channel->newExchange();
$exchange->setName('events');
$exchange->setType('topic');
$exchange->setFlags(AMQP_DURABLE);
$exchange->declareExchange();
$producer = new JsonProducer($exchange);
$producer->publish(
['order_id' => 42, 'total' => 1999],
'order.placed', // routing key
['delivery_mode' => DeliveryMode::PERSISTENT] // attributes
);
Then JsonConsumer on a bound queue:
$consumer = new JsonConsumer($queue); // queue bound to events / order.*
$channel->setQos(0, 10); // prefetch 10
$consumer->consume(function (array $body, \Humus\Amqp\Envelope $envelope) use ($consumer) {
// ... handle ...
$consumer->ack($envelope);
}, AMQP_NOPARAM); // no AMQP_AUTOACK => manual ack
Singleton connection, per-purpose channels. AMQPConnection is expensive; a channel is cheap and disposable. Bind one Connection as a singleton in a service provider, and create a fresh channel for each producer and for each consumer. Never share a channel between a blocking consumer and a publisher in the same process — a channel-level error closes it for everything using it.
// app/Providers/AmqpServiceProvider.php
$this->app->singleton(Connection::class, function () {
$c = new Connection(config('amqp.connection'));
$c->connect();
return $c;
});
$this->app->bind(JsonProducer::class, function ($app) {
$exchange = $app->make(Connection::class)->newChannel()->newExchange();
$exchange->setName(config('amqp.exchange'));
$exchange->setType('topic');
$exchange->setFlags(AMQP_DURABLE);
$exchange->declareExchange();
return new JsonProducer($exchange);
});
Topic exchange + routing keys as your event bus. One durable topic exchange (events), one routing key per event type (order.placed, user.registered), one durable queue per consumer group (billing, search-indexer) with bindings order.* / #. This is where the package pays off: adding a consumer is a declareQueue() + bind(), not a framework change. Keep topology declaration in a single deploy-time command (php artisan rabbit:declare) rather than scattering declareExchange() calls through request paths.
Consumers live in long-running Artisan commands under Supervisor. A consume() call blocks indefinitely. Signature like rabbit:consume {consumer}, resolve the queue by name, call setQos once, and install signal handlers so SIGTERM from Supervisor does a clean stop:
pcntl_async_signals(true);
pcntl_signal(SIGTERM, fn () => $consumer->stop());
Reliability knobs you will actually want. For at-least-once publishing, confirmSelect() on the channel and wait for confirms on critical messages (adds round-trip latency — don't do it for fire-and-forget logs). For visibility into unroutable messages, publish with AMQP_MANDATORY and register setReturnCallback(); otherwise a typo'd routing key silently drops the message. For failures, pre-declare a dead-letter exchange and set x-dead-letter-exchange in the queue arguments at creation time — you cannot add it later without deleting the queue.
Callback consumers wire naturally to Laravel services. Because the callback is a plain callable, you resolve a handler from the container inside the callback. For batch workloads, buffer in the callback and ack() after the batch commits; for per-message work, ack immediately after success and reject()/nack() with requeue on expected failures.
Testing. There is no in-memory transport. Run a real broker in CI (docker-compose or testcontainers), use a unique vhost per test run, and purge via the RabbitMQ management HTTP API between tests. Assert on published messages by consuming from a test-only queue with a short read_timeout, not by mocking the driver — mocking Envelope gives you almost no coverage of the parts that break.
ext-amqp is non-negotiable and awkward to install. pecl install amqp needs librabbitmq-dev; the official php images don't include it. Laravel Sail users must edit the Dockerfile, not just composer.json. If your host can't load the extension, the package is unusable — evaluate php-amqplib/vladimir-yuldashev/laravel-queue-rabbitmq instead.PRECONDITION_FAILED on redeclare. Declaring an existing queue or exchange with different arguments (durability, x-dead-letter-exchange, x-message-ttl) throws. RabbitMQ does not let you change arguments in place. Use a policy for tunables, and be prepared to delete/recreate during development. Idempotent declaration only works if the arguments match exactly.delivery_mode = 2 on the message. Miss any one and a broker restart loses data. JsonProducer will not set persistence for you.JsonConsumer decodes the body and hands your callback (array $body, Envelope $envelope); it will throw if a message arrives with a non-JSON body. If a non-PHP publisher writes to the same queue, use a plain Consumer and decode yourself, or partition the queue.AMQP_NOPARAM means you must ack()/reject(). If you forget, prefetch fills and the consumer stalls with no error — the classic "my consumer just stopped" bug. AMQP_AUTOACK is fine only for idempotent, lossy workloads.heartbeat and a read_timeout longer than the heartbeat interval. With heartbeats disabled, a NAT/idle timeout silently kills the socket and consume() returns without your callback ever firing again.consume() returns on connection loss rather than reconnecting. Wrap the run loop, catch AMQPConnectionException/AMQPChannelException, call disconnect() + reconnect(), and back off. Supervisor's autorestart=true is a blunt but acceptable fallback.DB::disconnect(), Model::clearBootedModels() where appropriate, and restart workers after N messages. Consider gc_collect_cycles() in the callback for large payloads.prolic/humus-amqp-module gives you config-driven producer/consumer factories and console commands, but it depends on laminas-servicemanager. Bridging that into Laravel's container costs more than the ~100 lines of manual wiring above. Use the core package only.ext-amqp build in your Dockerfile under version control, and isolate usage behind your own EventPublisher interface so a swap to php-amqplib or a Laravel-native queue driver is a one-class change.rabbitmqctl list_queues name messages consumers, and $channel->setReturnCallback() for dropped messages. Most "lost message" reports are an unbound routing key or a missing mandatory flag — check the exchange bindings before the PHP code.How can I help you explore Laravel packages today?