How to Rate‑Limit Asynchronous Tasks with ReactPHP‑MQ

ReactPHP‑MQ aims to efficiently rate‑limit concurrent operations, preventing resource exhaustion or remote server bans. It acts as an intelligent scheduler that queues and dispatches asynchronous tasks, ensuring that only a limited number run simultaneously—useful for scenarios like high‑volume web crawling.

Key Features

Asynchronous operation execution : Handles any number of async tasks with configurable concurrency and queue size, processing results immediately for faster response.

Lightweight SOLID design : Provides a clean abstraction built on well‑tested components, avoiding unnecessary complexity and letting developers focus on business logic.

Comprehensive test coverage : Includes an automated test suite that has been extensively exercised in real‑world use, ensuring stability and reliability.

Installation

Install via Composer:

composer require clue/mq-react:^1.7

Basic Usage Example

The following PHP script demonstrates how to limit concurrent HTTP requests to three using ReactPHP‑MQ:

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

$browser = new React\Http\Browser();
$urls = file('urls.txt');

$q = new Clue\React\Mq\Queue(3, null, function ($url) use ($browser) {
    return $browser->get($url);
});

foreach ($urls as $url) {
    $q($url)->then(
        function (Psr\Http\Message\ResponseInterface $response) use ($url) {
            echo $url . ': ' . $response->getBody()->getSize() . ' bytes' . PHP_EOL;
        },
        function (Exception $e) {
            echo 'Error: ' . $e->getMessage() . PHP_EOL;
        }
    );
}

This example creates a React\Http\Browser for HTTP requests, a Queue limited to three concurrent tasks, and processes each URL from urls.txt, outputting the response size or any errors.

Advanced Usage

all() Method

Queue::all()

runs multiple tasks concurrently and resolves when all succeed; any failure rejects the whole operation.

$browser = new React\Http\Browser();
$promise = Queue::all(3, $urls, function ($url) use ($browser) {
    return $browser->get($url);
});
$promise->then(function (array $responses) {
    echo 'All ' . count($responses) . ' successful!' . PHP_EOL;
});

any() Method

Queue::any()

resolves as soon as the first task succeeds; it rejects only if every task fails.

$browser = new React\Http\Browser();
$promise = Queue::any(3, $urls, function ($url) use ($browser) {
    return $browser->get($url);
});
$promise->then(function (ResponseInterface $response) {
    echo 'First response: ' . $response->getBody() . PHP_EOL;
});

Timeout Handling

By default, the library does not limit individual task duration, but you can add timeouts using react/promise-timer:

use React\Promise\Timer;

$q = new Queue(10, null, function ($uri) use ($browser) {
    return Timer\timeout($browser->get($uri), 2.0);
});

$promise = $q($uri);