纯php做类似php-fpm多进程管理服务

11-2016

纯php做类似php-fpm多进程管理服务

地狱中仰望天堂
2016-11-8
总结,我爱PHP,推荐文章
浏览:
评论:8

php-fpm，以及现在比较火的workman，swoole，其都是使用的多进程的管理模式，使用manger进程，管理其子进程，今天通过纯php做了一个类似的小东西，这样以后再做守护脚本的时候，可以使用这种方式，好处就是，多进程，更能利用多核的优势，并且可以做平滑重启。

本代码实现的功能，可以自定义子进程数，比如你定3个子进程，运行会产生1个manger进程，3个work进程，通过linux信号，来实现，进程的平滑重启。

<?php

/**
 * Class PcntlProccess
 * 多进程守护任务管理
 *
 * kill -s SIGUSR1 mange进程： 平滑重启所有子进程
 * kill | kill -s SIGTERM mange进程：平滑stop所有进程
 * kill | kill -s SIGUSR1 | kill -s SIGTERM 子进程，重启该子进程
 * author chao
 *
 */
class PcntlProccess{
    const      proccess_num  = 3;//子进程数
    const      proccess_name = 'php-proccess';//进程名前缀
    protected  $_pid;

    //入口进程
    public function run(){
        if(self::proccess_num > 0){
            //启动进程
            for ($i=0; $i<self::proccess_num; $i++)
            {
                $this->parentProccess();

            }
            //SIGUSR1 留给用户使用,此处做reload平滑重启使用
            pcntl_signal(SIGUSR1, function($this){
                $this->sig_handler(SIGUSR1);
            });
            // 程序结束(terminate)信号, 与SIGKILL不同的是该信号可以被阻塞和 处理. 通常用来要求程序自己正常退出.  shell命令kill缺省产生这 个信号.
            pcntl_signal(SIGTERM, function($this){
                $this->sig_handler(SIGTERM);
            });
            //终止进程  终端线路挂断
            pcntl_signal(SIGHUP, function($this){
                $this->sig_handler(SIGHUP);
            });
            cli_set_process_title(self::proccess_name . '-manger');
            while (true) {
                // do something
                $pid = pcntl_waitpid(-1, $status, WUNTRACED | WNOHANG);
                if($pid>0){
                    $this->parentProccess($pid);
                }
                pcntl_signal_dispatch(); // 接收到信号时，调用注册的signalHandler()
                usleep(100000);
            }
        }else{
            exit('proccess_num must > 0');
        }
    }

    function sig_handler($signo){
        switch ($signo) {
            case SIGUSR1:
                foreach($this->_pid as $key){
                    posix_kill($key,SIGUSR1);
                }
                echo date('Y-m-d H:i:s')." [manage] [SIGUSR1] all work reload\n";
                break;
            case SIGTERM:
                foreach($this->_pid as $key){
                    posix_kill($key,SIGUSR1);
                }
                echo date('Y-m-d H:i:s')." [manage] [SIGTERM] stop\n";
                die();
                break;
            case SIGHUP:
                echo date('Y-m-d H:i:s')." [manage] [SIGHUP] linux clinet exit\n";
                break;
        }
    }

    function sig_san_handler($signo){
        switch ($signo) {
            case SIGUSR1:
                echo date('Y-m-d H:i:s')." [work] [SIGUSR1] work die\n";
                die();
                break;
            case SIGTERM:
                echo date('Y-m-d H:i:s')." [work] [SIGTERM] work die\n";
                die();
                break;
            case SIGHUP:
                echo date('Y-m-d H:i:s')." [work] [SIGHUP] linux clinet exit\n";
                break;
        }
    }

    //父进程
    public function parentProccess($id=''){
        if(!empty($id)){
            unset($this->_pid[$id]);
        }
        $pid  = pcntl_fork();
        if($pid == -1)
        {
            return false;
        }
        else
        {
            if($pid)
            {
                #父进程获得子进程的pid，存入数组
                $this->_pid[$pid] = $pid;
                return true;
            }
            else
            {
                //开始发送，子进程执行完自己的任务后，退出。
                $this->sanProccess();
                exit;
            }
        }
    }

    public function sanProccess(){
        cli_set_process_title(self::proccess_name . '-work');
        echo date('Y-m-d H:i:s')." work proccess start\n";
        pcntl_signal(SIGUSR1, function($this){
            $this->sig_san_handler(SIGUSR1);
        });
        pcntl_signal(SIGHUP, function($this){
            $this->sig_san_handler(SIGHUP);
        });
        pcntl_signal(SIGTERM, function($this){
            $this->sig_san_handler(SIGTERM);
        });
        while(1){
            //todu
            pcntl_signal_dispatch();
            sleep(1);
        }
    }

}

