[Sw] 使用 Swoole Server task/协程 处理大数据量异步任务时注意

关于 Buffered Query 和 Unbuffered Query:http://www.php.net/manual/zh/mysqlinfo.concepts.buffering.php

对于结果集小的查询,一般就开启 Buffered Query 一次取回(fetchAll);

对于结果集很大的查询,可以开启 Unbuffered Query 来遍历资源一条条 fetch,避免撑爆客户端内存;

PDO 属性设置:http://php.net/manual/zh/pdo.setattribute.php

其它解决方案:

1. 高频投递(依赖进程数),少量处理(每批次数据),可以自己用 Process 实现进程池处理队列任务,或者使用自带的 task 功能。

2. 使用自带 task 功能的情况下,如果 worker 不需要与 task worker 通讯,那么 onTask 不要使用 return 返回数据,减少消耗。

3. worker 使用 task( ) 投递频率必须小于 task 进程数(task_worker_num),可以程序来限制。

  比方说 $taskWorkerNum 是 50,

  任务投递次数累加 $deliverNo,

  onTask 内完成任务时计数 $serv->atomic->add(1),

  完成任务数 $serv->atomic->get() 获得。

  那么在投递之后需要进行判断,投递总数 - 完成数 >= 任务进程数,说明投递次数满了,暂停一会儿,保证 task 进程不是满负荷工作。

/**
 * 调度工作
* https://cnblogs.com/farwish
*/ public function onWorkerStart(SwooleServer $server, $workerId) { if ($workerId == 0) { $data = [1, 2, 3]; foreach ($data as $item) { // 限流与投递 while (($server->deliverNo - $server->atomic->get()) >= $this->taskWorker) { echo "等待空闲 task 进程 "; sleep(1); } $server->task($item); $server->deliverNo++; } // 任务结束后退出 server while (true) { if ($server->deliverNo == $server->atomic->get()) { $server->shutdown(); } sleep(1); } } }

服务初始化部分:

    public function initTaskServer()
    {
        $server = new SwooleServer('0.0.0.0');

        $server->atomic = new SwooleAtomic(0);
        $server->deliverNo = 0;

        $server->set([
            'worker_num' => 1,
            'task_worker_num' => $this->taskWorkerNum,
            'task_ipc_mode' => 1,
            'task_max_request' => 5000,
        ]);

        $server->on('workerStart', [$this, 'onWorkerStart']);
        $server->on('task', [$this, 'onTask']);
        $server->on('receive', [$this, 'onReceive']);
        $server->on('finish', [$this, 'onFinish']);

        $server->start();
    }

    protected function onWorkerStart(Server $server, $workerId)
    {
    }

    protected function onTask(Server $server, $taskId, $fromId, $data)
    {
    }

    protected function onReceive(Server $server, $fd, $reactorId, $data)
    {
    }

    protected function onFinish(Server $server, $taskId, $data)
    {
    }

4. 不使用 server 和 task 多进程的情况,利用 swoole 协程中的 channel 实现 producer、consumer 模式,生产者 unbuffer query 持续 push 数据到通道,消费者持续 pop 消费;生产者没有数据时可退出,消费者检测到生产者退出后也随即退出。

缺点是在复杂场景下(比如多层查询再加循环处理),编程会比较困难,比如:等待所有子协程结束的功能(WaitGroup)、多 consumer 的场景,需要自己封装很多组件。

其它:

多进程、多协程的情况下,需要配合使用数据库连接池,因为数据库并发连接数资源有限。

多进程只是利用到了多核,计算密集型场景有优势;协程并发相比更轻量,单进程内利用I/O切换实现并发,适合IO密集型场景。 

Course:http://www.yzmedu.com/course/330

Docs:https://wiki.swoole.com/wiki/page/481.html

Link:https://www.cnblogs.com/farwish/p/10242294.html

原文地址:https://www.cnblogs.com/farwish/p/10242294.html