CloudAMQP with PHP

We recommend php-amqplib, a pure PHP client. It needs the sockets and mbstring extensions.

composer require php-amqplib/php-amqplib

Publish and consume

Set CLOUDAMQP_URL to the AMQP URL from your instance’s details page in the CloudAMQP console. Without it, the examples connect to a local broker.

The classes live in src/ under the Example namespace. Add this to composer.json and run composer dump-autoload:

"autoload": {
  "psr-4": { "Example\\": "src/" }
}

Save the connection helper as src/Amqp.php:

<?php

namespace Example;

use PhpAmqpLib\Connection\AbstractConnection;
use PhpAmqpLib\Connection\AMQPConnectionConfig;
use PhpAmqpLib\Connection\AMQPConnectionFactory;

final class Amqp
{
    // Connects to CLOUDAMQP_URL, e.g. amqps://user:password@host/vhost
    public static function connect(): AbstractConnection
    {
        return AMQPConnectionFactory::create(self::config(getenv('CLOUDAMQP_URL') ?: 'amqp://guest:guest@localhost'));
    }

    // rawurldecode, not urldecode, so a literal "+" in a password is kept
    public static function config(string $amqpUrl): AMQPConnectionConfig
    {
        $url = parse_url($amqpUrl);
        $secure = $url['scheme'] === 'amqps';

        $config = new AMQPConnectionConfig();
        $config->setHost($url['host']);
        $config->setPort($url['port'] ?? ($secure ? 5671 : 5672));
        $config->setUser(rawurldecode($url['user'] ?? 'guest'));
        $config->setPassword(rawurldecode($url['pass'] ?? 'guest'));
        $config->setVhost(rawurldecode(ltrim($url['path'] ?? '', '/')) ?: '/');
        $config->setIsSecure($secure);

        return $config;
    }
}

Use percent-encoding for special characters in credentials and the virtual host (for example, %2F for the default / virtual host).

Save the publisher as src/Publisher.php:

<?php

namespace Example;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;

final class Publisher
{
    // Waits until the broker has confirmed the message
    public static function publish(AMQPChannel $channel, string $queue, string $message): void
    {
        $channel->queue_declare($queue, false, true, false, false);
        $channel->confirm_select();
        $channel->basic_publish(new AMQPMessage($message, [
            'content_type' => 'text/plain',
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
        ]), '', $queue);
        $channel->wait_for_pending_acks(5);
        echo " [x] Sent '$message'\n";
    }
}

Save the consumer as src/Worker.php:

<?php

namespace Example;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;

final class Worker
{
    // The message is acked after the handler has processed it
    public static function consume(AMQPChannel $channel, string $queue, callable $handler): void
    {
        $channel->queue_declare($queue, false, true, false, false);
        $channel->basic_qos(0, 10, false);
        $channel->basic_consume($queue, '', false, false, false, false, function (AMQPMessage $msg) use ($handler) {
            $handler($msg->getBody());
            $msg->ack();
        });
    }
}

Save the publisher entry point as publisher.php:

<?php

require __DIR__ . '/vendor/autoload.php';

use Example\Amqp;
use Example\Publisher;

$connection = Amqp::connect();
Publisher::publish($connection->channel(), 'hello', 'Hello CloudAMQP!');
$connection->close();

Save the consumer entry point as worker.php. Worker::consume registers the callback; the channel’s consume method runs the event loop:

<?php

require __DIR__ . '/vendor/autoload.php';

use Example\Amqp;
use Example\Worker;

$connection = Amqp::connect();
$channel = $connection->channel();
Worker::consume($channel, 'hello', function (string $message) {
    echo " [x] Received '$message'\n";
});
echo " [*] Waiting for messages. To exit press CTRL+C\n";
$channel->consume();

Run php worker.php in one terminal and php publisher.php in another.

A runnable example is in php/php-amqplib in cloudamqp/examples.

Good to know

Connect with the URL. php-amqplib doesn’t parse URLs, so the example builds an AMQPConnectionConfig from the URL on your instance’s details page in the CloudAMQP console, with setIsSecure(true) for amqps://. PHP verifies the server certificate by default.
php-amqplib doesn’t reconnect. Let your process manager restart a worker whose connection is lost.
Use publisher confirms (confirm_select, wait_for_pending_acks) and persistent messages, so a message isn’t lost if the connection or broker goes down.
Ack after processing, and set a prefetch limit with basic_qos, so a consumer doesn’t buffer more messages than it can handle.

Other PHP libraries

PECL AMQP, an extension built on the RabbitMQ C client

← TypeScript

Java →