目录
背景
新的设计
遇到的问题
1、如何控制最大进程数
2、如何解决进程残留
新系统的评估
最终,经过测试,单核1G的服务器执行1到3秒的任务可以做到800的并发。
首页 后端开发 php教程 基于Swoole和Redis实现的并发队列处理系统

基于Swoole和Redis实现的并发队列处理系统

Jun 23, 2016 pm 01:08 PM

背景

由于PHP不支持多线程,但是作为一个完善的系统,有很多操作都是需要异步完成的。为了完成这些异步操作,我们做了一个基于Redis队列任务系统。

大家知道,一个消息队列处理系统主要分为两大部分:消费者和生产者。

在我们的系统中,主系统作为生产者,任务系统作为消费者。

具体的工作流程如下:1、主系统将需要需要处理的任务名称+任务参数push到队列中。2、任务系统实时的对任务队列进行pop,pop出来一个任务就fork一个子进程,由子进程完成具体的任务逻辑。

具体代码如下:

1

/** * 启动守护进程 */public function runAction() {    Tools::log_message('ERROR', 'daemon/run' . ' | action: restart', 'daemon-');    while (true) {        $this->fork_process();    }    exit;}/** * 创建子进程 */private function fork_process() {    $ppid = getmypid();    $pid = pcntl_fork();    if ($pid == 0) {//子进程        $pid = posix_getpid();        //echo "* Process {$pid} was created \n\n";        $this->mq_process();        exit;    } else {//主进程        $pid = pcntl_wait($status, WUNTRACED); //取得子进程结束状态        if (pcntl_wifexited($status)) {            //echo "\n\n* Sub process: {$pid} exited with {$status}";            //Tools::log_message('INFO', 'daemon/run succ' . '|status:' . $status . '|pid:' . $ppid . '|childpid:' . $pid );        } else {            Tools::log_message('ERROR', 'daemon/run fail' . '|status:' . $status . '|pid:' . $ppid . '|childpid:' . $pid, 'daemon-');        }    }}/** * 业务任务队列处理 */private function mq_process() {    $data_pop = $this->masterRedis->rPop($this->redis_list_key);    $data = json_decode($data_pop, 1);    if (!$data) {        return FALSE;    }    $worker = '_task_' . $data['worker'];    $class_name = isset($data['class']) ? $data['class'] : 'TaskproModel';    $params = $data['params'];    $class = new $class_name();    $class->$worker($params);    return TRUE;}

登录后复制

这是一个简单的任务处理系统。

通过这个任务系统帮助我们实现了异步,到目前为止已经稳定运行了将近一年。

但很可惜,它是一个单进程的系统。它是一直在不断的fork,如果有任务就处理,没有任务就跳过。

这样很稳定。

但问题有两个:一是不断地fork、pop会浪费服务器资源,二是不支持并发!

第一个问题还好,但第二个问题就很严重。

当主系统 同时 抛过来大量的任务时,任务的处理时间就会无限的拉长。

新的设计

为了解决并发的问题,我们计划做一个更加高效强壮的队里处理系统。

因为在PHP7之前不支持多线程,所以我们采用多进程。

从网上找了不少资料,大多所谓的多进程都是N个进程同时在后台运行。

显然这是不合适的。

我的预想是:每pop出一个任务就fork一个任务,任务执行完成后子进程结束。

遇到的问题


1、如何控制最大进程数

这个问题很简单,那就是每fork一个子进程就自增一次。而当子进程执行完成就自减一次。

自增没有问题,我们就在主进程中操作就完了。那么该如何自减呢?

可能你会说,当然是在子进程中啊。但这里你需要注意:当fork的时候是从主进程复制了一份资源给子进程,这就意味着你无法在子进程中操作主进程中的计数器!

所以,这里就需要了解一个知识点:信号。

具体的可以自行Google,这里直接看代码。

1

// install signal handler for dead kidspcntl_signal(SIGCHLD, array($this, "sig_handler"));

登录后复制

这就安装了一个信号处理器。当然还缺少一点。

1

declare(ticks = 1);

登录后复制

declare是一个控制结构语句,具体的用法也请去Google。

这句代码的意思就是每执行一条低级语句就调用一次信号处理器。

这样,每当子进程结束的时候就会调用信号处理器,我们就可以在信号处理器中进行自减。

2、如何解决进程残留

在多进程开发中,如果处理不当就会导致进程残留。

为了解决进程残留,必须得将子进程回收。

那么如何对子进程进行回收就是一个技术点了。

在pcntl的demo中,包括很多博文中都是说在主进程中回收子进程。

但我们是基于Redis的brpop的,而brpop是阻塞的。

这就导致一个问题:当执行N个任务之后,任务系统空闲的时候主进程是阻塞的,而在发生阻塞的时候子进程还在执行,所以就无法完成最后几个子进程的进程回收。。。

这里本来一直很纠结,但当我将信号处理器搞定之后就也很简单了。

进程回收也放到信号处理器中去。

新系统的评估

pcntl是一个进程处理的扩展,但很可惜它对多进程的支持非常乏力。

所以这里采用Swoole扩展中的Process。

具体代码如下:

1

declare(ticks = 1);class JobDaemonController extends Yaf_Controller_Abstract{    use Trait_Redis;    private $maxProcesses = 800;    private $child;    private $masterRedis;    private $redis_task_wing = 'task:wing'; //待处理队列    public function init(){        // install signal handler for dead kids        pcntl_signal(SIGCHLD, array($this, "sig_handler"));        set_time_limit(0);        ini_set('default_socket_timeout', -1); //队列处理不超时,解决redis报错:read error on connection    }    private function redis_client(){        $rds = new Redis();        $rds->connect('redis.master.host',6379);        return $rds;    }    public function process(swoole_process $worker){// 第一个处理        $GLOBALS['worker'] = $worker;        swoole_event_add($worker->pipe, function($pipe) {            $worker = $GLOBALS['worker'];            $recv = $worker->read();            //send data to master            sleep(rand(1, 3));            echo "From Master: $recv\n";            $worker->exit(0);        });        exit;    }    public function testAction(){        for ($i = 0; $i < 10000; $i++){            $data = [                'abc' => $i,                'timestamp' => time().rand(100,999)            ];            $this->masterRedis->lpush($this->redis_task_wing, json_encode($data));        }        exit;    }    public function runAction(){        while (1){//            echo "\t now we de have $this->child child processes\n";            if ($this->child < $this->maxProcesses){                $rds = $this->redis_client();                $data_pop = $rds->brpop($this->redis_task_wing, 3);//无任务时,阻塞等待                if (!$data_pop){                    continue;                }                echo "\t Starting new child | now we de have $this->child child processes\n";                $this->child++;                $process = new swoole_process([$this, 'process']);                $process->write(json_encode($data_pop));                $pid = $process->start();            }        }    }    private function sig_handler($signo) {//        echo "Recive: $signo \r\n";        switch ($signo) {            case SIGCHLD:                while($ret = swoole_process::wait(false)) {//                    echo "PID={$ret['pid']}\n";                    $this->child--;                }        }    }}

登录后复制

最终,经过测试,单核1G的服务器执行1到3秒的任务可以做到800的并发。

ps:欢迎各位大神与我交流,不知能否做到更好~


我是闫大伯,一只奋战了两个周末的野生程序猿

本站声明
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系admin@php.cn

热AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智能驱动的应用程序,用于创建逼真的裸体照片

AI Clothes Remover

AI Clothes Remover

用于从照片中去除衣服的在线人工智能工具。

Undress AI Tool

Undress AI Tool

免费脱衣服图片

Clothoff.io

Clothoff.io

AI脱衣机

Video Face Swap

Video Face Swap

使用我们完全免费的人工智能换脸工具轻松在任何视频中换脸!

热门文章

<🎜>:泡泡胶模拟器无穷大 - 如何获取和使用皇家钥匙
4 周前 By 尊渡假赌尊渡假赌尊渡假赌
北端:融合系统,解释
4 周前 By 尊渡假赌尊渡假赌尊渡假赌
Mandragora:巫婆树的耳语 - 如何解锁抓钩
3 周前 By 尊渡假赌尊渡假赌尊渡假赌

热工具

记事本++7.3.1

记事本++7.3.1

好用且免费的代码编辑器

SublimeText3汉化版

SublimeText3汉化版

中文版,非常好用

禅工作室 13.0.1

禅工作室 13.0.1

功能强大的PHP集成开发环境

Dreamweaver CS6

Dreamweaver CS6

视觉化网页开发工具

SublimeText3 Mac版

SublimeText3 Mac版

神级代码编辑软件(SublimeText3)

热门话题

Java教程
1672
14
CakePHP 教程
1428
52
Laravel 教程
1332
25
PHP教程
1277
29
C# 教程
1257
24
说明PHP中的安全密码散列(例如,password_hash,password_verify)。为什么不使用MD5或SHA1? 说明PHP中的安全密码散列(例如,password_hash,password_verify)。为什么不使用MD5或SHA1? Apr 17, 2025 am 12:06 AM

在PHP中,应使用password_hash和password_verify函数实现安全的密码哈希处理,不应使用MD5或SHA1。1)password_hash生成包含盐值的哈希,增强安全性。2)password_verify验证密码,通过比较哈希值确保安全。3)MD5和SHA1易受攻击且缺乏盐值,不适合现代密码安全。