$obj = new PcntlProccess();
$obj->run();

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

<?php

/**

* Class PcntlProccess

* 多进程守护任务管理

* kill -s SIGUSR1 mange进程：平滑重启所有子进程

* kill | kill -s SIGTERM mange进程：平滑stop所有进程

* kill | kill -s SIGUSR1 | kill -s SIGTERM 子进程，重启该子进程

* author chao

class PcntlProccess{

const proccess_num = 3;//子进程数

const proccess_name = 'php-proccess';//进程名前缀

protected $_pid;

//入口进程

public function run(){

if(self::proccess_num > 0){

//启动进程

for ($i=0; $i<self::proccess_num; $i++)

{

$this->parentProccess();

}

//SIGUSR1 留给用户使用,此处做reload平滑重启使用

pcntl_signal(SIGUSR1, function($this){

$this->sig_handler(SIGUSR1);

});

// 程序结束(terminate)信号, 与SIGKILL不同的是该信号可以被阻塞和处理. 通常用来要求程序自己正常退出. shell命令kill缺省产生这个信号.

pcntl_signal(SIGTERM, function($this){

$this->sig_handler(SIGTERM);

});

//终止进程终端线路挂断

pcntl_signal(SIGHUP, function($this){

$this->sig_handler(SIGHUP);

});

cli_set_process_title(self::proccess_name . '-manger');

while (true) {

// do something

$pid = pcntl_waitpid(-1, $status, WUNTRACED | WNOHANG);

if($pid>0){

$this->parentProccess($pid);

}

pcntl_signal_dispatch(); // 接收到信号时，调用注册的signalHandler()

usleep(100000);

}

}else{

exit('proccess_num must > 0');

}

function sig_handler($signo){

switch ($signo) {

case SIGUSR1:

foreach($this->_pid as $key){

posix_kill($key,SIGUSR1);

}

echo date('Y-m-d H:i:s')." [manage] [SIGUSR1] all work reload\n";

break;

case SIGTERM:

foreach($this->_pid as $key){

posix_kill($key,SIGUSR1);

}

echo date('Y-m-d H:i:s')." [manage] [SIGTERM] stop\n";

die();

break;

case SIGHUP:

echo date('Y-m-d H:i:s')." [manage] [SIGHUP] linux clinet exit\n";

break;

}

function sig_san_handler($signo){

switch ($signo) {

case SIGUSR1:

echo date('Y-m-d H:i:s')." [work] [SIGUSR1] work die\n";

die();

break;

case SIGTERM:

echo date('Y-m-d H:i:s')." [work] [SIGTERM] work die\n";

die();

break;

case SIGHUP:

echo date('Y-m-d H:i:s')." [work] [SIGHUP] linux clinet exit\n";

break;

}

//父进程

public function parentProccess($id=''){

if(!empty($id)){

unset($this->_pid[$id]);

}

$pid = pcntl_fork();

if($pid == -1)

{

return false;

}

else

{

if($pid)

{

#父进程获得子进程的pid，存入数组

$this->_pid[$pid] = $pid;

return true;

}

else

{

//开始发送，子进程执行完自己的任务后，退出。

$this->sanProccess();

exit;

}

public function sanProccess(){

cli_set_process_title(self::proccess_name . '-work');

echo date('Y-m-d H:i:s')." work proccess start\n";

pcntl_signal(SIGUSR1, function($this){

$this->sig_san_handler(SIGUSR1);

});

pcntl_signal(SIGHUP, function($this){

$this->sig_san_handler(SIGHUP);

});

pcntl_signal(SIGTERM, function($this){

$this->sig_san_handler(SIGTERM);

});

while(1){

//todu

pcntl_signal_dispatch();

sleep(1);

}

$obj = new PcntlProccess();

$obj->run();

php index.php命令行运行，会产生3个子进程，

[root@iZ28y2wmd7jZ ~]# ps -ef|grep php-proccess
root     11536 11051  0 18:38 pts/0    00:00:00 grep --color=auto php-proccess
root     13756 32171  0 Nov10 ?        00:00:06 php-proccess-work
root     13757 32171  0 Nov10 ?        00:00:06 php-proccess-work
root     13758 32171  0 Nov10 ?        00:00:06 php-proccess-work
root     32171     1  0 Nov09 ?        00:02:37 php-proccess-manger

[root@iZ28y2wmd7jZ ~]# ps -ef|grep php-proccess

root 11536 11051 0 18:38 pts/0 00:00:00 grep --color=auto php-proccess

root 13756 32171 0 Nov10 ? 00:00:06 php-proccess-work

root 13757 32171 0 Nov10 ? 00:00:06 php-proccess-work

root 13758 32171 0 Nov10 ? 00:00:06 php-proccess-work

root 32171 1 0 Nov09 ? 00:02:37 php-proccess-manger

你随意kill一个，管理进城会立即补充一个，传递SIGUSR1给子进程，会平滑重启比子进程，给主进城，就会平滑重启所有子进程，kill主进城，或者传递SIGHUP信号，会平滑的kill所有子进程，结束主进城。这样的好处就是主进城只负责监控子进程和自己的状态，运行更稳定，然后后期可以做一些子进程任务处理上限的控制，就可以试进城更稳定高效，信号试用kill -s 传递

升级版：

留作记录

<?php
use controllers\Member\MemberAsynchTask;
use controllers\Member\MemberConf;
use controllers\Log\Tlog;
/**
 * Class MemberTaskCommand
 * 会员积分相关守护任务
 * 多进程守护任务管理
 *
 * 监控脚本只需要监控此主任务正常运行即可
 *
 * kill -s SIGUSR1 | kill -10 manger进程： 平滑重启所有子进程
 * kill | kill -s SIGTERM mange进程：平滑stop所有进程
 * kill | kill -s SIGUSR1 | kill -s SIGTERM 子进程，重启该子进程
 * author zhangcunchao
 * date 2016-11-15
 *
 */
