Feature Request: Redis Queue Component (Similar to Laravel Queue)

[liuxiaojinla]

Summary

Requesting a new Redis Queue component for the amphp ecosystem, providing production-ready asynchronous queue support similar to Laravel Queue, to fill the current gap in distributed task queue capabilities.

Background & Motivation

The amphp ecosystem currently provides:

• amphp/amp - Async foundation
• amphp/redis - Redis client
• amphp/parallel - Parallel processing

However, a high-level queue abstraction is missing. In real-world applications, async task queues are essential (e.g., sending emails, generating reports, image processing, message consumption).

Laravel Queue's excellent features worth referencing:

• Multi-driver support (Redis/Database/Beanstalkd/SQS)
• Clean Job definition and dispatch API
• Delayed queues, priority queues
• Failed job retry and dead letter queues
• Queue monitoring with Horizon dashboard

Desired Features

Feature	Priority	Description
Basic Queue Operations	P0	push(), pop(), size(), flush()
Multi-Queue Support	P0	Support multiple queues (e.g., default, mail, webhook)
Delayed Jobs	P1	later($delay, $job) for delayed execution
Priority Queues	P1	High-priority jobs processed first
Failure Handling	P1	Auto-retry, failure logging, dead letter queues
Job Serialization	P0	Support PHP object/closure serialization
Worker Process Management	P2	Long-running process consumption like queue:work
Concurrency Control	P2	Limit concurrent job processing
Horizon-like Monitoring	P3	Web UI for queue status, throughput, failure rates

Proposed API Design (Draft)

<?php

use Amp\Redis\Queue\QueueManager;
use Amp\Redis\Queue\Job;

// Initialization
$queue = new QueueManager($redisConfig);

// Define Job (similar to Laravel)
class SendEmailJob implements Job {
    public function __construct(
        public string $to,
        public string $subject,
        public string $body
    ) {}

    public function handle(): void {
        // Async email sending
    }
}

// Dispatch job
$queue->push(new SendEmailJob('user@example.com', 'Hello', 'World'));

// Delay for 5 minutes
$queue->later(300, new SendEmailJob(...));

// Specify queue
$queue->on('mail')->push(new SendEmailJob(...));

// Worker consumption
$queue->worker('default')
    ->concurrency(10)      // Max 10 concurrent
    ->retry(3)             // Retry 3 times on failure
    ->run();

Technical Implementation Suggestions

1. Built on amphp/redis: Leverage existing async Redis connections
2. Redis Data Structures:
   • Use LPUSH/BRPOP for basic queues
   • Use ZADD/ZRANGEBYSCORE for delayed queues (score as execution timestamp)
   • Use HASH for job metadata storage
3. Serialization: Default to serialize(), support custom serializers (JSON, MessagePack)
4. Concurrency Safety: Use Redis transactions or Lua scripts for atomicity

Relationship with Existing Ecosystem

• No wheel reinvention: Complements amphp/beanstalk by providing a Redis option
• Progressive adoption: Can be a standalone package amphp/queue or amphp/redis-queue
• Framework integration: Easy integration with Amp frameworks (e.g., amphp/http-server)

Reference Implementations

• Laravel Queue
• Laravel Horizon
• bull (Node.js Redis Queue)

Willing to Contribute

• I can participate in API design discussions
• I can provide PR implementation
• I can write documentation and test cases

---

Environment

• amphp/redis version: 2.x
• PHP version: 8.2+
• Redis version: 6.0+