Yii2 基于PCNTL实现多进程导出CSV文件

网上一个大神写的，感觉挺有意思，记录一下。

use Yii;
use yii\base\Exception;
use yii\console\Controller;
use yii\helpers\Console;
use yii\helpers\FileHelper;

class ExportController extends Controller
{
    protected $csvPath;
    protected $user;
    protected $size = 1000;
    protected $step = 200000;

    public function beforeAction($action)
    {
        parent::beforeAction($action);
        $this->csvPath = dirname(Yii::$app->basePath) . '/csv';
        $this->user = 'user';
        return true;
    }

    //多进程的脚本
    public function actionSend($count)
    {
        if (!$count) exit('count is zero');
        $taskStartTime = microtime(true);
        //创建目录
        if (!file_exists($this->csvPath)) {
            try {
                FileHelper::createDirectory($this->csvPath, '0777', true);
            } catch (Exception $e) {
                $this->stdout($e->getMessage(), Console::BG_RED);
                return 1;
            }
        }
        $totalNum = ceil($count / $this->step);
        $childs = [];
        for ($i = 0; $i < $totalNum; $i++) {
            $pid = pcntl_fork();
            if ($pid == -1) {
                $this->stdout('Could not fork', Console::BG_RED);
                return 1;
            } elseif ($pid) {
                $this->stdout("I'm the Parent $i", Console::BG_YELLOW);
                $childs[] = $pid;
            } else {
                //子进程处理业务
                $this->handelChildProcess($i, $count);
            }
        }

        while (count($childs) > 0) {
            foreach ($childs as $key => $pid) {
                $res = pcntl_waitpid($pid, $status, WNOHANG);
                //-1代表error, 大于0代表子进程已退出,返回的是子进程的pid,非阻塞时0代表没取到退出子进程
                if ($res == -1 || $res > 0) {
                    $this->stdout("$key=> $pid", Console::BG_YELLOW);
                    unset($childs[$key]);
                }
            }
            sleep(1);
        }
        $lastTime = $this->getElapsedTime($taskStartTime);
        Yii::info("totalLastTime|" . $lastTime, __METHOD__);
        $this->stdout("success|$lastTime", Console::BG_GREEN);
        return 0;
    }

    public function getRows($start)
    {
        $size = $this->size;
        $end = $start + $size;
        $userList = Yii::$app->db->createCommand("SELECT field1,field2,field3,field4, field5  FROM `user` where (field4=2 or field4=4) LIMIT " . $size . " OFFSET " . $start)->queryAll();
        foreach ($userList as $value) {
            yield $value;
        }
    }

    public function handelChildProcess($processKey, $totalCount)
    {
        $this->stdout("process $processKey start", Console::BG_GREEN);
        $taskStartTime = microtime(true);
        $pageTotal = ceil($this->step / $this->size);
        for ($i = 1; $i <= $pageTotal; $i++) {
            //计算起始位置
            $start = $processKey * $this->step + ($i - 1) * $this->size;
            if ($start > $totalCount) {
                $lastTime = $this->getElapsedTime($taskStartTime);
                Yii::info("lastTime|process" . $processKey . $lastTime, __METHOD__);
                $this->stdout("process $processKey end", Console::BG_YELLOW);
                $this->stdout("超过总数了", Console::BG_YELLOW);
                return 4;
            }
            $userList = $this->getRows($start);
            foreach ($userList as $key => $value) {
                $this->writeRow($value, $this->csvPath . "/" . $this->user . $processKey . ".csv");
            }
            sleep(1);
        }
        $lastTime = $this->getElapsedTime($taskStartTime);
        Yii::info("lastTime|process" . $processKey . '|' . $lastTime, __METHOD__);
        $this->stdout("process $processKey end", Console::BG_BLUE);
        $this->stdout($lastTime, Console::BG_BLUE);
        return 0;
    }

    public function getElapsedTime($startTime)
    {
        $endTime = microtime(true);
        $elapsedTime = number_format($endTime - $startTime, 4);
        return $elapsedTime;
    }


    public function writeRow($row, $file)
    {
        $row = array_map(function ($v) {
            return iconv('utf-8', "gbk//IGNORE", $v);
        }, $row);
        $handle = fopen($file, 'a');
        fputcsv($handle, $row);
    }
}

测试导入

php yii export/send 2000000

原文地址: https://www.yiichina.com/code/1863