class MemberTaskCommand
{
    const      proccess_num  = 3;//子进程数
    const      proccess_name = 'php-member-task';//进程名前缀，此设置主要为监控脚本提供方便
    protected  $open_debug   = true;//开启debug,警告直接输出，不发邮件
    protected  $_pid;
    protected  $shm_key;      //共享内存key
    protected  $shm_id;
    protected  $max_job_limit= 0; //单个子进程最多处理的任务数，达到此任务就重启此子进程;0 没有限制
    protected  $NOW_DIR;
    protected  $max_try_num  = 3;//队列数据最大尝试次数//超过计入日志

    public function __construct(){
        $this->NOW_DIR =  dirname(__FILE__).'/';
    }
    //入口进程
    public function run(){
        //检查扩展
        if(!extension_loaded('pcntl')){
            print_r("需要安装pcntl扩展支持".PHP_EOL);
            exit();
        }
        if(!extension_loaded('shmop')){
            print_r("需要安装shmop扩展支持".PHP_EOL);
            exit();
        }

        $this->shm_key = ftok(__FILE__, 't');

        /**
         开辟一块共享内存
        int $key , string $flags , int $mode , int $size
        $flags: a:访问只读内存段
            c:创建一个新内存段，或者如果该内存段已存在，尝试打开它进行读写
            w:可读写的内存段
            n:创建一个新内存段，如果该内存段已存在，则会失败
        $mode: 八进制格式 0655
        $size: 开辟的数据大小 字节
         */
        $this->shm_id = shmop_open($this->shm_key, 'c', 0644, 1);
        if($this->shm_id === false){
            echo "开辟共享内存失败".PHP_EOL;
            exit();
        }
        /**
         * 写入数据 数据必须是字符串格式 , 最后一个指偏移量
         * 注意：偏移量必须在指定的范围之内，否则写入不了
         *
         */
        $size = shmop_write($this->shm_id, 1, 0);
        if($size<=0){
            echo "共享内存写入失败".PHP_EOL;
            exit();
        }

        if(self::proccess_num > 0){

            //服务检查
            try{
                $job = MemberAsynchTask::getInstance();
                $job->system_check();
            }catch (\Exception $e){
                print_r($e->getMessage());
                exit();
            }

            //启动进程
            for ($i=0; $i<self::proccess_num; $i++)
            {
                $this->parentProccess();

            }
            //SIGUSR1 留给用户使用,此处做reload平滑重启使用
            pcntl_signal(SIGUSR1, function($this){
                $this->sig_handler(SIGUSR1);
            });
            // 程序结束(terminate)信号, 与SIGKILL不同的是该信号可以被阻塞和 处理. 通常用来要求程序自己正常退出.  shell命令kill缺省产生这 个信号.
            pcntl_signal(SIGTERM, function($this){
                $this->sig_handler(SIGTERM);
            });
            //终止进程  终端线路挂断
            pcntl_signal(SIGHUP, function($this){
                $this->sig_handler(SIGHUP);
            });
            cli_set_process_title(self::proccess_name . '-manger');

            while (true) {
                // do something
                $pid = pcntl_waitpid(-1, $status, WUNTRACED | WNOHANG);
                if($pid>0){
                    $s = pcntl_wifexited($status);
                    //todo非正常退出报警
                    $this->parentProccess($pid);
                }
                pcntl_signal_dispatch(); // 接收到信号时，调用注册的signalHandler()
                $this->check_kill_all();
                usleep(100000);
            }

        }else{
            exit('proccess_num must > 0');
        }
    }

