Build Asynchronous RabbitMQ Clients with Workerman in PHP

Overview

workerman/rabbitmq is an asynchronous RabbitMQ client for PHP that uses the AMQP protocol. RabbitMQ is an open‑source message broker written in Erlang, supporting AMQP, STOMP, MQTT and other protocols, and is widely used for reliable, scalable message passing in distributed systems.

Installation

Install the library via Composer:

composer require workerman/rabbitmq

Consumer Example (receive.php)

<?php

declare(strict_types=1);

use Bunny\Channel;
use Bunny\Message;
use Workerman\Worker;
use Workerman\RabbitMQ\Client;

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

$worker = new Worker();
$worker->eventLoop = \Workerman\Events\Revolt::class;

$worker->onWorkerStart = function () {
    $client = Client::factory([
        'host' => '127.0.0.1',
        'port' => 5672,
        'user' => 'guest',
        'password' => 'guest',
        'vhost' => '/',
        'heartbeat' => 60,
        'heartbeat_callback' => function () { echo " [-] coroutine-consumer-heartbeat
"; },
        'interval' => [100, 300]
    ])->connect();
    $channel = $client->channel();
    $channel->queueDeclare('hello-coroutine');

    // Consumer
    $channel->consume(function (Message $message, Channel $channel, \Bunny\AbstractClient $client) {
        echo " [>] Received ", $message->content, "
";
    }, 'hello-coroutine', '', false, true);
    $client->run();
    echo " [*] Waiting for messages. To exit press CTRL+C
";

    // Producer (timer)
    \Workerman\Timer::add(5, function () use ($channel) {
        $channel->publish($message = 'Hello World By Self Timer. ' . time(), [], '', 'hello-coroutine');
        echo " [<] Sent $message
";
    });
    echo " [!] Producer timer created, interval: 5 s.
";
};
Worker::runAll();

Running the Consumer

php receive.php start

Producer Example (send.php) – Workerman

<?php

declare(strict_types=1);

use Workerman\RabbitMQ\Client;
use Workerman\Worker;

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

$worker = new Worker();
$worker->eventLoop = \Workerman\Events\Revolt::class;

$worker->onWorkerStart = function () {
    $client = Client::factory([
        'host' => 'host.docker.internal',
        'port' => 5672,
        'user' => 'guest',
        'password' => 'guest',
        'vhost' => '/',
        'heartbeat' => 60,
        'heartbeat_callback' => function () { echo "coroutine-producer-heartbeat
"; }
    ])->connect();
    $channel = $client->channel();
    $channel->queueDeclare('hello-coroutine');

    // Send a message every 5 seconds
    \Workerman\Timer::add(5, function () use ($channel) {
        $channel->publish($message = 'Hello World By Workerman Env Producer. ' . time(), [], '', 'hello-coroutine');
        echo " [x] Sent '$message'
";
    });
    echo " [!] Producer timer created, interval: 5 s.
";
};
Worker::runAll();

Running the Workerman Producer

php send.php start

Producer Example (script.php) – PHP‑FPM

<?php

declare(strict_types=1);

use Workerman\RabbitMQ\Client;

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

$client = Client::factory([
    'host' => 'host.docker.internal',
    'port' => 5672,
    'user' => 'guest',
    'password' => 'guest',
    'vhost' => '/',
    'heartbeat' => 60,
    'heartbeat_callback' => function () { echo "coroutine-producer-heartbeat
"; }
])->connect();
$channel = $client->channel();
$channel->queueDeclare('hello-coroutine');
$res = $channel->publish($message = 'Hello World By Normal Producer. ' . time(), [], '', 'hello-coroutine');
echo " [x] Sent '$message', success: $res
";

Running the PHP‑FPM Producer

php script.php

Asynchronous Consumer Example

<?php

use Bunny\Channel;
use Bunny\Message;
use Workerman\Worker;
use Workerman\RabbitMQ\Client;

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

$worker = new Worker();
$worker->onWorkerStart = function () {
    (new Client())->connect()->then(function (Client $client) {
        return $client->channel();
    })->then(function (Channel $channel) {
        return $channel->queueDeclare('hello', false, false, false, false)->then(function () use ($channel) {
            return $channel;
        });
    })->then(function (Channel $channel) {
        echo " [*] Waiting for messages. To exit press CTRL+C
";
        $channel->consume(function (Message $message, Channel $channel, Client $client) {
            echo " [x] Received ", $message->content, "
";
        }, 'hello', '', false, true);
    });
};
Worker::runAll();

Running the Asynchronous Consumer

php receive.php start

Asynchronous Producer Example

<?php

use Bunny\Channel;
use Workerman\Worker;
use Workerman\RabbitMQ\Client;

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

$worker = new Worker();
$worker->onWorkerStart = function () {
    (new Client())->connect()->then(function (Client $client) {
        return $client->channel();
    })->then(function (Channel $channel) {
        return $channel->queueDeclare('hello', false, false, false, false)->then(function () use ($channel) {
            return $channel;
        });
    })->then(function (Channel $channel) {
        echo " [x] Sending 'Hello World!'
";
        return $channel->publish('Hello World!', [], '', 'hello')->then(function () use ($channel) {
            echo " [x] Sent 'Hello World!'
";
            $client = $channel->getClient();
            return $channel->close()->then(function () use ($client) {
                return $client;
            });
        });
    })->then(function (Client $client) {
        $client->disconnect();
    });
};
Worker::runAll();

Running the Asynchronous Producer

php send.php start