PHPackages                             yuntian001/multi-process-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. yuntian001/multi-process-queue

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

yuntian001/multi-process-queue
==============================

基于swoole的多进程队列系统，支持一键协程、多进程、低延时(最低毫秒级)、低资源占用、高可用。可与 laravel thinkphp 等框架配合使用

2.1.3(3y ago)93905↓100%11[1 issues](https://github.com/yuntian001/multi-process-queue/issues)MITPHPPHP >=7.1

Since Apr 21Pushed 3y ago4 watchersCompare

[ Source](https://github.com/yuntian001/multi-process-queue)[ Packagist](https://packagist.org/packages/yuntian001/multi-process-queue)[ RSS](/packages/yuntian001-multi-process-queue/feed)WikiDiscussions main Synced 1mo ago

READMEChangelog (7)Dependencies (3)Versions (17)Used By (0)

multi-process-queue
===================

[](#multi-process-queue)

基于swoole的多进程队列系统，manage进程管理子进程，master进程监听队列分发任务，worker进程执行任务， 多进程、低延时(最低毫秒级)、低资源占用、高可用、可多服务器分布式部署。可与 laravel thinkphp 等框架配合使用。

版本要求：

- php>=7.1
- swoole >= 4.4

当前支持的驱动

- redis （redis版本需大于3.0.2,redis扩展版本建议>=4.0.0）

### 特性

[](#特性)

- 最低毫秒级任务延时
- 自定义重试次数和错误回调
- 自定义超时时间
- 自定义启动预执行函数，方便初始化其他框架与其他项目配合使用
- master协程监听队列降低延时
- 多worker进程消费任务
- worker进程支持一键化协程
- 支持后台守护运行,无需其余进程管理工具
- 支持多服务器分布式部署
- 高可用，根据超时时间判断，确保进程挂掉后,回调函数执行成功。

### 进程结构图

[](#进程结构图)

[![](process_structure.png)](process_structure.png)

### 配置说明

[](#配置说明)

配置项类型是否必填默认值说明basicsarray是无基础配置项basics.namestring是无当前队列服务名称,多个服务同时启动时需要分别设置名字(名字自定义确保队列服务名字不重复即可)basics.pid\_pathstring否/tmp主进程pid文件存放路径basics.driverstring是无队列驱动必须是MPQueue\\Queue\\Driver\\DriverInterface的实现类worker\_start\_handlecallable否空字符串worker进程启动后会调用(当前服务所有队列有效)logarray是无日志配置log.pathstring否/tmp日志存放路径log.levelint否Monolog\\Logger::INFO日志记录级别 Monolog\\Logger::DEBUG/Monolog\\Logger::INFOlog.dirverstring/class否RotatingFileLogDriver日志驱动，必须是MPQueue\\Log\\Driver\\LogDriverInterface的实现queue二维数组是无队列配置queue\[0\].namestring是无队列名称queue\[0\].worker\_numberint否3工作进程数量queue\[0\].memory\_limitint否128工作进程最大使用内存数(单位mb)(0无限制)queue\[0\].sleep\_secondsfloot否1监视进程休眠时间（秒，最小到0.001）queue\[0\].timeoutint否120超时时间(s)以投递任务方为准queue\[0\].fail\_numberint否3最大失败次数以投递任务方为准,达到最大次数前(包括次数)失败后会重新投递(即任务最多允许执行fail\_number+1次)queue\[0\].fail\_expireint否3失败延时投递时间(s 支持小数精度到0.001)以投递任务方为准queue\[0\].fail\_handlecallable否空任务失败执行函数(当任务超时或者达到最大投递次数后会执行)queue\[0\].worker\_start\_handlecallable否空worker进程启动加载函数（当前队列有效）queue\[0\].modelint否\\MPQueue\\Config\\QueueConfig::MODEl\_DISTRIBUTE队列运行模式（\\MPQueue\\Config\\QueueConfig::MODEl\_DISTRIBUTE 分发模式 \\MPQueue\\Config\\QueueConfig::MODEL\_GRAB 抢占模式）fail\_handle 会传入两个参数 $jobInfo 任务详细信息、$e出错的异常类 fail\_handle的执行时间也受timeout的控制

#### 注意：任务超时后会直接记录为失败，不会根据fail\_number进行失败重试,可以在fail\_handle中根据$jobInfo\['type'\]判断是否是超时任务。

[](#注意任务超时后会直接记录为失败不会根据fail_number进行失败重试可以在fail_handle中根据jobinfotype判断是否是超时任务)

#### 2.0改动

[](#20改动)

1. 取消超时回调，超时后记录为失败，触发失败回调。
2. 由原来的每次失败都触发失败回调改为，超时或超过失败重试次数后才触发。
3. 增加抢占模式:此模式下worker进程会主动抢任务会增加数据库查询和连接数量（1.0版本的模式为分发模式）
4. 增加高可用性，超时时间适用于失败回调，如果进程在失败回调执行期间挂掉，其余进程会在超时时间到达后再次执行失败回调。
5. 优化分发模式性能,按当前空闲进程数一次获取多个任务进行分发，qps大大提高。

### 下个版本规划

[](#下个版本规划)

- redis集群(Redis Cluster)支持

### 配置示例

[](#配置示例)

```
[
    'basics'=>[
        'name'=>'mp-queue-1',//多个服务器同时启动时需要分别设置名字
        'driver'=> new \MPQueue\Queue\Driver\Redis('127.0.0.1'),
    ],
    'queue' => [
        [
            'name' => 'test',//队列名称
            'fail_handle'=>function(){
                var_dump('失败了');
            },//失败执行函数
        ],
        [
            'name' => 'test2',//队列名称
            'worker_number' => 4,//当前队列工作进程数量
            'memory_limit' => 0, //当前队列工作进程的最大使用内存，超出则重启。单位 MB
        ]
    ]
]

```

快速上手
----

[](#快速上手)

### 1.安装

[](#1安装)

- composer 安装

```
 composer require yuntian001/multi-process-queue

```

- 或者直接下载压缩包 然后执行composer install

### 2. 启动队列

[](#2-启动队列)

- 新建 main.php

```