PHP 怎么非阻塞IO

wen PHP项目 2026-08-04 185

本文目录导读：

PHP 怎么非阻塞IO

使用 stream_set_blocking()
使用 stream_select() 实现多路复用
使用 Swoole 扩展（高性能异步）
使用 ReactPHP 库
使用 pcntl_fork() 多进程（适合 CPU 密集型）
使用非阻塞 Socket 实现完整示例
使用信号和定时器实现定时任务
最佳实践建议
注意事项

在 PHP 中实现非阻塞 IO 有几种方式，我来详细介绍：

使用 stream_set_blocking()

最简单的方式,将流设置为非阻塞模式：

<?php
// 非阻塞模式读取
$fp = fsockopen("example.com", 80, $errno, $errstr, 30);
if (!$fp) {
    echo "$errstr ($errno)<br />\n";
} else {
    stream_set_blocking($fp, 0); // 设置为非阻塞
    fwrite($fp, "GET / HTTP/1.0\r\nHost: example.com\r\n\r\n");
    // 非阻塞读取，没有数据立即返回
    while (!feof($fp)) {
        $data = fgets($fp, 128);
        if ($data !== false) {
            echo $data;
        } else {
            // 没有数据，可以做其他事情
            usleep(100000); // 等待100ms
        }
    }
    fclose($fp);
}
?>

使用 stream_select() 实现多路复用

<?php
// 创建多个非阻塞连接
$connections = [];
$urls = ['example.com', 'google.com', 'github.com'];
foreach ($urls as $url) {
    $fp = fsockopen($url, 80, $errno, $errstr, 5);
    if ($fp) {
        stream_set_blocking($fp, 0);
        fwrite($fp, "GET / HTTP/1.0\r\nHost: $url\r\n\r\n");
        $connections[$url] = $fp;
    }
}
// 使用 stream_select 等待数据
$read = $connections;
$write = NULL;
$except = NULL;
$timeout = 5;
if (stream_select($read, $write, $except, $timeout) > 0) {
    foreach ($read as $url => $fp) {
        echo "从 $url 获取数据: " . fread($fp, 1024) . "\n";
    }
}
// 关闭所有连接
foreach ($connections as $fp) {
    fclose($fp);
}
?>

使用 Swoole 扩展（高性能异步）

<?php
// Swoole 协程实现异步
Co\run(function() {
    // 并发发起多个 HTTP 请求
    $results = [];
    $tasks = [
        'https://example.com',
        'https://google.com',
        'https://github.com'
    ];
    foreach ($tasks as $i => $url) {
        go(function() use ($url, &$results, $i) {
            $client = new Swoole\Coroutine\Http\Client(parse_url($url, PHP_URL_HOST), 443, true);
            $client->set(['timeout' => 10]);
            $client->get(parse_url($url, PHP_URL_PATH) ?: '/');
            $results[$i] = $client->body;
            $client->close();
        });
    }
    // 等待所有协程完成
    while (count($results) < count($tasks)) {
        usleep(1000);
    }
    var_dump($results);
});
?>

使用 ReactPHP 库

<?php
require 'vendor/autoload.php';
use React\EventLoop\Factory;
use React\Socket\Connector;
use React\Promise\Promise;
$loop = Factory::create();
$connector = new Connector($loop);
// 并发连接多个服务器
$promises = [];
$urls = ['www.google.com:80', 'www.github.com:80', 'www.php.net:80'];
foreach ($urls as $url) {
    $promise = $connector->connect($url)->then(
        function ($conn) use ($loop) {
            $conn->write("GET / HTTP/1.0\r\nHost: example.com\r\nConnection: close\r\n\r\n");
            return new Promise(function ($resolve) use ($conn, $loop) {
                $data = '';
                $conn->on('data', function ($chunk) use (&$data) {
                    $data .= $chunk;
                });
                $conn->on('close', function () use ($resolve, &$data) {
                    $resolve($data);
                });
            });
        },
        function ($e) {
            return "错误: " . $e->getMessage();
        }
    );
    $promises[$url] = $promise;
}
// 等待所有请求完成
\React\Promise\all($promises)->then(function ($results) {
    foreach ($results as $url => $result) {
        echo "从 $url 获取响应\n";
    }
});
$loop->run();
?>

使用 pcntl_fork() 多进程（适合 CPU 密集型）

<?php
// 创建多个进程处理任务
$processes = [];
$tasks = ['task1', 'task2', 'task3'];
foreach ($tasks as $task) {
    $pid = pcntl_fork();
    if ($pid == -1) {
        // fork 失败
        die('无法创建子进程');
    } elseif ($pid) {
        // 父进程
        $processes[] = $pid;
    } else {
        // 子进程
        // 这里是异步任务
        sleep(rand(1,3));
        echo "任务 $task 完成 (PID: " . getmypid() . ")\n";
        exit(0);
    }
}
// 等待所有子进程完成
foreach ($processes as $pid) {
    pcntl_waitpid($pid, $status);
}
?>

使用非阻塞 Socket 实现完整示例

<?php
class NonBlockingSocket {
    private $connections = [];
    public function addConnection($host, $port) {
        $fp = stream_socket_client("tcp://$host:$port", $errno, $errstr, 5);
        if (!$fp) {
            throw new Exception("连接失败: $errstr");
        }
        stream_set_blocking($fp, 0);
        $this->connections[(int)$fp] = ['fp' => $fp, 'host' => $host, 'data' => ''];
        return $fp;
    }
    public function sendData($fp, $data) {
        return fwrite($fp, $data);
    }
    public function waitForData($timeout = 1) {
        $read = array_column($this->connections, 'fp');
        $write = NULL;
        $except = NULL;
        if (stream_select($read, $write, $except, $timeout) > 0) {
            $messages = [];
            foreach ($read as $fp) {
                $data = fread($fp, 8192);
                if ($data !== false && strlen($data) > 0) {
                    $conn = $this->connections[(int)$fp];
                    $messages[] = [
                        'host' => $conn['host'],
                        'data' => $data
                    ];
                } else if (feof($fp)) {
                    $this->closeConnection($fp);
                }
            }
            return $messages;
        }
        return [];
    }
    public function closeConnection($fp) {
        if (isset($this->connections[(int)$fp])) {
            fclose($fp);
            unset($this->connections[(int)$fp]);
        }
    }
    public function closeAll() {
        foreach ($this->connections as $conn) {
            fclose($conn['fp']);
        }
        $this->connections = [];
    }
}
// 使用示例
$socket = new NonBlockingSocket();
$socket->addConnection('www.google.com', 80);
$socket->addConnection('www.github.com', 80);
// 发送请求
foreach ($socket->connections as $conn) {
    $socket->sendData($conn['fp'], "GET / HTTP/1.0\r\nHost: {$conn['host']}\r\n\r\n");
}
// 等待响应
$responses = $socket->waitForData(2);
foreach ($responses as $response) {
    echo "收到来自 {$response['host']} 的响应\n";
}
$socket->closeAll();
?>

使用信号和定时器实现定时任务

<?php
declare(ticks = 1);
function signalHandler($signo) {
    switch ($signo) {
        case SIGALRM:
            echo "定时任务触发\n";
            // 执行异步任务
            break;
        case SIGUSR1:
            echo "收到用户信号\n";
            break;
    }
}
pcntl_signal(SIGALRM, 'signalHandler');
pcntl_signal(SIGUSR1, 'signalHandler');
// 设置定时器
pcntl_alarm(2);
while (true) {
    echo "主程序运行中...\n";
    sleep(1);
}
?>