    //监测是否有严重子进程错误，进行强制进程结束
    function check_kill_all(){
        //读取的范围也必须在申请的内存范围之内,否则失败
        //其他异常强制全部退出
        $status = shmop_read($this->shm_id, 0, 1);
        if(1 != $status){
            shmop_delete($this->shm_id);
            shmop_close($this->shm_id);
            $this->sig_handler(SIGTERM);
        }
    }

    //子进程使用
    function san_kill_all_task(){
        shmop_write($this->shm_id, 0, 0);
    }

    function sig_handler($signo){
        switch ($signo) {
            case SIGUSR1:
                foreach($this->_pid as $key){
                    posix_kill($key,SIGUSR1);
                }
                echo date('Y-m-d H:i:s')." [manage] [SIGUSR1] all work reload\n";
                break;
            case SIGTERM:
                foreach($this->_pid as $key){
                    posix_kill($key,SIGUSR1);
                }
                echo date('Y-m-d H:i:s')." [manage] [SIGTERM] stop\n";
                die();
                break;
            case SIGHUP:
                echo date('Y-m-d H:i:s')." [manage] [SIGHUP] linux clinet exit\n";
                break;
        }
    }

    function sig_san_handler($signo){
        switch ($signo) {
            case SIGUSR1:
                echo date('Y-m-d H:i:s')." [work] [SIGUSR1] work die\n";
                die();
                break;
            case SIGTERM:
                echo date('Y-m-d H:i:s')." [work] [SIGTERM] work die\n";
                die();
                break;
            case SIGHUP:
                echo date('Y-m-d H:i:s')." [work] [SIGHUP] linux clinet exit\n";
                break;
        }
    }

    //父进程
    public function parentProccess($id=''){
        if(!empty($id)){
            unset($this->_pid[$id]);
        }
        $pid  = pcntl_fork();
        if($pid == -1)
        {
            return false;
        }
        else
        {
            if($pid)
            {
                #父进程获得子进程的pid，存入数组
                $this->_pid[$pid] = $pid;
                return true;
            }
            else
            {
                //开始发送，子进程执行完自己的任务后，退出。
                $this->sanProccess();
                exit;
            }
        }
    }