PHP类型提示如何起作用,包括标量类型,返回类型,联合类型和无效类型? PHP类型提示如何起作用,包括标量类型,返回类型,联合类型和无效类型? Apr 17, 2025 am 12:25 AM

PHP类型提示提升代码质量和可读性。1)标量类型提示:自PHP7.0起,允许在函数参数中指定基本数据类型,如int、float等。2)返回类型提示:确保函数返回值类型的一致性。3)联合类型提示:自PHP8.0起,允许在函数参数或返回值中指定多个类型。4)可空类型提示:允许包含null值,处理可能返回空值的函数。

PHP和Python:解释了不同的范例 PHP和Python:解释了不同的范例 Apr 18, 2025 am 12:26 AM

PHP主要是过程式编程,但也支持面向对象编程(OOP);Python支持多种范式,包括OOP、函数式和过程式编程。PHP适合web开发,Python适用于多种应用,如数据分析和机器学习。

PHP和Python:代码示例和比较 PHP和Python:代码示例和比较 Apr 15, 2025 am 12:07 AM

PHP和Python各有优劣,选择取决于项目需求和个人偏好。1.PHP适合快速开发和维护大型Web应用。2.Python在数据科学和机器学习领域占据主导地位。

您如何防止PHP中的SQL注入? (准备的陈述,PDO) 您如何防止PHP中的SQL注入? (准备的陈述,PDO) Apr 15, 2025 am 12:15 AM

