sports_dapi/datainf/logic/HttpServerRedisToSql.php

<?php
/**
 * Created by PhpStorm.
 * User: Administrator
 * Date: 2019/6/26
 * Time: 12:03
 */

namespace datainf\logic;

use App\Lib\ModelBase;
use App\Http\Response\Response;

use Illuminate\Database\Capsule\Manager as DB;
use datainf\lib\GlobConfigs;
use swoole;

class HttpServerRedisToSql
{
    private $httpserver;
    private $config;
    private $redisonfig = [];
    const   SQLKEY = 'ALLSQLKEY';

    public function __construct($config)
    {
        $this->httpserver = new \swoole\http\server($config['host'], $config['port']);
        $this->httpserver->set($config['sets']);
        $this->config = $config;

        $this->redisonfig = GlobConfigs::getKey('redis');

        $taskWorkingNum = new \swoole\Atomic();
        $this->httpserver->taskWorkingNum = $taskWorkingNum;

        $this->httpserver->on('request', array($this, 'OnRequest'));
        $this->httpserver->on('WorkerStart', array($this, 'onWorkerStart'));
        $this->httpserver->on('task', array($this, 'onTask'));
        $this->httpserver->on('finish', array($this, 'onFinish'));

    }

    public function onWorkerStart($serv, $worker_id)
    {
        if (!$serv->taskworker) {
            ModelBase::init();

            $obj = $this;
            Swoole\Timer::tick(500, function ($id) use ($obj) {
                $obj->BatchSqlRedisToDB();
            });
        }

    }

    //sql批量写入数据库
    private function BatchSqlRedisToDB()
    {
        $redisconfig = $this->redisonfig;
        go(function () use ($redisconfig) {
            $redis = new Swoole\Coroutine\Redis();
            $redis->connect($redisconfig['host'], $redisconfig['port']);

            if ($redis->llen(self::SQLKEY) <= 0) {
                return;
            }

            $batchsql = [];
            $max = 5000;

            for ($i = 0; $i < $max; $i++) {
                $now = $redis->rpop(self::SQLKEY);
                if (empty($now)) {
                    break;
                } else {
                    $batchsql[] = str_replace(";", ":", trim(trim($now), ";"));
                }
            }

            if (!empty($batchsql)) {
                $pdo = DB::getPdo();
                $sqlstr = implode(";", $batchsql);
                $pdo->exec($sqlstr);
            }

            return;
        });
    }


    public function OnRequest($request, $response)
    {
        $data = Response::generate('', 1, '', '任务正常运行.');
        $response->end($data);
        return;
    }

    public function onTask($serv, $task)
    {

    }


    public function onFinish($serv, int $task_id, $data)
    {

    }


    public function start()
    {
        $this->httpserver->start();
    }

}