Codes

trait MultiProcessTrait { protected $processes = []; protected $maxFork = 14; /** * @param $action \Closure * @param array $params */ public function fork($action, $params = []) { $this->processes[] = [ "action" => $action, "params" => $params ]; } protected $runningProcess = []; public function waitProcessRun() { while (count($this->runningProcess) > 0) { $mypid = pcntl_waitpid(-1, $status, WNOHANG); foreach ($this->runningProcess as $key => $pid) { if ($mypid == $pid || $mypid == -1) { echo "child $key completed\n"; unset($this->runningProcess[$key]); //判断是否还有未fork进程 $this->runOne(); } } } } public function runOne() { $process = array_shift($this->processes); if ($process) { $pid = pcntl_fork(); if ($pid == -1) { die("could not fork"); } elseif ($pid) { $this->runningProcess[$pid] = $pid; echo "create child: $pid \n"; } else { //执行子进程 call_user_func_array($process['action'], $process['params']); exit;// 一定要注意退出子进程,否则pcntl_fork() 会被子进程再fork,带来处理上的影响。 } } } public function runProcess() { if (empty($this->processes)) { return; } for ($i = 0; $i < $this->maxFork; $i++) { $this->runOne(); } $this->waitProcessRun(); } }

route.php Route::group(['prefix' => 'fapi/', 'middleware' => ['web']], function () { Route::any('{slug}', 'Cloud\Rent\Classes\Api\ApiRouter@route')->where('slug', '(.*)?'); }); ApiRouter.php <?php namespace Cloud\Rent\Classes\Api; use App; use Illuminate\Routing\Controller as ControllerBase; use October\Rain\Router\Helper as RouterHelper; use Illuminate\Support\Str; /** * Class ApiRouter * API例子：在Now\Youpin\Api下建类Auth，里面有public function hello()，可以通过http://host/api/auth/hello * @package Now\Youpin\Classes\Api */ class ApiRouter extends ControllerBase { public function __construct() { header('Access-Control-Allow-Origin: *'); header('Access-Control-Allow-Methods:*'); header('Access-Control-Allow-Headers:*'); header("Access-Control-Allow-Headers:DNT,X-Mx-ReqToken,Keep-Alive,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type, Accept-Language, Origin, Accept-Encoding"); } /** * @var string Allows early access to page action. */ public static $action; /** * @var array Allows early access to page parameters....

自定义的一个小功能,线程池处理,用法很简单 def handleThread(a,b): pass main = ThreadPool(5) req = WorkRequest(handleThread, args=[1, 2], kwds={}) main.putRequest(req) #-*-encoding:utf-8-*- ''' Created on 2012-3-9 @summary: 线程池 ''' import sys import threading import queue import traceback import statistics # 定义一些Exception，用于自定义异常处理 class NoResultsPending(Exception): """All works requests have been processed""" pass class NoWorkersAvailable(Exception): """No worket threads available to process remaining requests.""" pass def _handle_thread_exception(request, exc_info): """默认的异常处理函数，只是简单的打印""" traceback.print_exception(*exc_info) #classes class WorkerThread(threading.Thread): """后台线程，真正的工作线程，从请求队列(requestQueue)中获取work， 并将执行后的结果添加到结果队列(resultQueue)""" def __init__(self,requestQueue,resultQueue,poll_timeout=5,**kwds): threading.Thread.__init__(self,**kwds) '''设置为守护进行''' self.setDaemon(True) self._requestQueue = requestQueue self....