Потоки PHP: от основ к практическому применению
Основы работы с потоками в PHP
Потоки (streams) в PHP - это абстракция для работы с последовательными данными, которая единообразно обрабатывает файлы, сетевые соединения, сжатые данные или временные хранилища. Благодаря такому подходу одни и те же функции (fopen, fread, fwrite) применимы к любому источнику данных.
Как эффективно прочитать тело HTTP запроса или отправить запрос с кастомными заголовками?
Самым универсальным и производительным решением является использование встроенных обёрток php://input и stream_context_create. Первая позволяет получить необработанные POST данные, вторая - настроить HTTP или HTTPS запрос с любыми параметрами.
<?php
// Чтение тела запроса
$input = fopen('php://input', 'rb');
$data = stream_get_contents($input);
fclose($input);
echo $data;
?>
Stream php id (идентификатор потока php)
Для отправки запросов с пользовательскими заголовками применяется контекст:
<?php
$opts = ['http' => [
'method' => 'POST',
'header' => "Content-type: application/json\r\n",
'content' => json_encode(['key' => 'value']),
'timeout' => 10
]];
$context = stream_context_create($opts);
$result = file_get_contents('http://example.com/api', false, $context);
echo $result;
?>
Stream index php (индекс потока php)
Типичные ошибки: если allow_url_fopen выключен, file_get_contents вернёт false. Решение - использовать cURL или включить директиву. При работе с php://input поток может быть прочитан только один раз, так как данные уже получены от сервера.
Как читать большие файлы, не загружая их целиком в память?
Для этого используется цикл с fread и проверкой на конец файла (feof). Такой подход потребляет минимум оперативной памяти независимо от размера файла.
<?php
$handle = fopen('bigfile.log', 'rb');
if (!$handle) die('Не удалось открыть файл');
while (!feof($handle)) {
$buffer = fread($handle, 8192); // читаем блоками по 8 КБ
echo $buffer;
// обработка блока
}
fclose($handle);
?>
Php streaming (потоки php)
Проблема: если не закрыть дескриптор после ошибки, ресурсы останутся занятыми. Используйте исключения или finally блок.
Как сохранить данные во временный файл, который автоматически удалится после закрытия?
Обёртка php://temp или php://memory сохраняет данные в памяти, а php://temp при превышении лимита - сбрасывается на диск. Это удобно для буферизации вывода или временного хранения.
<?php
$temp = fopen('php://temp/maxmemory:2097152', 'rw'); // лимит 2 МБ
fwrite($temp, 'Временные данные');
rewind($temp);
$content = stream_get_contents($temp);
fclose($temp); // данные удалены
echo $content;
?>
Как применить фильтр к потоку, например, base64-кодирование на лету?
php://filter позволяет наложить один или несколько фильтров на любой поток. Например, прочитать файл, декодировав его из Base64.
<?php
$path = 'file.b64';
$handle = fopen('php://filter/read=convert.base64-decode/resource=' . $path, 'rb');
$decoded = stream_get_contents($handle);
fclose($handle);
echo $decoded;
?>
Ошибка: если фильтр задан неправильно, поток не откроется. Проверяйте имена фильтров (convert.base64-encode, convert.base64-decode, string.rot13, zlib.deflate и др.).
Как создать свой потоковый обработчик (stream wrapper) для работы с произвольным протоколом?
Пользовательский wrapper позволяет зарегистрировать новый протокол (например, myprotocol://) и определить поведение для функций fopen, fread, fwrite и т.д. Для этого реализуется класс, имплементирующий streamWrapper или его отдельные методы.
<?php
class MyStream {
private $data = [];
private $pos = 0;
private $name;
public function stream_open($path, $mode, $options, &$opened_path) {
$this->name = substr($path, strpos($path, '://') + 3);
$this->data = &$_SESSION['mystream_' . $this->name];
$this->pos = 0;
return true;
}
public function stream_read($count) {
$ret = substr($this->data, $this->pos, $count);
$this->pos += strlen($ret);
return $ret;
}
public function stream_write($data) {
$len = strlen($data);
$this->data = substr_replace($this->data, $data, $this->pos, $len);
$this->pos += $len;
return $len;
}
public function stream_tell() { return $this->pos; }
public function stream_eof() { return $this->pos >= strlen($this->data); }
public function stream_seek($offset, $whence) { /* ... */ }
}
stream_wrapper_register('mysess', 'MyStream');
$fp = fopen('mysess://myvar', 'w+');
fwrite($fp, 'Сохраняем в сессии');
?>
Сложности: отсутствие реализации всех необходимых методов вызовет ошибку. Для корректной работы требуется как минимум stream_open, stream_read/stream_write, stream_close, stream_eof.
Расширенные примеры работы с потоками
Пример 1: Потоковое копирование из удалённого источника в локальный файл
Используется stream_copy_to_stream, которая читает из исходного потока и пишет в целевой, по умолчанию блоками по 8 КБ. Подходит для загрузки больших файлов без использования памяти.
<?php
$src = fopen('http://example.com/largefile.zip', 'rb');
$dst = fopen('/local/path/largefile.zip', 'wb');
$bytes = stream_copy_to_stream($src, $dst);
fclose($src);
fclose($dst);
echo "Скопировано $bytes байт.";
?>
Скопировано 52428800 байт.
При ошибке соединения $src откроется как false, поэтому необходима проверка.
Пример 2: Отправка POST запроса с JSON и обработка ответа
Контекст позволяет задать метод, заголовки, тело и таймаут. Ответ читается через file_get_contents с тем же контекстом.
<?php
$data = json_encode(['username' => 'admin', 'password' => '123']);
$opts = ['http' => [
'method' => 'POST',
'header' => "Content-Type: application/json\r\n" .
"Content-Length: " . strlen($data) . "\r\n",
'content' => $data,
'timeout' => 5
]];
$context = stream_context_create($opts);
$result = @file_get_contents('https://api.example.com/login', false, $context);
if ($result === false) {
$error = error_get_last();
echo "Ошибка: " . $error['message'];
} else {
echo "Ответ сервера: $result";
}
?>
Ответ сервера: {"status":"ok","token":"abc123"}
Пример 3: Пользовательский wrapper для доступа к базе данных через потоки
Регистрируется протокол db://, который позволяет выполнять SQL-запросы при чтении/записи.
<?php
class DbStream {
private $pdo;
private $result;
private $pos;
public function stream_open($path, $mode, $options, &$opened_path) {
$url = parse_url($path);
$query = $url['host']; // имя запроса, например 'select'
// Подключение к БД (упрощённо)
$this->pdo = new PDO('mysql:host=localhost;dbname=test', 'user', 'pass');
if ($mode === 'r') {
$this->result = $this->pdo->query('SELECT * FROM users');
$this->pos = 0;
}
return true;
}
public function stream_read($count) {
$row = $this->result->fetch(PDO::FETCH_ASSOC);
if (!$row) return '';
$encoded = json_encode($row) . "\n";
return substr($encoded, $this->pos, $count);
}
public function stream_eof() {
return $this->result->rowCount() === 0;
}
public function stream_close() {
$this->pdo = null;
}
}
stream_wrapper_register('db', 'DbStream');
$fp = fopen('db://select', 'r');
while (!feof($fp)) {
echo fgets($fp);
}
fclose($fp);
?>
{"id":1,"name":"Иван"}
{"id":2,"name":"Мария"}
Пример 4: Сжатие данных на лету при записи в файл
Обёртка compress.zlib:// с фильтром Deflate позволяет архивировать данные сразу при записи.
<?php
$fp = fopen('compress.zlib://archive.gz', 'wb');
$text = str_repeat('Hello, World! ', 1000);
fwrite($fp, $text);
fclose($fp);
echo 'Файл создан: ' . filesize('archive.gz') . ' байт (было ' . strlen($text) . ' байт)';
?>
Файл создан: 157 байт (было 13000 байт)
Пример 5: Неблокирующее чтение из сокета с помощью потоков
Функция stream_set_blocking отключает блокировку, позволяя обрабатывать данные по мере поступления.
<?php
$socket = stream_socket_client('tcp://localhost:8000', $errno, $errstr, 30);
if (!$socket) die("$errstr ($errno)");
stream_set_blocking($socket, false); // неблокирующий режим
$output = '';
while (true) {
$data = fread($socket, 4096);
if ($data === false) {
// ошибка при чтении
break;
}
if (strlen($data) > 0) {
$output .= $data;
} else {
// нет данных, можно делать другую работу
usleep(10000); // 10 ms
}
// условие выхода, например по таймауту или наличию полного ответа
if (strpos($output, "\r\n\r\n") !== false) break;
}
fclose($socket);
echo $output;
?>