PHP项目流式HTTP与回调

wen PHP项目 2026-07-25 177

本文目录导读：

PHP项目流式HTTP与回调

流式HTTP（Streaming HTTP）
回调机制（Callbacks）
流式HTTP与回调结合
性能优化建议

我来详细介绍一下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;
    }
}

这些实现提供了灵活、高效的流式处理和回调机制，适用于大文件处理、实时数据流、异步任务等场景。

本文地址： https://www.bmjup.com/post/29421.html