MySQLi不会在pthreads worker中建立连接 [英] MySQLi wont establish connection within a pthreads worker

问题描述

为什么在使用 pthread

Why cannot a mysqli connection be created when using a pthread worker in PHP? When establishing a mysqli and attempting to execute a query, will generatate an error

PHP警告:mysqli :: query():无法在其中获取mysqli 第47行的W:\ workspace \ Sandbox \ application.php

PHP Warning: mysqli::query(): Couldn't fetch mysqli in W:\workspace\Sandbox\application.php on line 47

这是我的脚本

<?php
    // Application

    // Exceptions
    class MySQLWorkerException extends Exception {}
    class MySQLConnectionException extends Exception {}
    class MySQLQueryException extends Exception {}

    // Worker
    class SQLWorker extends Worker {
        // Class Fields
        protected   $ready;
        protected   $name;

        // MySQLi Object
        protected   $link;

        // MySQL Credentials
        private
                    $host,
                    $username,
                    $password,
                    $database,
                    $port;

        // Constructor
        function __construct($host, $username, $password, $database, $port = 3306) {
            // Set Credentials
            $this->host     = $host;
            $this->username = $username;
            $this->password = $password;
            $this->database = $database;
            $this->port     = $port;
        }

        // Methods
        public function run() {
            // Create MySQL link
            $this->link = new mysqli(
                $this->host,
                $this->username,
                $this->password,
                $this->database,
                $this->port
            );

            $this->link->query("SELECT 1");
            //var_dump($this->link);
            exit;

            // Check for connection error
            if($this->link->connect_errno) {
                throw new MySQLConnectionException('(' . $this->link->connect_errno . '): ' . $this->link->connect_error, E_WARNING);
            }

            //$this->name = sprintf("%s (%lu)", __CLASS__, $this->getThreadId());
            $this->isReady(true);
        }

        public function getConnection() {
            return $this->link;
        }

        protected function isReady($flag = null) { 
            if(is_null($flag)) {
                return ($this->ready);
            } else {
                $this->ready = $flag;
            }
        }
    }

    // MySQL Method Classes
    // STUB: class SQLQuery extends Stackable {}
    // STUB: class SQLResult extends Stackable {}
    // STUB: class SQLPrepare extends Stackable {}

    class SQLQuery extends Stackable {
        // Class Fields
        protected $complete = false;

        // SQL Query
        private $query;
        private $resultmode;

        // SQL Result
        private $result;

        // Constructor
        function __construct($query, $resultmode = MYSQLI_STORE_RESULT) {
            $this->query = $query;
            $this->resultmode = $resultmode;
        }

        // Methods
        public function run() {
            if($this->worker->isReady()) {
                if(!$this->result = mysqli_query($this->worker->link, $this->query, $this->resultmode)) {
                    throw new MySQLQueryException('(' . $this->worker->link->errno . '): ' . $this->worker->link->error, E_WARNING);
                }
            } else {
                throw new MySQLWorkerException('Worker is not ready', E_ERROR);
            }

            $this->complete = true;
            $this->notify();
        }

        public function isComplete() {
            return ($this->complete);
        }

        public function GetResults() {
            return $this->result;
        }

        public function ReadResults() {
            $response = null;
            while($row = $this->result->fetch_assoc()) {
                $response[] = $row;
            }

            if($translateRows) {
                if(count($response) == 1) {
                    $response = end($response);

                    if(count($response) == 1) {
                        $response = end($response);
                    }
                }
            }

            $this->result->close();
            return $response;
        }
    }


    // Program
    $config = [
        'host'      => '127.0.0.1',
        'username'  => 'root',
        'password'  => '',
        'database'  => 'testdatabase',
        'port'      => 3306
    ];

    // MySQL Worker
    $worker = new SQLWorker($config['host'], $config['username'], $config['password'], $config['database'], $config['port']);
    $worker->start();

    // Create Query
    $query = new SQLQuery("SELECT username FROM users WHERE id = 1 LIMIT 1");

    // Give the query to the worker
    // $worker->stack($query);

    /* In reality we would have done a lot of shit here while the query executes. */

    // If the query isnt complete, wait up.
    $query->wait();

    // Holy shit, lets var_dump this bad boy
    // var_dump($query);

    // Shutdown the Worker Thread. We don't want any memleaks!
    $worker->shutdown();

推荐答案

以下是一些有效的代码...

Here is some working code ...

<?php

class SQLWorker extends Worker {

    public function __construct($hostname, $username, $password, $database, $port = 3306) {
        $this->hostname = $hostname;
        $this->username = $username;
        $this->password = $password;
        $this->database = $database;
    }

    public function run() {
        /* the static class scope acts as a kind of thread local storage for this class */
        self::$connection = new MySQLi
            ($this->hostname, $this->username, $this->password, $this->port);
    }

    public function getConnection() {
        return self::$connection;
    }

    protected $hostname;
    protected $username;
    protected $password;
    protected $database;
    protected $port;

    protected static $connection;
}

class SQLTask extends Threaded {
    public function __construct($sql) {
        $this->sql = $sql;
    }

    public function run() {
        /* fetch static (thread local) connection */
        $link = $this->worker->getConnection();

        if ($link) {
            /* execute query, keep result in local scope */
            $result = $link->query($this->sql);

            /* build up results as normal array */
            while (($row = $result->fetch_assoc())) {
                $rows[] = $row;
            }
        }

        /* store results in a fetchable form */
        $this->rows = $rows;
    }

    public function getResults() {
        return $this->rows;
    }
}

$pool = new Pool(4, 
    SQLWorker::class, 
    ["localhost", "username", "password", "database"]);

$pool->submit(
    $task = new SQLTask("SELECT 1"));

$pool->shutdown();

/* you have your normal array of data here */
var_dump($task->getResults());

基本问题是，像mysqli对象这样的复杂对象不能很好地实现对pthreads对象的安全性，因此不要将此类对象写入对象范围，也不要将它们保持在静态(线程局部)，方法或全局范围内(全球人太恐怖了.

The basic problem is that complex objects like mysqli objects will not play nice with safety implemented on pthreads objects, therefore do not write such objects to the object scope, keep them in static (thread local), method, or global scope (globals are horrible).

除此之外，您对同步的使用也不是明智的.您应该只等要，如下所述: https://gist.github .com/krakjoe/6437782

In addition to this, your use of synchronization is not sane. You should only ever wait for something, as explained: https://gist.github.com/krakjoe/6437782

这篇关于MySQLi不会在pthreads worker中建立连接的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！