    //子进程
    public function sanProccess(){

        //信号绑定
        cli_set_process_title(self::proccess_name . '-work');
        echo date('Y-m-d H:i:s')." work proccess start\n";
        pcntl_signal(SIGUSR1, function($this){
            $this->sig_san_handler(SIGUSR1);
        });
        pcntl_signal(SIGHUP, function($this){
            $this->sig_san_handler(SIGHUP);
        });
        pcntl_signal(SIGTERM, function($this){
            $this->sig_san_handler(SIGTERM);
        });

        if($this->max_job_limit){
            $san_job_i = 0;
        }
        //job分发
        $job = MemberAsynchTask::getInstance();
        while(1){
            try{
                $job->queue_ping();
            }catch (\Exception $e){
                //重试一次
                try{
                    if(10002 == $e->getCode()){
                        echo '重试一次连接'.PHP_EOL;
                        $job->queue_ping();
                    }

                }catch (\Exception $e){
                    //队列服务已不可用
                    if(10002 == $e->getCode()){
                        //kill 所有子进程，报警
                        $this->san_kill_all_task();
                        $job->waring_email(var_export($e,true));
                        sleep(3);
                    }

                }
            }
            //初始容器，避免循环影响
            $job_data = array();
            $job_key  = '';
            try{
                $jrs = $job->queue_pop(MemberConf::$_MEMBER_QUEUE_LIST);
                if($jrs && !empty($jrs[0]) && !empty($jrs[1])){
                    $job_data = json_decode($jrs[1],true);
                    $job_key  = $jrs[0];
                    //分发任务
                    switch($job_key){
                        //积分更新任务
                        case MemberConf::$_MEMBER_QUEUE_LIST['evente_member_integral_queue']:
                            require_once $this->NOW_DIR.'MemberIntegralJob.php';
                            $MemberIntegralJob = new MemberIntegralJob();
                            $MemberIntegralJob->integralChangeJob($job_data);
                            break;
                        //用户会员激活任务
                        case MemberConf::$_MEMBER_QUEUE_LIST['evente_member_user_activate']:
                            require_once $this->NOW_DIR.'MemberIntegralJob.php';
                            $MemberIntegralJob = new MemberIntegralJob();
                            $MemberIntegralJob->userMemberActivate($job_data);
                            break;
                        //成长值更新任务
                        case MemberConf::$_MEMBER_QUEUE_LIST['evente_member_growth_queue']:
                            require_once $this->NOW_DIR.'MemberGrowthJob.php';
                            $MemberGrowthJob = new MemberGrowthJob();
                            $MemberGrowthJob->growthChangeJob($job_data);
                            break;
                        //主办会员等级规则修改
                        case MemberConf::$_MEMBER_QUEUE_LIST['evente_member_rule_up_parent']:
                            require_once $this->NOW_DIR.'MemberGrowthJob.php';
                            $MemberGrowthJob = new MemberGrowthJob();
                            $MemberGrowthJob->orgLevelRuleUpJob($job_data);
                            break;
                        //等级reset任务
                        case MemberConf::$_MEMBER_QUEUE_LIST['evente_member_user_level_reset']:
                            require_once $this->NOW_DIR.'MemberGrowthJob.php';
                            $MemberGrowthJob = new MemberGrowthJob();
                            $MemberGrowthJob->userLevelResetJob($job_data);
                            break;
                        default:
                            throw new \Exception("未定义的守护任务处理case:".$job_key,10002);
                    }
                }
            }catch (\Exception $e){
                $code = $e->getCode();
                //统一重试，失败记录重试队列
                if(!empty($job_data)&&!empty($job_key)){
                    if(isset($job_data['_job_error_num'])){
                        $job_data['_job_error_num']=$job_data['_job_error_num']+1;
                    }else{
                        $job_data['_job_error_num'] = 1;
                    }
                }
                $mes = '本数据第'.$job_data['_job_error_num'].'次错误尝试'.PHP_EOL;
                //重试上限
                if($this->max_try_num <= $job_data['_job_error_num']){
                    $mes .= '数据已达处理次数上限，重试结束，记入日志'.PHP_EOL;
                    $log = new Tlog(MemberConf::QUEUE_EXCEPTION_FILE,MemberConf::LOG_DIR);
                    unset($job_data['_job_error_num']);
                    $ld = array(
                        'k' => $job_key,
                        'v' => $job_data,
                    );
                    $m = '[error_code:'.$code.' |'.json_encode($ld).'|]';
                    $log->debugwriteLog($m);

                    //邮件
                    switch($code){
                        case 10002:
                            //任务返回到队列
                            $mes .= '严重错误,需要紧急修复，子进程已全部kill'.PHP_EOL;
                            break;
                        case 10003:
                            $mes .= '严重错误,需要紧急修复，单任务重入队列'.PHP_EOL;
                            break;
                        default:
                            $mes .= '非紧急异常,异常数据已记录到日志：'. MemberConf::TASK_EXCEPTION_FILE . PHP_EOL;
                    }
                    $mes .= var_export($e,true);
                    if($this->open_debug){
                        print_r($mes);
                    }else{
                        $job->waring_email($mes,'异步任务会员积分报警邮件');
                    }

                }else{
                    //统一重入队列
                    if(!empty($job_data)&&!empty($job_key)){
                        $job->queue_push($job_key,json_encode($job_data));
                    }

                }

                //容错处理
                switch($code){
                    case 10002:
                        //严重致命错误，需要停止所有进程运行
                        //任务返回到队列
                        //kill 所有子进程，报警
                        $this->san_kill_all_task();
                        sleep(3);
                        break;
                }
            }

            //控制单个进程最大任务数
            if($this->max_job_limit){
                ++$san_job_i;
                if($this->max_job_limit <= $san_job_i){
                    echo 'work job limit reload'.PHP_EOL;
                    exit();
                }
            }
            pcntl_signal_dispatch();
        }
    }

}

$obj = new MemberTaskCommand();
$obj->run();

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

<?php

use controllers\Member\MemberAsynchTask;

use controllers\Member\MemberConf;

