用于数据库访问的 PHP 并行网络服务器

Question

出于安全原因，我正在编写一个多线程网络服务器，以允许在 PHP 中通过网络进行非直接数据库访问。我遇到的问题是 PHP pthread 已停止使用，取而代之的是一种称为并行 (https://www.php.net/manual/en/book.parallel.php) 的新方法。

我遇到的问题是文档指出所有内容都必须按值传递，并且您不能传递内部对象。好吧，套接字是内部对象以及数据库链接连接器。但是，文档还说包含的文件没有限制。套接字是一种内部类型，我一辈子都想不出如何绕过它。

在传统服务器中，主线程位于传入套接字上并等待客户端连接。当一个连接进来时，主线程产生一个带有套接字信息的新线程并返回监听。根据 PHP 现在正在做的事情，这似乎是不可能的。

我在 Internet 上进行了大量搜索，并找到了使用套接字或数据库连接器联网的 -0- 并行示例。那么，这是可能的还是我需要用另一种语言（如 C++）编写服务器？

建议？

编辑 2022 年 1 月 18 日 @ 23:30 PST（-8:00 UTC）：

这是我迄今为止所拥有的代码......用于非线程实现。网络部分确实有效，因为我已经能够对其进行测试。

文件：config.php

<?php
/*

Configuration File

*/


// Network Parameters
const LISTEN_IPADDR = '0.0.0.0';
const LISTEN_PORT = 8476;
const MAX_CONNECTION_QUEUE = 1024;
const ACCEPT_MODE = 0;
const ALLOW_HOSTS = array();
const BLOCK_HOSTS = array();

// Operational Parameters
const MEMORY_MAX = '1GB';
const LOG_FILE = './vserver.log';
const DEBUG = false;

// Module List
const MODULE_LIST = array(
    'test.php',
);


?>

文件：module.php

<?php
/*

Module Object File

*/


// All modules must implement this interface and also extend the
// class below.

interface moduleInterface
{
    const CMD_READ =        100;    // Read: Sends data back to client
    const CMD_WRITE =       101;    // Write: Writes data to memory
    const CMD_CHECK =       102;    // Check: Checks a value in memory
    const CMD_PURGE =       103;    // Purge: Removed expired data
    const CMD_AUDIT =       104;    // Audit: Data integrity check

    public static function initialize();
    public function process($socket, $command, $data);
}

class moduleObject extends Thread implements moduleInterface
{
    // These have to be set on a per module basis.
    const KEY = NULL;
    const ID = 0x00000000;

    private static $datastore = array();

    function __construct()
    {
        // $class = get_called_class();
        // moduleRegister($class, $this, self::ID, self::KEY);
    }

    function __destruct()
    {
    }

    public static function initialize()
    {
        $class = get_called_class();
        $object = new $class();
        moduleRegister($class, $object, self::ID, self::KEY);
    }

    protected function process($socket, $command, $data, $addr, $port)
    {
        switch ($command)
        {
            case self::CMD_READ:
            case self::CMD_WRITE:
            case self::CMD_CHECK:
            case self::CMD_PURGE:
            case self::CMD_AUDIT:
            default:
                $result = $this->processCustom($socket, $command, $data,
                    $addr, $port);
                if ($result == false)
                {
                    writeLog("Invalid command received from $addr:$port",
                        LOG__WARNING);
                }
                break;
        }
    }

    private function processCustom($socket, $command, $data, $addr, $port)
    {
        return false;
    }

    private function dataRead($socket, $data, $addr, $port)
    {
        return false;
    }

    private function dataWrite($socket, $data, $addr, $port)
    {
        return false;
    }

    private function dataCheck($socket, $data, $addr, $port)
    {
        return false;
    }

    private function dataPurge($socket, $data, $addr, $port)
    {
        return false;
    }

    private function dataAudit($socket, $data, $addr, $port)
    {
        return false;
    }
}



?>

文件：main.php

<?php
/*

Main Server Program

*/

require_once 'config.php';
require_once 'module.php';


// ********************************************************************
// Iinitialize
// ********************************************************************

// This is to work around an issue with PHP on Windows machines.
// Turns out that the Windows Event Viewer has fewer log levels
// than Unix machines, so some of the log levels are mapped to
// the same number.  See https://bugs.php.net/bug.php?id=55129
// for details. We can do this since we are not using syslog.

// Logging Levels
define('LOG__EMERG', 0);
define('LOG__ALERT', 1);
define('LOG__CRIT', 2);
define('LOG__ERR', 3);
define('LOG__WARNING', 4);
define('LOG__NOTICE', 5);
define('LOG__INFO', 6);
define('LOG__DEBUG', 7);

// Global Variables
$LOGFILE = NULL;

// Set Parameters
ini_set('memory_limit', MEMORY_MAX);

// Array that holds all the class references.
// The data format of this array is as follows:
// ID => array(
//      'class' => classname,
//      'reference' => class reference,
//      'key' => access key,
//      'id' => class ID,
//  ),
$moduleRegisterArray = array();


// ******** Start server
openLogFile();
moduleLoad();
moduleStart();
initiateNetwork();
exit(0);



// ********************************************************************
// Functions
// ********************************************************************



// **** Error Handling/Logging

// Opens the log file.
function openLogFile()
{
    global $LOGFILE;

    $LOGFILE = fopen(LOG_FILE, 'a');
    if ($LOGFILE == false)
    {
        fprintf(STDERR, "Error opening log file. Aborting.\n");
        exit(1);
    }
    writeLog("Server started", LOG__NOTICE);
}

// Writes a log message to the log file or to the system console,
// depending on debug mode.
function writeLog($msg, $level)
{
    global $LOGFILE;

    switch ($level)
    {
        case LOG__EMERG:
            $type = '*****EMERGENCY*****';
            break;
        case LOG__ALERT:
            $type = '****ALERT****';
            break;
        case LOG__CRIT:
            $type = '***CRITICAL***';
            break;
        case LOG__ERR:
            $type = '**ERROR**';
            break;
        case LOG__WARNING:
            $type = '*WARNING*';
            break;
        case LOG__NOTICE:
            $type = 'NOTICE';
            break;
        case LOG__INFO:
            $type = 'INFORMATION';
            break;
        case LOG__DEBUG:
            $type = 'DEBUG';
            break;
        default:
            $type = 'UNKNOWN';
            break;
    }

    $date = date('Y-m-d H:m:s');
    if (!DEBUG)
    {
        if ($level != LOG__DEBUG)
        {
            fprintf($LOGFILE, "%s ::-%s-:: %s\n", $date, $type, $msg);
        }
        else
        {
            fprintf(STDOUT, "%s ::-%s-:: %s\n", $date, $type, $msg);
        }
    }
    else
    {
        fprintf($LOGFILE, "%s ::-%s-:: %s\n", $date, $type, $msg);
    }
}

// Handles socket errors
function socketError($socket, $func, $die)
{
    $code = socket_last_error($socket);
    $msg = socket_strerror($code);
    $txt = "Network Error: " . $func . " (" . $code . ") " . $msg;
    writeLog($txt, LOG__ERR);
    if ($die)
    {
        socket_close($socket);
        exit(1);
    }
}




// ******** Module Handling

// Load defined modules.
function moduleLoad()
{
    $path = './modules/';

    foreach(MODULE_LIST as $kx)
    {
        $fileMod = $path . $kx;
        $fileEx = file_exists($fileMod);
        if ($fileEx == true)
        {
            writeLog("Loading module file: $fileMod", LOG__NOTICE);
            require_once $fileMod;
        }
        else
        {
            writeLog("Module does not exist: $fileMod", LOG__WARNING);
        }
    }
}

// Starts off the module registration process.
function moduleStart()
{
    $classList = get_declared_classes();
    foreach($classList as $kx => $vx)
    {
        $position = strpos($vx, 'mod_');
        if ($position === false) continue;
        $vx::initialize();
    }
}

// Each module calls this so it can be registered.
function moduleRegister($class, $reference, $id, $key)
{
    global $moduleRegisterArray;

    $module = array(
        'class' => $class,
        'reference' => $reference,
        'key' => $key,
        'id' => $id,
    );

    $moduleRegisterArray[$id] = $module;
}

// The main server function.
// Does not return.
function initiateNetwork()
{
    global $LOGFILE;

    $func = 'initiate';
    $addr = NULL;
    $port = NULL;

    // Create the network socket.
    $socket = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
    if ($socket === false) socketError($socket, $func, true);

    // Bind the socket.
    $result = socket_bind($socket, LISTEN_IPADDR, LISTEN_PORT);
    if ($result === false) socketError($socket, $func, true);

    // Now listen using an infinite loop.
    // The server will stay in this state until stopped.
    writeLog("The server is now listening on port " . LISTEN_PORT .
        " on address " . LISTEN_IPADDR, LOG__INFO);
    
    // Close the stdin, stdout, and stderr file descriptors.
    if (!DEBUG)
    {
        fclose(STDIN);
        fclose(STDOUT);
        fclose(STDERR);
    }

    while (true)
    {
        $result = socket_listen($socket, MAX_CONNECTION_QUEUE);
        if ($result === false) socketError($socket, $func, true);
        $spawn = socket_accept($socket);
        if ($spawn === false) socketError($socket, $func, true);
        $result = socket_getpeername($spawn, $addr, $port);
        if ($result === false) socketError($socket, $func, true);
        writeLog("Connection accepted from $addr:$port", LOG__INFO);
        process($spawn);
    }

}

// Test Function
function process($socket)
{
    $func = 'process';

    $msg = date('Y-m-d H:m:sP T') . "\r\n";
    $bindata = openssl_random_pseudo_bytes(36);
    $msg .= bin2hex($bindata) . "\r\n";
    $result = socket_write($socket, $msg, strlen($msg));
    if ($result === false) socketError($socket, $func, true);
    socket_close($socket);
}


?>

这只是我一直在研究的一个框架。最终，我希望它执行以下操作：

• 通过网络从客户端接收命令/数据/查询。
• 处理信息并将其转换为 SQL 命令。
• 向数据库服务器发送 SQL 命令。
• 从所述数据库服务器接收结果。
• 处理上述结果。
• 通过网络将结果发送给客户端。

通过网络传输的实际数据是 JSON 格式，因为它是平台中立的。这也是我使用 PHP 的原因。我可能可以使用 Java，但问题是我上次检查时，*nix 系统上的 Java 支持充其量只是零星的。我能看到的唯一其他选择是使用 C++。我还没有探索过 Node.js 提供了什么。

我想要的是传统的服务器线程模型，当客户端连接通过网络进入时会产生新线程，并让它根据客户端的请求执行数据库处理。所以软件需要同时访问网络和数据库。

Answer 1

这两个都是从之前的 SO 答案中借用的，遗憾的是我没有加入书签。

如果您只想通过可以提前生成的 ID 拆分“线程”，则此方法适用。在任一示例中执行以下操作之前，请确保关闭\重新打开所有 SQL 连接，因为您不希望它们共享。

$some_array_of_ids = [.....];

$max_children = 10;   // Max number of forked processes
$pids = [];           // Child process tracking array

foreach ($some_array_of_ids as $unqiue_id) {
    // Limit the number of child processes
    // If $maxChildren or more processes exist, wait until one exits
    if (count($pids) >= $max_children) {
        $pid = pcntl_waitpid(-1, $status);
        unset($pids[$pid]); // Remove PID that exited from the list
    }

    $pid = pcntl_fork();

    if ($pid) { // Parent
        if ($pid < 0) {
            // Unable to fork process, handle error here
            continue;
        } else {
            // Add child PID to tracker array
            // Use PID as key for easy use of unset()
            $pids[$pid] = $unqiue_id;
        }
    } else { // Child
        echo "\nChild {$unqiue_id} has begun process...";

        .....

        echo "\nChild {$unqiue_id} has finished process...";
        exit(0);
    }
}

// Now wait for the child processes to exit. This approach may seem overly
// simple, but because of the way it works it will have the effect of
// waiting until the last process exits and pretty much no longer

foreach ($pids as $pid => $unqiue_id) {
    pcntl_waitpid($pid, $status);
    unset($pids[$pid]);
}

如果你真的想流式传输套接字数据，还有一个更复杂的方法：

$some_array_of_data = [.....];

foreach ($some_array_of_data as $group) {
    $sockets = array();
    if (!socket_create_pair(AF_UNIX, SOCK_STREAM, 0, $sockets)) {
        die(socket_strerror(socket_last_error()));
    }
    list($reader, $writer) = $sockets;

    $pid = pcntl_fork();

    if ($pid == -1) {
        die('cannot fork');
    } elseif ($pid) {
        socket_close($reader);
        $line = sprintf("{$group}\n", getmypid());
        if (!socket_write($writer, $line, strlen($line))) {
            socket_close($writer);
            die(socket_strerror(socket_last_error()));
        }
        socket_close($writer);

    } else {

        socket_close($writer);
        $line = socket_read($reader, 1024, PHP_NORMAL_READ);

        $id = rtrim($line);
        $child_id = getmypid();
        socket_close($reader);

        echo "\nChild {$child_id} has begun process: {$id}.";
        
        ....

        echo "\nChild {$child_id} has finished process: {$id}.";
        exit(0);
    }
}

while(pcntl_waitpid(0, $status) != -1);

虽然 fork != threading，但 pthreads 本质上是 fork 上的一个抽象层，而不是真正的线程，所以......嗯。无论哪种方式，都有两种不同的方式来处理这个问题，具体取决于您的具体需求。

除非你真的需要使用流来做某事，否则我会推荐前一个。在这种情况下，我需要更多细节来给你更好的洞察力。如果是这种情况，底部的示例应该为您指明正确的方向，如果此代码示例还不够，我可以向您展示如何根据您的需要对其进行优化。