PHP小码农

Appearance

流的概念 ​

概述 ​

PHP 流（Streams）是 PHP 中处理数据的统一抽象层。无论是文件、网络连接、压缩数据还是进程管道，都可以通过流的接口来读写。流由三层组成：流本身、流包装器（Wrappers）和流上下文（Context）。理解流模型有助于掌握 PHP 的高级 I/O 操作。

适用场景

  • 统一的文件/网络 I/O 操作
  • 自定义协议处理
  • 流过滤器实现数据转换
  • 高效处理大文件和网络数据

基础概念 ​

流的三层架构 ​

┌─────────────────────────┐
│     流过滤器 (Filters)    │  数据转换层（如 base64 编码）
├─────────────────────────┤
│   流传输 (Transport)     │  实际的 I/O 操作
├─────────────────────────┤
│   流包装器 (Wrappers)     │  协议/路径解析（如 file://, http://）
├─────────────────────────┤
│   流上下文 (Context)      │  配置选项（超时、头部等）
└─────────────────────────┘

核心流函数 ​

函数功能
fopen()打开流
fread() / fgets()从流读取
fwrite() / fputs()写入流
fclose()关闭流
stream_get_contents()读取剩余内容
stream_get_line()按行读取
stream_select()多路复用 I/O
stream_filter_append()添加过滤器
stream_context_create()创建上下文

统一接口

所有流使用相同的函数接口（fopen/fread/fwrite），无需关心底层是文件、HTTP 还是其他协议。

语法与代码示例 ​

基本流操作 ​

php
<?php

// 文件流
$handle = fopen('/tmp/data.txt', 'r');
$content = stream_get_contents($handle);
fclose($handle);

// HTTP 流
$handle = fopen('https://example.com/api/data', 'r');
$response = stream_get_contents($handle);
fclose($handle);

// php:// 流
$stdin = fopen('php://stdin', 'r');
$input = stream_get_line($stdin, 1024, "\n");
fclose($stdin);

// 内存流
$memory = fopen('php://memory', 'r+');
fwrite($memory, 'Hello, Stream!');
rewind($memory);
echo stream_get_contents($memory); // Hello, Stream!
fclose($memory);

stream_get_contents ​

php
<?php

// 读取指定长度
$handle = fopen('/tmp/data.txt', 'r');
$content = stream_get_contents($handle, 1024); // 最多 1024 字节
fclose($handle);

// 从偏移位置读取
$handle = fopen('/tmp/data.txt', 'r');
$content = stream_get_contents($handle, -1, 100); // 从第 100 字节开始读到末尾
fclose($handle);

// 读取到末尾
$handle = fopen('/tmp/data.txt', 'r');
// 先读了一部分
$partial = fread($handle, 100);
// 读取剩余
$rest = stream_get_contents($handle);
fclose($handle);

stream_select 多路复用 ​

php
<?php

// stream_select 同时监听多个流
$read = [];
$write = [];
$except = [];

// 添加要监听的流
$stream1 = fopen('http://example.com/api/1', 'r');
$stream2 = fopen('http://example.com/api/2', 'r');
$stream3 = fopen('php://stdin', 'r');

$read[] = $stream1;
$read[] = $stream2;
$read[] = $stream3;

// 等待最多 5 秒
$changed = stream_select($read, $write, $except, 5);

if ($changed === false) {
    echo "select 失败\n";
} elseif ($changed === 0) {
    echo "超时\n";
} else {
    foreach ($read as $ready) {
        $data = fread($ready, 4096);
        if (strlen($data) === 0) {
            fclose($ready);
        } else {
            echo "收到数据: " . substr($data, 0, 50) . "...\n";
        }
    }
}

流的元信息 ​

php
<?php

// stream_get_meta_data 获取流的元信息
$handle = fopen('https://example.com', 'r');
$meta = stream_get_meta_data($handle);

print_r($meta);
/*
[
    'wrapper_data' => [...],  // 包装器特定的数据/头部
    'wrapper_type' => 'http', // 包装器类型
    'stream_type'  => 'tcp_socket/ssl', // 流类型
    'mode'         => 'r',     // 访问模式
    'unread_bytes' => 0,      // 未读字节数
    'seekable'     => false,  // 是否可定位
    'eof'          => false,  // 是否到达末尾
    'blocked'      => true,   // 是否阻塞
    'uri'          => 'https://example.com', // URI
]
*/

// 判断是否到达末尾
while (!feof($handle)) {
    $data = fread($handle, 4096);
    // 处理数据
}

fclose($handle);

实战示例 ​

简单 HTTP 客户端 ​

php
<?php

declare(strict_types=1);

