[玩具代码]swoole多进程导出大量数据

说明

针对大量数据导出采用 Swoole 多进程进行 csv 文件写入，代码仅供参考，此方法导出不能保证数据的顺序性。

句柄类

class FileHandler
{
    private $fp;

    public function __construct(string $filePath)
    {
        $this->setFp($filePath);
    }

    /**
     * @return resource|false
     */
    public function getFp()
    {
        return $this->fp;
    }

    /**
     * @param string $filePath
     */
    public function setFp(string $filePath): void
    {
        $this->fp = fopen($filePath, 'w');
    }

    /**
     * @param array $row
     * @return false|int
     * @date 2020/10/13
     */
    public function putCsv(array $row)
    {
        return fputcsv($this->getFp(), $row);
    }
}

Swoole 导出类

class SwooleExporter
{
    /**
     * 每个进程处理的数据条数
     */
    const PER_PROCESS_LIMIT = 1000;

    /**
     * @var FileHandler
     */
    private $writer;

    /**
     * SwooleExporter constructor.
     * @param string $filePath
     */
    public function __construct(string $filePath)
    {
        $this->writer = new FileHandler($filePath);
    }

    /**
     * @param array $head
     * @return SwooleExporter
     * @date 2020/10/13
     */
    public function writeHead(array $head)
    {
        $this->writer->putCsv($head);
        return $this;
    }


    /**
     * @param int $count
     * @param \Closure $write
     * @date 2020/10/13
     */
    public function writeBody(int $count, \Closure $write)
    {
        $groupQty = ceil($count / static::PER_PROCESS_LIMIT);
        for ($i = 0; $i < $groupQty; $i++) {
            $process = new \swoole_process(
                function (\swoole_process $worker) use ($write, $i) {
                    $worker->write($write($i, static::PER_PROCESS_LIMIT, $this->writer));
            }, true);
            $process->start();

            // 将子进程弄成异步
            \swoole_event_add($process->pipe, function ($pipe) use ($process, $i) {
                echo "{$i} process start\n";
                echo $process->read() . PHP_EOL;

                swoole_event_del($pipe);//socket处理完成后，从epoll事件中移除管道
            });
        }

        while ($swooleResult = \swoole_process::wait()) {
            $pid = $swooleResult['pid'];

            echo "process {$pid} exit\n";
        }
    }
}

外部调用

$swooleExporter = new SwooleExporter('test.csv');
$swooleExporter->writeHead(['A1'])
    ->writeBody(10000, function ($currentPage, $pageSize, FileHandler $writer) {
        // 模拟数据库查询
        sleep(1);
        // 根据 $currentPage,$pageSize 进行数据库分页查询
        // 查询完毕后遍历数据写入
        for ($i = 0; $i < 1000; $i++) {
            $writer->putCsv(['data' . ($currentPage * $pageSize + $i)]);
        }

        return $currentPage . ' success';
    });

本作品采用《CC 协议》，转载必须注明作者和本文链接