本文目录导读：
我来详细介绍一下PHP项目中流式HTTP和回调机制的实现方案。
流式HTTP（Streaming HTTP）
1 使用cURL实现流式下载
<?php
class StreamDownloader {
public function streamFile($url, $outputFile) {
$ch = curl_init();
$fp = fopen($outputFile, 'wb');
curl_setopt_array($ch, [
CURLOPT_URL => $url,
CURLOPT_FILE => $fp,
CURLOPT_FOLLOWLOCATION => true,
CURLOPT_PROGRESSFUNCTION => [$this, 'progressCallback'],
CURLOPT_NOPROGRESS => false,
CURLOPT_BUFFERSIZE => 8192, // 8KB chunks
]);
curl_exec($ch);
$error = curl_error($ch);
curl_close($ch);
fclose($fp);
if ($error) {
throw new Exception("Download failed: $error");
}
}
public function progressCallback($ch, $downloadSize, $downloaded, $uploadSize, $uploaded) {
if ($downloadSize > 0) {
$percent = ($downloaded / $downloadSize) * 100;
printf("Downloaded: %.2f%%\r", $percent);
}
}
}
2 流式响应（服务端发送流式数据）
<?php
// 服务端 - 流式输出
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');
header('Connection: keep-alive');
class StreamResponse {
public function sendEvents($dataGenerator) {
foreach ($dataGenerator as $chunk) {
echo "data: " . json_encode($chunk) . "\n\n";
ob_flush();
flush();
if (connection_aborted()) {
break;
}
usleep(100000); // 100ms delay
}
echo "event: complete\ndata: {}\n\n";
}
}
// 使用示例
$stream = new StreamResponse();
$generator = function() {
for ($i = 0; $i < 10; $i++) {
yield ['progress' => $i * 10, 'message' => "Processing item $i"];
}
};
$stream->sendEvents($generator);
3 客户端接收流式数据
<?php
// 客户端 - 接收流式数据
class StreamClient {
public function consumeStream($url) {
$ch = curl_init();
curl_setopt_array($ch, [
CURLOPT_URL => $url,
CURLOPT_WRITEFUNCTION => function($ch, $data) {
$lines = explode("\n", $data);
foreach ($lines as $line) {
if (strpos($line, 'data: ') === 0) {
$payload = substr($line, 6);
$this->handleData(json_decode($payload, true));
} elseif (strpos($line, 'event: complete') !== false) {
return -1; // Stop cURL
}
}
return strlen($data);
},
CURLOPT_TIMEOUT => 0, // No timeout
]);
curl_exec($ch);
curl_close($ch);
}
private function handleData($data) {
echo "Received: " . print_r($data, true) . "\n";
}
}
回调机制（Callbacks）
1 简单的回调函数实现
<?php
class CallbackHandler {
private $callbacks = [];
public function registerCallback($event, callable $callback) {
$this->callbacks[$event][] = $callback;
}
public function trigger($event, ...$args) {
if (isset($this->callbacks[$event])) {
foreach ($this->callbacks[$event] as $callback) {
call_user_func_array($callback, $args);
}
}
}
public function removeCallback($event, callable $callback) {
if (isset($this->callbacks[$event])) {
$this->callbacks[$event] = array_filter(
$this->callbacks[$event],
function($cb) use ($callback) {
return $cb !== $callback;
}
);
}
}
}
// 使用示例
$handler = new CallbackHandler();
// 注册回调
$handler->registerCallback('user.created', function($userId, $name) {
echo "User created: ID=$userId, Name=$name\n";
// 发送欢迎邮件等
});
// 触发事件
$handler->trigger('user.created', 123, 'John Doe');
2 异步回调处理
<?php
class AsyncCallbackProcessor {
private $queue = [];
private $processing = false;
public function addCallback(callable $callback, ...$args) {
$this->queue[] = ['callback' => $callback, 'args' => $args];
if (!$this->processing) {
$this->processQueue();
}
}
private function processQueue() {
$this->processing = true;
while (!empty($this->queue)) {
$item = array_shift($this->queue);
try {
call_user_func_array($item['callback'], $item['args']);
} catch (Exception $e) {
// Log error
error_log("Callback failed: " . $e->getMessage());
}
// Yield control to allow other operations
if (function_exists('fastcgi_finish_request')) {
fastcgi_finish_request();
}
}
$this->processing = false;
}
}
3 使用闭包和匿名函数
<?php
class TaskRunner {
private $onProgress;
private $onComplete;
private $onError;
public function setOnProgress(callable $callback) {
$this->onProgress = $callback;
}
public function setOnComplete(callable $callback) {
$this->onComplete = $callback;
}
public function setOnError(callable $callback) {
$this->onError = $callback;
}
public function execute($tasks) {
$total = count($tasks);
$completed = 0;
foreach ($tasks as $index => $task) {
try {
$result = $this->processTask($task);
$completed++;
if ($this->onProgress) {
call_user_func($this->onProgress, $completed, $total, $result);
}
} catch (Exception $e) {
if ($this->onError) {
call_user_func($this->onError, $e, $task);
}
}
}
if ($this->onComplete) {
call_user_func($this->onComplete, $completed, $total);
}
}
private function processTask($task) {
// Simulate task processing
sleep(1);
return "Processed: $task";
}
}
// 使用示例
$runner = new TaskRunner();
$runner->setOnProgress(function($completed, $total, $result) {
$percent = ($completed / $total) * 100;
echo "Progress: $percent% - $result\n";
});
$runner->setOnComplete(function($completed, $total) {
echo "Completed $completed of $total tasks\n";
});
$runner->setOnError(function($error, $task) {
echo "Error processing '$task': " . $error->getMessage() . "\n";
});
$runner->execute(['task1', 'task2', 'task3', 'task4', 'task5']);
流式HTTP与回调结合
1 大文件处理系统
<?php
class FileStreamProcessor {
private $onChunk;
private $onComplete;
private $onError;
public function __construct() {
$this->onChunk = function($data) {};
$this->onComplete = function() {};
$this->onError = function($error) {};
}
public function setOnChunk(callable $callback) {
$this->onChunk = $callback;
}
public function setOnComplete(callable $callback) {
$this->onComplete = $callback;
}
public function setOnError(callable $callback) {
$this->onError = $callback;
}
public function streamAndProcess($url) {
$ch = curl_init();
curl_setopt_array($ch, [
CURLOPT_URL => $url,
CURLOPT_WRITEFUNCTION => function($ch, $data) {
try {
// Process each chunk
$processed = $this->processChunk($data);
// Trigger chunk callback
call_user_func($this->onChunk, $processed, strlen($data));
// Check if connection is still alive
if (connection_aborted()) {
return -1;
}
return strlen($data);
} catch (Exception $e) {
call_user_func($this->onError, $e);
return -1;
}
},
CURLOPT_HEADERFUNCTION => function($ch, $headerLine) {
// Parse headers if needed
return strlen($headerLine);
},
CURLOPT_BUFFERSIZE => 16384, // 16KB chunks
CURLOPT_FOLLOWLOCATION => true,
]);
$result = curl_exec($ch);
if (curl_errno($ch)) {
call_user_func($this->onError, new Exception(curl_error($ch)));
} else {
call_user_func($this->onComplete, $result);
}
curl_close($ch);
}
private function processChunk($data) {
// Example: calculate stats on the fly
return [
'size' => strlen($data),
'hash' => md5($data),
'timestamp' => time(),
];
}
}
// 使用示例
$processor = new FileStreamProcessor();
$processor->setOnChunk(function($stats, $chunkSize) {
echo "Processed chunk: {$stats['size']} bytes, hash: {$stats['hash']}\n";
});
$processor->setOnComplete(function($result) {
echo "Stream completed successfully\n";
});
$processor->setOnError(function($error) {
echo "Error: " . $error->getMessage() . "\n";
});
// 处理远程大文件
$processor->streamAndProcess('https://example.com/large-file.zip');
性能优化建议
1 使用生成器减少内存消耗
<?php
function streamFileInChunks($filePath, $chunkSize = 8192) {
$handle = fopen($filePath, 'rb');
while (!feof($handle)) {
yield fread($handle, $chunkSize);
}
fclose($handle);
}
// 使用生成器处理大文件
foreach (streamFileInChunks('large-file.txt') as $chunk) {
processChunk($chunk);
}
2 使用并行处理提高效率
<?php
class ParallelStreamProcessor {
private $maxParallel = 5;
private $callbacks = [];
public function processMultiple($urls) {
$multiHandle = curl_multi_init();
$handles = [];
foreach ($urls as $index => $url) {
$ch = curl_init();
curl_setopt_array($ch, [
CURLOPT_URL => $url,
CURLOPT_RETURNTRANSFER => true,
CURLOPT_WRITEFUNCTION => function($ch, $data) use ($index) {
if (isset($this->callbacks['data'])) {
call_user_func($this->callbacks['data'], $index, $data);
}
return strlen($data);
},
]);
curl_multi_add_handle($multiHandle, $ch);
$handles[$index] = $ch;
if (count($handles) >= $this->maxParallel) {
$this->processBatch($multiHandle, $handles);
$handles = [];
}
}
// Process remaining
if (!empty($handles)) {
$this->processBatch($multiHandle, $handles);
}
curl_multi_close($multiHandle);
}
private function processBatch($multiHandle, &$handles) {
$running = null;
do {
$status = curl_multi_exec($multiHandle, $running);
if ($status !== CURLM_OK) {
break;
}
// Check for completed transfers
while ($info = curl_multi_info_read($multiHandle)) {
$ch = $info['handle'];
$index = array_search($ch, $handles);
if ($info['result'] === CURLE_OK) {
if (isset($this->callbacks['complete'])) {
call_user_func($this->callbacks['complete'], $index);
}
} else {
if (isset($this->callbacks['error'])) {
call_user_func($this->callbacks['error'], $index, curl_error($ch));
}
}
curl_multi_remove_handle($multiHandle, $ch);
curl_close($ch);
unset($handles[$index]);
}
// Allow other processes to run
if ($running) {
curl_multi_select($multiHandle, 0.1);
}
} while ($running > 0);
}
public function on($event, callable $callback) {
$this->callbacks[$event] = $callback;
}
}
这些实现提供了灵活、高效的流式处理和回调机制，适用于大文件处理、实时数据流、异步任务等场景。