class SimpleHttpClient
{
    public function get(string $url, array $headers = [], int $timeout = 30): string
    {
        $context = stream_context_create([
            'http' => [
                'method'  => 'GET',
                'timeout' => $timeout,
                'header'  => $this->buildHeaders($headers),
                'ignore_errors' => true, // 不抛出 HTTP 错误
            ],
        ]);

        $handle = fopen($url, 'r', false, $context);
        if ($handle === false) {
            throw new RuntimeException("无法打开 URL: {$url}");
        }

        // 获取响应头和状态码
        $meta = stream_get_meta_data($handle);
        $statusCode = $this->parseStatusCode($meta['wrapper_data'] ?? []);

        $body = stream_get_contents($handle);
        fclose($handle);

        if ($statusCode >= 400) {
            throw new RuntimeException("HTTP 错误: {$statusCode}");
        }

        return $body;
    }

    public function post(string $url, array $data, array $headers = [], int $timeout = 30): string
    {
        $body = http_build_query($data);

        $context = stream_context_create([
            'http' => [
                'method'  => 'POST',
                'timeout' => $timeout,
                'header'  => array_merge(
                    $this->buildHeaders($headers),
                    ['Content-Type: application/x-www-form-urlencoded', 'Content-Length: ' . strlen($body)]
                ),
                'content' => $body,
                'ignore_errors' => true,
            ],
        ]);

        $response = file_get_contents($url, false, $context);
        if ($response === false) {
            throw new RuntimeException("POST 请求失败: {$url}");
        }

        return $response;
    }

    private function buildHeaders(array $headers): array
    {
        return array_map(
            fn(string $key, string $value) => "{$key}: {$value}",
            array_keys($headers),
            $headers
        );
    }

    private function parseStatusCode(array $wrapperData): int
    {
        foreach ($wrapperData as $header) {
            if (preg_match('/^HTTP\/\d\.\d\s+(\d+)/', $header, $matches)) {
                return (int)$matches[1];
            }
        }
        return 200;
    }
}

// 使用
$client = new SimpleHttpClient();
$data = $client->get('https://api.example.com/users');
$result = $client->post('https://api.example.com/login', ['username' => 'test', 'password' => 'secret']);

流式大文件下载 ​

php
<?php

declare(strict_types=1);

function streamDownload(string $url, string $destPath, int $timeout = 60): void
{
    $context = stream_context_create([
        'http' => ['timeout' => $timeout, 'follow_location' => true],
    ]);

    $source = fopen($url, 'rb', false, $context);
    if ($source === false) {
        throw new RuntimeException("无法打开远程文件: {$url}");
    }

    $dest = fopen($destPath, 'wb');
    if ($dest === false) {
        fclose($source);
        throw new RuntimeException("无法创建本地文件: {$destPath}");
    }

    // 流式复制，不占用大内存
    $totalBytes = stream_copy_to_stream($source, $dest);
    fclose($source);
    fclose($dest);

    echo "下载完成: {$totalBytes} 字节\n";
}

// 使用
streamDownload('https://example.com/large-file.zip', '/tmp/download.zip');

注意事项 ​

流的超时处理 ​

php
<?php

// 设置流的超时
$context = stream_context_create([
    'http' => [
        'timeout' => 10, // 连接 + 传输总超时
    ],
    'socket' => [
        'bindto' => '192.168.1.1:0', // 绑定本地 IP
    ],
]);

// 默认超时（php.ini default_socket_timeout）
echo ini_get('default_socket_timeout') . PHP_EOL;

// 运行时修改超时
$handle = fopen('https://slow-api.example.com', 'r');
stream_set_timeout($handle, 5); // 5 秒超时

$info = stream_get_meta_data($handle);
echo "Timed out: " . ($info['timed_out'] ? 'yes' : 'no') . PHP_EOL;

流的阻塞与非阻塞 ​

php
<?php

// 阻塞模式（默认）
$handle = fopen('https://example.com', 'r');
// fread 会等到数据到达才返回

// 非阻塞模式
stream_set_blocking($handle, false);
// fread 立即返回，可能返回空字符串

最佳实践 ​

1. 使用 stream_copy_to_stream 代替 file_get_contents ​

php
<?php

// 不好：大文件占用内存
$data = file_get_contents('http://example.com/large-file.zip');
file_put_contents('/tmp/file.zip', $data); // 内存翻倍

// 好：流式传输，内存友好
$source = fopen('http://example.com/large-file.zip', 'rb');
$dest = fopen('/tmp/file.zip', 'wb');
stream_copy_to_stream($source, $dest);
fclose($source);
fclose($dest);

2. 总是关闭流 ​

php
<?php

$handle = fopen('http://example.com', 'r');
try {
    $data = stream_get_contents($handle);
} finally {
    fclose($handle); // 确保关闭
}

3. 使用 try-finally 管理流资源 ​

php
<?php

function processStream(string $url): string
{
    $handle = fopen($url, 'r');
    if ($handle === false) {
        throw new RuntimeException("无法打开: {$url}");
    }

    try {
        return stream_get_contents($handle);
    } finally {
        fclose($handle);
    }
}

参考链接 ​