在PHP中使用预处理语句和PDO可以有效防范SQL注入攻击。1)使用PDO连接数据库并设置错误模式。2)通过prepare方法创建预处理语句,使用占位符和execute方法传递数据。3)处理查询结果并确保代码的安全性和性能。

PHP:处理数据库和服务器端逻辑 PHP:处理数据库和服务器端逻辑 Apr 15, 2025 am 12:15 AM

PHP在数据库操作和服务器端逻辑处理中使用MySQLi和PDO扩展进行数据库交互,并通过会话管理等功能处理服务器端逻辑。1)使用MySQLi或PDO连接数据库,执行SQL查询。2)通过会话管理等功能处理HTTP请求和用户状态。3)使用事务确保数据库操作的原子性。4)防止SQL注入,使用异常处理和关闭连接来调试。5)通过索引和缓存优化性能,编写可读性高的代码并进行错误处理。

PHP的目的:构建动态网站 PHP的目的:构建动态网站 Apr 15, 2025 am 12:18 AM

PHP用于构建动态网站,其核心功能包括:1.生成动态内容,通过与数据库对接实时生成网页;2.处理用户交互和表单提交,验证输入并响应操作;3.管理会话和用户认证,提供个性化体验;4.优化性能和遵循最佳实践,提升网站效率和安全性。

在PHP和Python之间进行选择:指南 在PHP和Python之间进行选择:指南 Apr 18, 2025 am 12:24 AM

PHP适合网页开发和快速原型开发,Python适用于数据科学和机器学习。1.PHP用于动态网页开发,语法简单,适合快速开发。2.Python语法简洁,适用于多领域,库生态系统强大。

See all articles