PHP 怎么PHP数据流

本文目录导读：

1. 流式处理（Streams）
2. 数据流编程（Pipeline / Generator）
3. 最终推荐：完整实例
4. 实用建议
5. 关键函数总结

在 PHP 中，“数据流”这个概念通常有两种主要理解：流式处理数据（如文件、网络）和数据流编程（如处理大量数据），下面从这两个角度分别说明，并提供实用示例。

流式处理（Streams）

PHP 有一套成熟的流抽象层，用于统一处理文件、网络、内存等数据源。

基础流操作

<?php
// 1. 读取整个文件
$content = file_get_contents('data.txt');
// 2. 流式读取大文件（避免内存溢出）
$handle = fopen('large_file.log', 'r');
while (($line = fgets($handle)) !== false) {
    // 处理每一行
    echo $line; // 或做其他处理
}
fclose($handle);
// 3. 写入文件
file_put_contents('output.txt', 'Hello World');
// 4. 流式写入
$handle = fopen('output.txt', 'w');
fwrite($handle, '第一行数据');
fwrite($handle, '第二行数据');
fclose($handle);
?>

不同类型的流

<?php
// 文件流
$fileStream = fopen('data.txt', 'r');
// 网络流（HTTP）
$httpStream = fopen('https://api.example.com/data', 'r');
// 内存流
$memoryStream = fopen('php://memory', 'r+');
fwrite($memoryStream, '临时数据');
rewind($memoryStream);
$data = fread($memoryStream, 1024);
// 输出流
$outputStream = fopen('php://output', 'w');
fwrite($outputStream, '发送内容');
?>

使用流包装器

<?php
// 压缩流
$gzipStream = gzopen('data.gz', 'r');
while (!gzeof($gzipStream)) {
    echo gzgets($gzipStream);
}
gzclose($gzipStream);
// 自定义流过滤
class MyFilter extends php_user_filter
{
    public function filter($in, $out, &$consumed, $closing)
    {
        while ($bucket = stream_bucket_make_writeable($in)) {
            $bucket->data = strtoupper($bucket->data);
            $consumed += $bucket->datalen;
            stream_bucket_append($out, $bucket);
        }
        return PSFS_PASS_ON;
    }
}
stream_filter_register('my_filter', 'MyFilter');
$handle = fopen('data.txt', 'r');
stream_filter_append($handle, 'my_filter');
while (!feof($handle)) {
    echo fread($handle, 1024);
}
fclose($handle);
?>

数据流编程（Pipeline / Generator）

用于处理大量数据时,避免一次性加载到内存。

使用生成器（Generator）

<?php
// 定义一个生成器函数
function getData()
{
    for ($i = 0; $i < 1000000; $i++) {
        yield $i;
    }
}
// 流式处理，不占用大量内存
foreach (getData() as $value) {
    if ($value % 10000 === 0) {
        echo "处理到: $value\n";
    }
    // 做数据处理
}
?>

管道式处理（类似 Unix Pipeline）

<?php
// 数据处理管道类
class DataPipeline
{
    private array $stages = [];
    public function addStage(callable $callable): self
    {
        $this->stages[] = $callable;
        return $this;
    }
    public function process(Generator $data): Generator
    {
        foreach ($data as $item) {
            $result = $item;
            foreach ($this->stages as $stage) {
                $result = $stage($result);
            }
            yield $result;
        }
    }
}
// 创建数据源
function source(): Generator
{
    for ($i = 1; $i <= 10; $i++) {
        yield $i;
    }
}
// 使用管道
$pipeline = new DataPipeline();
$pipeline->addStage(fn($data) => $data * 2)      // 乘以2
         ->addStage(fn($data) => $data + 1)      // 加1
         ->addStage(fn($data) => "结果: $data");  // 格式化
// 处理数据
foreach ($pipeline->process(source()) as $result) {
    echo $result . "\n";
}
?>

最终推荐：完整实例

下面是一个完整的“PHP 数据流处理”示例，展示从文件读取、处理、写出的完整流程：

<?php
// 创建一个综合示例：处理大文件并统计
class LogProcessor
{
    private string $inputFile;
    private string $outputFile;
    public function __construct(string $input, string $output)
    {
        $this->inputFile = $input;
        $this->outputFile = $output;
    }
    public function process(): array
    {
        $stats = [
            'total_lines' => 0,
            'error_lines' => 0,
            'warning_lines' => 0,
        ];
        // 读取流
        $input = fopen($this->inputFile, 'r');
        $output = fopen($this->outputFile, 'w');
        if (!$input || !$output) {
            throw new RuntimeException('无法打开文件');
        }
        // 流式处理
        while (!feof($input)) {
            $line = fgets($input);
            if ($line === false) break;
            $stats['total_lines']++;
            // 处理逻辑
            if (str_contains($line, 'ERROR')) {
                $stats['error_lines']++;
                fwrite($output, '[ERROR] ' . $line);
            } elseif (str_contains($line, 'WARNING')) {
                $stats['warning_lines']++;
                fwrite($output, '[WARN]  ' . $line);
            }
        }
        fclose($input);
        fclose($output);
        return $stats;
    }
}
// 使用示例
try {
    $processor = new LogProcessor('app.log', 'filtered.log');
    $stats = $processor->process();
    echo "处理完成:\n";
    echo "总行数: {$stats['total_lines']}\n";
    echo "错误行: {$stats['error_lines']}\n";
    echo "警告行: {$stats['warning_lines']}\n";
} catch (Exception $e) {
    echo "错误: " . $e->getMessage() . "\n";
}
?>

实用建议

关键函数总结

• 读取流：fopen(), fgets(), fread(), file_get_contents()
• 写入流：fwrite(), file_put_contents()
• 流控制：fclose(), feof(), rewind(), fflush()
• 内存流：php://memory, php://temp
• 网络流：fopen('http://...'), cURL
• 生成器：yield