Node.js é conhecido por sua capacidade de lidar com operações de E/S com eficiência, e no coração dessa capacidade está o conceito de conceito de fluxos. Os fluxos permitem que você processe dados de dados por peça, em vez de carregar tudo na memória de uma só vez-perfeito para lidar com arquivos grandes, solicitações de rede ou dados em tempo real. Quando você combina fluxos com a forte digitação do TypeScript, você obtém uma combinação poderosa: o desempenho atende à segurança.

Neste guia, mergulharemos profundamente no Node.js, exploraremos seus tipos e passaremos por exemplos práticos usando o TypeScript. Seja você um novato node.js ou um entusiasta do texto digital que procura subir de nível, este post você cobriu.

Por que os fluxos são importantes?

Imagine isto: você tem a tarefa de processar um arquivo de log de 50 GB. Carregá -lo inteiramente na memória esgotaria os recursos do seu servidor, levando a falhas ou desempenho lento. Os fluxos resolvem isso, permitindo que você lide com dados enquanto eles fluem, como beber de uma palha em vez de tomar um jarro de galão.

Essa eficiência é por que os fluxos são uma pedra angular do Node.js, ligando tudo, desde operações de arquivo a servidores HTTP. O TypeScript aprimora isso adicionando definições de tipo, capturando erros no momento da compilação e melhorando a legibilidade do código. Vamos mergulhar nos fundamentos e ver como essa sinergia funciona na prática.

Os quatro tipos de fluxos

Node.js oferece Quatro tipos de fluxo principalcada um com um propósito específico:

1. Fluxos legíveis: Fontes de dados que você pode ler (por exemplo, arquivos, respostas HTTP).
2. Fluxos graváveis: Destinos para os quais você pode escrever (por exemplo, arquivos, solicitações HTTP).
3. Fluxos duplex: Tanto legível quanto gravável (por exemplo, soquetes TCP).
4. Transformar fluxos: Um fluxo duplex especial que modifica os dados à medida que passa (por exemplo, compactação).

O TypeScript aprimora isso, permitindo -nos definir interfaces para os dados que fluem através deles. Vamos dividi -los com exemplos.

Configurando seu ambiente de texto datilografado

Antes de mergulharmos no código, verifique se você possui o Node.js e o TypeScript instalado.

Crie um novo projeto:

mkdir node-streams-typescript
cd node-streams-typescript
npm init -y
npm install typescript @types/node --save-dev
npx tsc --init

Atualize seu tsconfig.json para incluir:

{
  "compilerOptions": {
    "target": "ES2020",
    "module": "commonjs",
    "strict": true,
    "outDir": "./dist"
  },
  "include": ("src/**/*")
}

Crie uma pasta SRC e vamos começar a codificar!

Exemplo 1: lendo um arquivo com um fluxo legível

Vamos ler um pedaço de arquivo de texto por Chunk. Primeiro, crie um arquivo chamado data.txt no diretório raiz do seu projeto com algum texto de amostra (por exemplo, “Olá, fluxos!”).

Agora, no SRC/Readstream.ts:

import { createReadStream } from 'fs';
import { Readable } from 'stream';

const readStream: Readable = createReadStream('data.txt', { encoding: 'utf8' });

readStream
  .on('data', (chunk: string) => {
    console.log('Chunk received:', chunk);
  })
  .on('end', () => {
    console.log('Finished reading the file.');
  })
  .on('error', (err: Error) => {
    console.error('Error:', err.message);
  });

Execute -o com:

npx tsc && node dist/readStream.js

Aqui, o TypeScript garante que o pedaço adere à nossa interface de chunk e o manipulador de eventos de erro espera um tipo de erro. Este fluxo lê data.txt em pedaços (padrão de 64kb para arquivos) e os registra.

Exemplo 2: Escrevendo dados com um fluxo gravável

Agora, vamos escrever dados em um novo arquivo. Em src/writestream.ts:

import { createWriteStream } from 'fs';
import { Writable } from 'stream';

const writeStream: Writable = createWriteStream('output.txt', { encoding: 'utf8' });

const data: string() = ('Line 1n', 'Line 2n', 'Line 3n');

data.forEach((line: string) => {
  writeStream.write(line);
});

writeStream.end(() => {
  console.log('Finished writing to output.txt');
});

writeStream.on('error', (err: Error) => {
  console.error('Error:', err.message);
});

Compilar e executar:

npx tsc && node dist/writeStream.js

Isso cria saída.txt com três linhas. O TypeScript garante que a linha seja uma string e forneça conchamagem automática para métodos de fluxo.

Exemplo 3: tubulação com um fluxo de transformação

A tubulação é onde os fluxos brilham, conectando um fluxo legível a um fluxo gravável. Vamos adicionar uma torção com um Transformar Transmita para manchas de texto em maiúsculas.

Em src/transformstream.ts:

import { createReadStream, createWriteStream } from 'fs';
import { Transform, TransformCallback } from 'stream';


