在 NodeJS 中，数据未在消费者中作为块（流）接收

Question

我有两个服务生产者和消费者。

Producer 在服务器中有很大的 json 文件。我想通过 REST API 通过网络提供服务，我使用 NodeJS 流技术将字节加载到内存中，开销很小，并传递给快速响应对象。我期待这个大 json 文件 读取小字节（即块）并通过网络发送。

消费者点击生产者API来接收字节（块）并在nodejs流中进行处理。

我编写了下面的代码来实现此功能。 但是当我打印 console.log("收到的块") 时。我预计它应该打印多次以表示消费者接收字节（块）。但在控制台日志中只打印一次。

为什么在 Nodejs 流中出现这种行为。根据文档和参考资料，我发现我的实现是正确的，并且我正在本地运行。

我的问题是为什么消费者无法接收字节数据。

// Producer.js

const fs = require('fs');
const path = require('path');
var express = require('express');
var router = express.Router();

/* GET home page. */
router.get('/', function(req, res, next) {
  res.render('index', { title: 'Express' });
});

router.get('/transactions', function(req, res, next) {
  const stream = fs.createReadStream('transactions.json');
  res.setHeader('Content-Type', 'application/json');
  stream.on('data', (chunk) => {
    // Send each chunk as it's read
    res.write(chunk);
  });

  stream.on('end', () => {
    // Signal the end of the response
    res.end();
  });

});

module.exports = router;

// Consumer.js

const { Readable } = require('stream');
const http = require('http');

// Custom Readable Stream class
class JSONResponseStream extends Readable {
  constructor(url) {
    super({ objectMode: true });
    this.url = url;
  }

  _read() {
    http.get(this.url, (res) => {
      let data = '';
      res.on('data', (chunk) => {
        console.log('chunk received'); // It printed only once
        data += chunk;
      });
      res.on('end', () => {
        try {
          const jsonData = JSON.parse(data);
          console.log(typeof jsonData);
          this.push(jsonData); // Push the JSON data into the stream
        } catch (error) {
          this.emit('error', error);
        }
        this.push(null); // No more data to read
      });
    }).on('error', (error) => {
      this.emit('error', error);
    });
  }
}

module.exports = {
    JSONResponseStream
}

// Using the custom stream
const jsonStream = new JSONResponseStream('https://api.example.com/data');
jsonStream.on('data', (data) => {
  console.log('Received data:', data);
});
jsonStream.on('end', () => {
  console.log('No more data.');
});
jsonStream.on('error', (error) => {
  console.error('Error:', error);
});

我的示例 JSON 文件大小为 64 K.B。

我尝试在nodejs中进行流式传输

Answer 1

我一直在寻找答案，终于找到了。原因是节点流默认块大小为 64 K.B，而我的 transactions.json 文件大小小于 64 k.B。因此，当我们从消费者那里读取它时，它收到了一个块，因此只打印一次。

要修改此设置，我们可以在流上设置 hightwatermark 选项，这样我们就可以减少需要处理的 64 K.B 大小。

我希望这个答案能帮助其他人。