use controllers\Log\Tlog;

/**

* Class MemberTaskCommand

* 会员积分相关守护任务

* 多进程守护任务管理

* 监控脚本只需要监控此主任务正常运行即可

* kill -s SIGUSR1 | kill -10 manger进程：平滑重启所有子进程

* kill | kill -s SIGTERM mange进程：平滑stop所有进程

* kill | kill -s SIGUSR1 | kill -s SIGTERM 子进程，重启该子进程

* author zhangcunchao

* date 2016-11-15

class MemberTaskCommand

{

const proccess_num = 3;//子进程数

const proccess_name = 'php-member-task';//进程名前缀，此设置主要为监控脚本提供方便

protected $open_debug = true;//开启debug,警告直接输出，不发邮件

protected $_pid;

protected $shm_key; //共享内存key

protected $shm_id;

protected $max_job_limit= 0; //单个子进程最多处理的任务数，达到此任务就重启此子进程;0 没有限制

protected $NOW_DIR;

protected $max_try_num = 3;//队列数据最大尝试次数//超过计入日志

public function __construct(){

$this->NOW_DIR = dirname(__FILE__).'/';

}

//入口进程

public function run(){

//检查扩展

if(!extension_loaded('pcntl')){

print_r("需要安装pcntl扩展支持".PHP_EOL);

exit();

}

if(!extension_loaded('shmop')){

print_r("需要安装shmop扩展支持".PHP_EOL);

exit();

}

$this->shm_key = ftok(__FILE__, 't');

/**

开辟一块共享内存

int $key , string $flags , int $mode , int $size

$flags: a:访问只读内存段

c:创建一个新内存段，或者如果该内存段已存在，尝试打开它进行读写

w:可读写的内存段

n:创建一个新内存段，如果该内存段已存在，则会失败

$mode: 八进制格式 0655

$size: 开辟的数据大小字节

$this->shm_id = shmop_open($this->shm_key, 'c', 0644, 1);

if($this->shm_id === false){

echo "开辟共享内存失败".PHP_EOL;

exit();

}

/**

* 写入数据数据必须是字符串格式 , 最后一个指偏移量

* 注意：偏移量必须在指定的范围之内，否则写入不了

$size = shmop_write($this->shm_id, 1, 0);

if($size<=0){

echo "共享内存写入失败".PHP_EOL;

exit();

}

if(self::proccess_num > 0){

//服务检查

try{

$job = MemberAsynchTask::getInstance();

$job->system_check();

}catch (\Exception $e){

print_r($e->getMessage());

exit();

}

//启动进程

for ($i=0; $i<self::proccess_num; $i++)

{

$this->parentProccess();

}

//SIGUSR1 留给用户使用,此处做reload平滑重启使用

pcntl_signal(SIGUSR1, function($this){

$this->sig_handler(SIGUSR1);

});

// 程序结束(terminate)信号, 与SIGKILL不同的是该信号可以被阻塞和处理. 通常用来要求程序自己正常退出. shell命令kill缺省产生这个信号.

pcntl_signal(SIGTERM, function($this){

$this->sig_handler(SIGTERM);

});

//终止进程终端线路挂断

pcntl_signal(SIGHUP, function($this){

$this->sig_handler(SIGHUP);

});

cli_set_process_title(self::proccess_name . '-manger');

while (true) {

// do something

$pid = pcntl_waitpid(-1, $status, WUNTRACED | WNOHANG);

if($pid>0){

$s = pcntl_wifexited($status);

//todo非正常退出报警

$this->parentProccess($pid);

}

pcntl_signal_dispatch(); // 接收到信号时，调用注册的signalHandler()

$this->check_kill_all();

usleep(100000);

}

}else{

exit('proccess_num must > 0');

}

//监测是否有严重子进程错误，进行强制进程结束

function check_kill_all(){

//读取的范围也必须在申请的内存范围之内,否则失败

//其他异常强制全部退出

$status = shmop_read($this->shm_id, 0, 1);

if(1 != $status){

shmop_delete($this->shm_id);

shmop_close($this->shm_id);

$this->sig_handler(SIGTERM);

}

//子进程使用

function san_kill_all_task(){

shmop_write($this->shm_id, 0, 0);

}

function sig_handler($signo){

switch ($signo) {

case SIGUSR1:

foreach($this->_pid as $key){

posix_kill($key,SIGUSR1);

}

echo date('Y-m-d H:i:s')." [manage] [SIGUSR1] all work reload\n";

break;

case SIGTERM:

foreach($this->_pid as $key){

posix_kill($key,SIGUSR1);

}

echo date('Y-m-d H:i:s')." [manage] [SIGTERM] stop\n";

die();

break;

case SIGHUP:

echo date('Y-m-d H:i:s')." [manage] [SIGHUP] linux clinet exit\n";

break;

}

function sig_san_handler($signo){

switch ($signo) {

case SIGUSR1:

echo date('Y-m-d H:i:s')." [work] [SIGUSR1] work die\n";

die();

break;

case SIGTERM:

echo date('Y-m-d H:i:s')." [work] [SIGTERM] work die\n";

die();

break;

case SIGHUP:

echo date('Y-m-d H:i:s')." [work] [SIGHUP] linux clinet exit\n";

break;

}

//父进程

public function parentProccess($id=''){

if(!empty($id)){

unset($this->_pid[$id]);

}

$pid = pcntl_fork();

if($pid == -1)

{

return false;

}

else

{

if($pid)

{

#父进程获得子进程的pid，存入数组

$this->_pid[$pid] = $pid;

return true;

}

else

{

//开始发送，子进程执行完自己的任务后，退出。

$this->sanProccess();

exit;

}

//子进程

public function sanProccess(){

//信号绑定

cli_set_process_title(self::proccess_name . '-work');

echo date('Y-m-d H:i:s')." work proccess start\n";

pcntl_signal(SIGUSR1, function($this){

$this->sig_san_handler(SIGUSR1);

});

pcntl_signal(SIGHUP, function($this){

$this->sig_san_handler(SIGHUP);

});

pcntl_signal(SIGTERM, function($this){

$this->sig_san_handler(SIGTERM);

});

if($this->max_job_limit){

$san_job_i = 0;

}

//job分发

$job = MemberAsynchTask::getInstance();

while(1){

try{

$job->queue_ping();

}catch (\Exception $e){

//重试一次

try{

if(10002 == $e->getCode()){

echo '重试一次连接'.PHP_EOL;

$job->queue_ping();

}

}catch (\Exception $e){

//队列服务已不可用

if(10002 == $e->getCode()){

//kill 所有子进程，报警

$this->san_kill_all_task();

$job->waring_email(var_export($e,true));

sleep(3);

}

//初始容器，避免循环影响

$job_data = array();

$job_key = '';

try{

$jrs = $job->queue_pop(MemberConf::$_MEMBER_QUEUE_LIST);

if($jrs && !empty($jrs[0]) && !empty($jrs[1])){

$job_data = json_decode($jrs[1],true);

$job_key = $jrs[0];

//分发任务

switch($job_key){

//积分更新任务

case MemberConf::$_MEMBER_QUEUE_LIST['evente_member_integral_queue']:

require_once $this->NOW_DIR.'MemberIntegralJob.php';

$MemberIntegralJob = new MemberIntegralJob();

$MemberIntegralJob->integralChangeJob($job_data);

break;

//用户会员激活任务

case MemberConf::$_MEMBER_QUEUE_LIST['evente_member_user_activate']:

require_once $this->NOW_DIR.'MemberIntegralJob.php';

$MemberIntegralJob = new MemberIntegralJob();

$MemberIntegralJob->userMemberActivate($job_data);

break;

//成长值更新任务

case MemberConf::$_MEMBER_QUEUE_LIST['evente_member_growth_queue']:

require_once $this->NOW_DIR.'MemberGrowthJob.php';

$MemberGrowthJob = new MemberGrowthJob();

$MemberGrowthJob->growthChangeJob($job_data);

break;

//主办会员等级规则修改

case MemberConf::$_MEMBER_QUEUE_LIST['evente_member_rule_up_parent']:

require_once $this->NOW_DIR.'MemberGrowthJob.php';

$MemberGrowthJob = new MemberGrowthJob();

$MemberGrowthJob->orgLevelRuleUpJob($job_data);

break;

//等级reset任务

case MemberConf::$_MEMBER_QUEUE_LIST['evente_member_user_level_reset']:

require_once $this->NOW_DIR.'MemberGrowthJob.php';

$MemberGrowthJob = new MemberGrowthJob();

$MemberGrowthJob->userLevelResetJob($job_data);

break;

default:

throw new \Exception("未定义的守护任务处理case:".$job_key,10002);

}

}catch (\Exception $e){

$code = $e->getCode();

//统一重试，失败记录重试队列

if(!empty($job_data)&&!empty($job_key)){

if(isset($job_data['_job_error_num'])){

$job_data['_job_error_num']=$job_data['_job_error_num']+1;

}else{

$job_data['_job_error_num'] = 1;

}

$mes = '本数据第'.$job_data['_job_error_num'].'次错误尝试'.PHP_EOL;

//重试上限

if($this->max_try_num <= $job_data['_job_error_num']){

$mes .= '数据已达处理次数上限，重试结束，记入日志'.PHP_EOL;

$log = new Tlog(MemberConf::QUEUE_EXCEPTION_FILE,MemberConf::LOG_DIR);

unset($job_data['_job_error_num']);

$ld = array(

'k' => $job_key,

'v' => $job_data,

);

$m = '[error_code:'.$code.' |'.json_encode($ld).'|]';

$log->debugwriteLog($m);

//邮件

switch($code){

case 10002:

//任务返回到队列

$mes .= '严重错误,需要紧急修复，子进程已全部kill'.PHP_EOL;

break;

case 10003:

$mes .= '严重错误,需要紧急修复，单任务重入队列'.PHP_EOL;

break;

default:

$mes .= '非紧急异常,异常数据已记录到日志：'. MemberConf::TASK_EXCEPTION_FILE . PHP_EOL;

}

$mes .= var_export($e,true);

if($this->open_debug){

print_r($mes);

}else{

$job->waring_email($mes,'异步任务会员积分报警邮件');

}

}else{

//统一重入队列

if(!empty($job_data)&&!empty($job_key)){

$job->queue_push($job_key,json_encode($job_data));

}

//容错处理

switch($code){

case 10002:

//严重致命错误，需要停止所有进程运行

//任务返回到队列

//kill 所有子进程，报警

$this->san_kill_all_task();

sleep(3);

break;

}

//控制单个进程最大任务数

if($this->max_job_limit){

++$san_job_i;

if($this->max_job_limit <= $san_job_i){

echo 'work job limit reload'.PHP_EOL;

exit();

}

pcntl_signal_dispatch();

}

$obj = new MemberTaskCommand();

$obj->run();

程序本天成，妙手偶得之！我们只是代码的搬运工！

转载请注明：http://www.521php.com/archives/2017/

php清除文章样式 css、js、html样式 2012-09-18 博采众长/我爱PHP

php截取中文字符 2012-07-06 我爱PHP

织梦内置标签小优化之四—dede:channelartlist标签 2013-04-03 小功能/总结/我爱PHP/推荐文章

本站获取指定网站favicon.ico网站图标接口升级(v3) 2013-04-16 小功能/我爱PHP

thrift接口框架小解 2014-04-25 安全/总结/我爱PHP

php捕捉各大搜索引擎的蜘蛛、爬虫记录 2012-12-28 小功能/我爱PHP/推荐文章

WdatePicker日历控件使用方法 2013-03-27 博采众长/总结

php正则修改配置文件 2012-07-06 我爱PHP

Linux的chattr与lsattr命令 2015-07-28 Linux/博采众长/总结

php fseek读取超过内存大小文件 2016-09-23 总结/我爱PHP/推荐文章

目前有8 条留言

2016年11月08日下午 10:11 ljw | 引用 | #1

看不动啊。。。。

[回复]
- 2018年09月13日下午 6:29 JCJC | 引用 | #2
  
  感谢分享~~~~~~~~~~~~~
  
  [回复]
2016年11月17日上午 9:03 增达网 | 引用 | #3

你的博客就像冬天里的一把火！

[回复]
2016年12月02日上午 8:15 增达网 | 引用 | #4

今天办公室里几个男女在讨论用避孕套的好处和坏处，男方坚决抵制避孕套，女方大力支持使用避孕套，双方各持一词不分伯仲，这时我们办公室一个大姐问我平日用套套吗？我说:我……我…我都是包皮打个结。当时讨论就结束了。

[回复]
2018年09月16日上午 12:35 头条 | 引用 | #5

文章不错非常喜欢

[回复]
2018年11月28日下午 1:12 百万链 | 引用 | #6

文章写的不错，能否交换友链啊，站长！

[回复]
2018年12月06日上午 12:16 新闻 | 引用 | #7

文章不错非常喜欢

[回复]
2021年02月16日下午 1:35 资源导航 | 引用 | #8

这篇文章写的真棒！

[回复]

纯php做类似php-fpm多进程管理服务

目前有8 条留言

发表评论

橱窗

文章分类

进入本站推广

本站接口

推荐文章

学习之所得

生活之所感

标签云

最新留言

文章归档