class UppercaseTransform extends Transform {
  _transform(chunk: Buffer, encoding: string, callback: TransformCallback): void {
    const upperChunk = chunk.toString().toUpperCase();
    this.push(upperChunk);
    callback();
  }
}

const readStream = createReadStream('data.txt', { encoding: 'utf8' });
const writeStream = createWriteStream('output_upper.txt');
const transformStream = new UppercaseTransform();

readStream
  .pipe(transformStream)
  .pipe(writeStream)
  .on('finish', () => {
    console.log('Transform complete! Check output_upper.txt');
  })
  .on('error', (err: Error) => {
    console.error('Error:', err.message);
  });

Execute:

npx tsc && node dist/transformStream.js

Isso lê data.txt, transforma o texto em maiúsculas e grava em output_upper.txt.

O tipo de transformCallback do TypeScript garante que nosso método _transform seja implementado corretamente.

Exemplo 4: Compressionando arquivos com um fluxo duplex

Vamos abordar um cenário mais avançado: comprimindo um arquivo usando o módulo ZLIB, que fornece um fluxo duplex. Ele vem com o pacote ‘@types/nó’, que instalamos anteriormente.

Em src/compressstream.ts:

import { createReadStream, createWriteStream } from 'fs';
import { createGzip } from 'zlib';
import { pipeline } from 'stream';

const source = createReadStream('data.txt');
const destination = createWriteStream('data.txt.gz');
const gzip = createGzip();

pipeline(source, gzip, destination, (err: Error | null) => {
  if (err) {
    console.error('Compression failed:', err.message);
    return;
  }
  console.log('File compressed successfully! Check data.txt.gz');
});

Execute:

npx tsc && node dist/compressStream.js

Aqui, o pipeline garante o manuseio e a limpeza de erros adequados. O fluxo GZIP comprime data.txt em data.txt.gz. A inferência do tipo de datilografia mantém nosso código limpo e seguro.

Exemplo 5: Streaming HTTP Responses

Os fluxos brilham nas operações de rede. Vamos simular dados de streaming de um servidor HTTP usando AXIOS. Instale:

npm install axios @types/axios

Em src/httpstream.ts:

import axios from 'axios';
import { createWriteStream } from 'fs';
import { Writable } from 'stream';

async function streamHttpResponse(url: string, outputFile: string): Promisevoid> {
  const response = await axios({
    method: 'get',
    url,
    responseType: 'stream',
  });

  const writeStream: Writable = createWriteStream(outputFile);
  response.data.pipe(writeStream);

  return new Promise((resolve, reject) => {
    writeStream.on('finish', () => {
      console.log(`Downloaded to ${outputFile}`);
      resolve();
    });
    writeStream.on('error', (err: Error) => {
      console.error('Download failed:', err.message);
      reject(err);
    });
  });
}

streamHttpResponse('https://example.com', 'example.html').catch(console.error);

Execute:

npx tsc && node dist/httpStream.js

Isso transmite uma resposta HTTP (por exemplo, uma página da web) para exemplo.html. O TypeScript garante que os parâmetros de URL e saída de saída sejam strings, e a promessa de digitação adiciona clareza.

Também podemos usar a API de busca embutida do Node.js (disponível desde o nó V18) ou bibliotecas como a Fetch Node, que também suportam respostas de streaming, embora os tipos de fluxo possam diferir (fluxos de fluxos da Web vs. Node.js).

Exemplo:

const response = await fetch('https://example.com');
const writeStream = createWriteStream(outputFile);
response.body.pipe(writeStream);

Exemplo 6: Processamento de dados em tempo real com um fluxo legível personalizado

Vamos criar um fluxo personalizado e legível para simular dados em tempo real, como leituras de sensores. Em SRC/CustomReadable.ts:

import { Readable } from 'stream';

class SensorStream extends Readable {
  private count: number = 0;
  private max: number = 10;

  constructor(options?: any) {
    super(options);
  }

  _read(): void {
    if (this.count  this.max) {
      const data = `Sensor reading ${this.count}: ${Math.random() * 100}n`;
      this.push(data);
      this.count++;
    } else {
      this.push(null); 
    }
  }
}

const sensor = new SensorStream({ encoding: 'utf8' });

sensor
  .on('data', (chunk: string) => {
    console.log('Received:', chunk.trim());
  })
  .on('end', () => {
    console.log('Sensor stream complete.');
  })
  .on('error', (err: Error) => {
    console.error('Error:', err.message);
  });

Execute:

npx tsc && node dist/customReadable.js

Isso gera 10 “leituras de sensores” aleatórias e as transmite. A digitação de classe do TypeScript garante que nossa implementação alinhe com a interface legível.

Exemplo 7: encadeamento de vários fluxos de transformação

Vamos se transformar em cadeia para processar o texto em etapas: maiúsculas e depois prenda um registro de data e hora. Em SRC/Chaintransform.ts:

import { createReadStream, createWriteStream } from 'fs';
import { Transform, TransformCallback } from 'stream';

class UppercaseTransform extends Transform {
  _transform(chunk: Buffer, encoding: string, callback: TransformCallback): void {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
}

class TimestampTransform extends Transform {
  _transform(chunk: Buffer, encoding: string, callback: TransformCallback): void {
    const timestamp = new Date().toISOString();
    this.push(`(${timestamp}) ${chunk.toString()}`);
    callback();
  }
}

const readStream = createReadStream('data.txt', { encoding: 'utf8' });
const writeStream = createWriteStream('output_chain.txt');
const upper = new UppercaseTransform();
const timestamp = new TimestampTransform();

readStream
  .pipe(upper)
  .pipe(timestamp)
  .pipe(writeStream)
  .on('finish', () => {
    console.log('Chained transform complete! Check output_chain.txt');
  })
  .on('error', (err: Error) => {
    console.error('Error:', err.message);
  });

Execute:

npx tsc && node dist/chainTransform.js

Isso lê data.txt, uppercases os dados, adiciona um registro de data e hora e grava o resultado em output_chain.txt. As transformações de encadeamento mostram a modularidade de Streams.

Melhores práticas para fluxos no TypeScript

1. Digite seus dados: Defina interfaces para os pedaços para capturar erros do tipo mais cedo.
2. Lidar com erros: Sempre anexe os ouvintes de eventos de erro para evitar exceções não tratadas.
3. Use tubos com sabedoria: A tubulação reduz o manuseio manual de eventos e melhora a legibilidade.
4. Backpressure: Para dados grandes, Monitor WriteStream.WritableHighwatermark para evitar sobrecarregar o destino.

Caso de uso do mundo real: streaming de respostas da API

Imagine que você está construindo uma API que transmita um conjunto de dados grande. Usando expressos e fluxos:

import express from 'express';
import { Readable } from 'stream';

const app = express();

app.get('/stream-data', (req, res) => {
  const data = ('Item 1n', 'Item 2n', 'Item 3n');
  const stream = Readable.from(data);

  res.setHeader('Content-Type', 'text/plain');
  stream.pipe(res);
});

app.listen(3000, () => {
  console.log('Server running on port 3000');
});

Instale as dependências (NPM Install Express @Types/Express) e execute -o. Visita Para ver o fluxo de dados no seu navegador!

Dicas avançadas: manuseando a contrapressão

Quando um fluxo gravável não consegue acompanhar um fluxo legível, ocorre a contrapressão. O Node.js lida com isso automaticamente com tubos, mas você pode monitorá -lo manualmente:

const writeStream = createWriteStream('large_output.txt');

if (!writeStream.write('data')) {
  console.log('Backpressure detected! Pausing...');
  writeStream.once('drain', () => {
    console.log('Resuming...');
  });
}

Isso garante que seu aplicativo permaneça responsivo sob cargas pesadas.

Precauções para usar a contrapressão: Ao escrever grandes quantidades de dados, o fluxo legível pode produzir dados mais rapidamente do que o fluxo gravável pode consumi -los. Enquanto o tubo e o pipeline lidam com isso automaticamente, se gravar manualmente, verifique se a write () retorna falsa e aguarde o evento ‘drenagem’ antes de escrever mais.

Além disso, os iteradores assíncronos (para aguardar … de) são alternativas modernas para o consumo de fluxos legíveis, que geralmente podem simplificar o código em comparação com o uso de .on (‘dados’) e .on (‘end’).

Exemplo:

async function processStream(readable: Readable) {
  for await (const chunk of readable) {
    console.log('Chunk:', chunk);
  }
  console.log('Finished reading.');
}

Pontos adicionais:

Certifique -se de limpeza de recursos: Isso é especialmente importante nas implementações de fluxo personalizado ou ao usar o stream.pipline. Ligue explicitamente a Stream.Destroy () em cenários de erro ou quando o fluxo não é mais necessário para liberar recursos subjacentes e evitar vazamentos. Stream.Pipline lida com isso automaticamente para fluxos canalizados.

Use readertable.From () por conveniência: Quando você precisa criar um fluxo a partir de um iterável existente (como uma matriz) ou um iterável assíncrono, readável.From () geralmente é a abordagem mais simples e moderna, exigindo um código menos caldeiro do que criar uma classe legível personalizada.

Conclusão

Fluxos são um divisor de águas no Node.js, e o TypeScript os aprimora ainda mais, introduzindo a segurança e a clareza do tipo. De leitura de arquivos à transformação de dados em tempo real, o domínio dos fluxos abre um mundo de possibilidades de E/S eficientes. Os exemplos aqui – lendo, escrevendo, alterando, comprimindo e transmitindo sobre HTTP – critica a superfície do que é possível.

Experimente seus próprios pipelines: tente transmitir logs, processar arquivos CSV ou criar um sistema de bate -papo ao vivo. Quanto mais você explora, mais apreciará a versatilidade dos fluxos.