本文目录导读:

- 使用 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);
}
?>
最佳实践建议
- 普通场景:使用
stream_select()足矣 - 高并发:推荐 Swoole 或 ReactPHP
- 简单任务:使用
stream_set_blocking() - CPU 密集型:考虑
pcntl_fork()
注意事项
- 非阻塞 IO 需要注意数据完整性
- 需要处理超时和错误情况
- 内存管理要谨慎
- 避免 CPU 空转(添加 sleep 或合理等待)
选择哪种方式取决于你的具体需求和应用场景,对于简单任务,内置函数就够了;对于高并发场景,建议使用 Swoole 等专业工具。