PHPackages                             reactphp-x/queue - PHPackages - PHPackages  [Skip to content](#main-content)[PHPackages](/)[Directory](/)[Categories](/categories)[Trending](/trending)[Leaderboard](/leaderboard)[Changelog](/changelog)[Analyze](/analyze)[Collections](/collections)[Log in](/login)[Sign up](/register)

1. [Directory](/)
2. /
3. [Queues & Workers](/categories/queues)
4. /
5. reactphp-x/queue

ActiveLibrary[Queues & Workers](/categories/queues)

reactphp-x/queue
================

A queue implementation for ReactPHP using Redis

v1.0.0(1y ago)01MITPHP

Since Apr 3Pushed 1y agoCompare

[ Source](https://github.com/reactphp-x/queue)[ Packagist](https://packagist.org/packages/reactphp-x/queue)[ RSS](/packages/reactphp-x-queue/feed)WikiDiscussions master Synced 1mo ago

READMEChangelog (1)Dependencies (4)Versions (2)Used By (0)

ReactPHP Queue
==============

[](#reactphp-queue)

基于 ReactPHP 和 Redis 实现的异步队列系统，提供轻量级、高性能的消息队列解决方案。

特性
--

[](#特性)

- 基于 Redis 实现的轻量级队列
- 支持异步操作和非阻塞式处理
- 支持多队列和优先级队列
- 提供阻塞式出队操作
- 内置任务状态管理
- 支持多工作进程处理
- 支持内存队列（ArrayQueue）
- 支持队列前缀配置

安装
--

[](#安装)

```
composer require reactphp-x/queue
```

基础使用
----

[](#基础使用)

### 创建队列实例

[](#创建队列实例)

```
use React\EventLoop\Loop;
use Clue\React\Redis\RedisClient;
use ReactphpX\Queue\Queue;

$loop = Loop::get();
$redis = new RedisClient('redis://127.0.0.1:6379');
$queue = new Queue($redis, 'example');
```

### 基本队列操作

[](#基本队列操作)

```
// 入队操作
$queue->enqueue(['task' => 'task1', 'data' => 'some data'])->then(function () {
    echo "Task1 入队成功\n";
});

// 获取队列大小
$queue->size()->then(function ($size) {
    echo "当前队列大小: {$size}\n";
});

// 出队操作
$queue->dequeue()->then(function ($data) {
    echo "出队数据: " . json_encode($data, JSON_UNESCAPED_UNICODE) . "\n";
});

// 阻塞式出队（等待5秒）
$queue->blockingDequeue(5)->then(function ($data) {
    if ($data) {
        echo "获取到数据: " . json_encode($data, JSON_UNESCAPED_UNICODE) . "\n";
    } else {
        echo "等待超时，没有数据\n";
    }
});
```

高级特性
----

[](#高级特性)

### 多队列支持

[](#多队列支持)

支持创建多个优先级队列，实现任务优先级处理：

```
// 向不同优先级的队列添加任务
$queue->enqueue($task, 'high');
$queue->enqueue($task, 'medium');
$queue->enqueue($task, 'low');

// 按优先级处理任务
$queue->dequeue('high');
$queue->dequeue('medium');
$queue->dequeue('low');
```

### 任务状态管理

[](#任务状态管理)

使用 JobManager 管理任务状态和执行：

```
use ReactphpX\Queue\JobManager;
use ReactphpX\Queue\Storage\RedisStorageDriver;

$storage = new RedisStorageDriver($redis);
$jobManager = new JobManager($storage, $queue);

// 推送带状态跟踪的任务
$jobManager->pushJob('job-1', function () {
    return "Job completed";
}, 'default', true);

// 获取任务状态
$jobManager->getAllJobs()->then(function ($jobs) {
    foreach ($jobs as $jobId => $job) {
        echo "Job ID: $jobId, Status: {$job['status']}\n";
    }
});
```

### 工作进程

[](#工作进程)

支持多工作进程并行处理任务：

```
use ReactphpX\Queue\Consumer;

$consumer = new Consumer($queue);

// 注册任务处理器
$consumer->consume(function ($data) {
    echo "Processing task: " . json_encode($data) . "\n";
    // 处理任务逻辑
    return true;
});
```

### 内存队列

[](#内存队列)

提供基于内存的队列实现，适用于单进程场景：

```
use ReactphpX\Queue\ArrayQueue;

$queue = new ArrayQueue();

$queue->enqueue('task1')->then(function () {
    echo "Task added\n";
});

$queue->dequeue()->then(function ($data) {
    echo "Processing: $data\n";
});
```

许可证
---

[](#许可证)

MIT License

###  Health Score

24

—

LowBetter than 32% of packages

Maintenance46

Moderate activity, may be stable

Popularity1

Limited adoption so far

Community6

Small or concentrated contributor base

Maturity37

Early-stage or recently created project

 Bus Factor1

Top contributor holds 100% of commits — single point of failure

How is this calculated?**Maintenance (25%)** — Last commit recency, latest release date, and issue-to-star ratio. Uses a 2-year decay window.

**Popularity (30%)** — Total and monthly downloads, GitHub stars, and forks. Logarithmic scaling prevents top-heavy scores.

**Community (15%)** — Contributors, dependents, forks, watchers, and maintainers. Measures real ecosystem engagement.

**Maturity (30%)** — Project age, version count, PHP version support, and release stability.

###  Release Activity

Cadence

Unknown

Total

1

Last Release

411d ago

### Community

Maintainers

![](https://avatars.githubusercontent.com/u/76907477?v=4)[wpjscc](/maintainers/wpjscc)[@wpjscc](https://github.com/wpjscc)

---

Top Contributors

[![wpjscc](https://avatars.githubusercontent.com/u/76907477?v=4)](https://github.com/wpjscc "wpjscc (11 commits)")

###  Code Quality

TestsPest

### Embed Badge

![Health badge](/badges/reactphp-x-queue/health.svg)

```
[![Health](https://phpackages.com/badges/reactphp-x-queue/health.svg)](https://phpackages.com/packages/reactphp-x-queue)
```

###  Alternatives

[league/geotools

Geo-related tools PHP 7.3+ library

1.4k5.3M26](/packages/league-geotools)[amphp/parser

A generator parser to make streaming parsers simple.

14952.8M16](/packages/amphp-parser)[amphp/serialization

Serialization tools for IPC and data storage in PHP.

13451.1M18](/packages/amphp-serialization)[enqueue/enqueue

Message Queue Library

19820.0M56](/packages/enqueue-enqueue)[deliciousbrains/wp-background-processing

WP Background Processing can be used to fire off non-blocking asynchronous requests or as a background processing tool, allowing you to queue tasks.

1.1k409.8k6](/packages/deliciousbrains-wp-background-processing)[react/async

Async utilities and fibers for ReactPHP

2238.8M171](/packages/react-async)

PHPackages © 2026

[Directory](/)[Categories](/categories)[Trending](/trending)[Changelog](/changelog)[Analyze](/analyze)