标签: nodejs-stream

多个进程使用 writestream 将数据写入同一个文件是否可以？性能影响和数据丢失怎么样？

var fs = require('fs');
var cluster = require('cluster');
if(cluster.isMaster){
for(var i=0;i<10; i++)
{
    var child = cluster.fork();
}
}
else{
var stream = fs.createWriteStream('data.txt', {flags:'a'});
stream.on('error', function(error){
    console.log(process.pid+' error occured', error);
})
stream.write(process.pid+' msg\n');
//lots of writes
}

我正在使用npm install socket.io-stream\n 我在我的角度组件上实现socket.io-stream如下所示：

import * as io from 'socket.io-client';\nimport * as ss from 'socket.io-stream';\n

我只想创建一个像这样的双工流：

stream=ss.createStream();\n

当我运行时出现此错误ng build：

ERROR in ./node_modules/socket.io-stream/lib/iostream.js\nModule not found: Error: Can't resolve 'stream' in ' \n'C:\\Users\\geoffroy\\Documents\\Mines Al\xc3\xa8s\\2A\\Stage\\WebService based \nGUI\\WebApp\\node_modules\\socket.io-stream\\lib'\nERROR in ./node_modules/socket.io-stream/lib/blob-read-stream.js\nModule not found: Error: Can't resolve 'stream' in \n'C:\\Users\\geoffroy\\Documents\\Mines Al\xc3\xa8s\\2A\\Stage\\WebService based \nGUI\\WebApp\\node_modules\\socket.io-stream\\lib'\n

我不明白，因为在我的 server.js 上它似乎有效..

我尝试运行该命令npm install stream，但没有出现错误ng build。但是当我启动我的应用程序时，我的浏览器控制台中出现了一个新错误：

inherits_browser.js:5 Uncaught TypeError: Cannot read property 'prototype' of …

我有一个 lambda 函数，它返回 base64 字符串，当我从它的代码中调用 lambda 时，它可以工作，但是当我在 ALB 后面调用 lambda 并且 base64 字符串很大时，ALB 给我错误 502 Bad Gateway。注意：对于小字符串 ALB 也有效。

// Lambda function handler

'use strict';


module.exports.handler = async (event, context) => {
  // ALB expects following response format
  // see: https://docs.aws.amazon.com/lambda/latest/dg/services-alb.html
  const response = {
    headers: {
      'Access-Control-Allow-Origin': '*',
      'Content-Type': 'application/json',
    },
    isBase64Encoded: true,
    statusCode: 200,
    statusDescription: '200 OK',
  };
// ALB gives error 502 when buffer size is large
  const answer = 'This is my audio buffer'.toString('base64');
  response.body = …

请注意，有许多转换流可以执行此操作：

JSON -> JS

但我希望创建一个可以执行以下操作的 Node.js 转换流：

JS -> JSON

我有一个可读的流：

const readable = getReadableStream({objectMode:true});

可读流输出对象，而不是字符串。

我需要创建一个转换流，它可以过滤其中一些对象并将这些对象转换为 JSON，如下所示：

const t = new Transform({
  objectMode: true,
  transform(chunk, encoding, cb) {
    if(chunk && chunk.marker === true){
       this.push(JSON.stringify(chunk));
     }
    cb();
  },
  flush(cb) {
    cb();
  }
});

但是，由于某种原因，我的转换流无法接受转换方法的对象，只能接受字符串和缓冲区，我该怎么办？

我尝试添加这两个选项：

  const t = new Transform({
      objectMode: true,
      readableObjectMode: true,  // added this
      writableObjectMode: true,  // added this too
      transform(chunk, encoding, cb) {
        this.push(chunk);
        cb();
      },
      flush(cb) {
        cb();
      }
    });

不幸的是，我的转换流仍然不能接受对象，只能接受字符串/缓冲区。

使用 Express，如何确定响应已发送/响应已完全写入？

app.use((req,res,next) => {

   if(res.ended){
     //
   }

   if(res.finished){
     //
   }

});

我怎么知道是否res.end()已被调用？我正在寻找一个可以在 res 上阅读的布尔值。

我正在创建一个 Firebase HTTP 函数，该函数进行 BigQuery 查询并返回查询结果的修改版本。该查询可能会返回数百万行，因此在响应 HTTP 客户端之前我无法将整个查询结果存储在内存中。我正在尝试使用 Node.js 流，并且由于我需要在将结果发送到客户端之前修改结果，因此我正在尝试使用转换流。但是，当我尝试通过转换流传输查询流时，Firebase 函数崩溃并显示以下错误消息：finished with status: 'response error'。

我的最小可重现示例如下。我正在使用缓冲区，因为我不想一次处理一行（块），因为我需要进行异步网络调用来转换数据。

return new Promise((resolve, reject) => {
    const buffer = new Array(5000)
    let bufferIndex = 0
    const [job] = await bigQuery.createQueryJob(options)
    const bqStream = job.getQueryResultsStream()

    const transformer = new Transform({
        writableObjectMode: true,
        readableObjectMode: false,
        transform(chunk, enc, callback) {
            buffer[bufferIndex] = chunk
            if (bufferIndex < buffer.length - 1) {
                bufferIndex++
            }
            else {
                this.push(JSON.stringify(buffer).slice(1, -1)) // Transformation should happen here.
                bufferIndex = 0
            } …

这个问题我在面试中遇到过很多次，但我仍然找不到答案。我已经用谷歌搜索过，但没有找到确切的答案。请任何人告诉我这个问题的确切答案。

问题：NodeJs 中的错误代码是什么？

我回答了错误状态代码。但面试官说，我不是问状态码。请解释答案。先感谢您。

我尝试将数据插入 mongodb。它已成功插入，但我面临错误的文档插入 mongodb 的问题。如何使用 mongodb 中的更新查询来更新文档中的键和值。

Json数据格式

{
             "brand": "LG",
             "colour": "Black",
             "item height": "14 Millimeters",
             "item width": "14.1 Centimeters",
             "item weight": "200 g",
             "product dimensions": "13.5 x 14.1 x 1.4 cm",
             "item model number": "GP65NB60",
             "included components": "Power2Go, PowerBackup, YouCam, LabelPrint and Norton internet security (60 days trial)"
}

如何使用 mongodb 中的更新查询将关键前品牌更新为品牌。

我正在使用 Twilio 语音流功能，但我不想使用 Twilio 录音功能。当 Twilio 开始向我的服务器发送语音流时，我想将其作为音频文件实时存储到磁盘中。

我正在逐行读取 CSV 文件并在 MongoDB 中插入/更新。预期的输出将是 1。 console.log(row); 2. console.log(光标); 3.console.log("流");

但是得到像 1.console.log(row); 这样的输出。控制台日志（行）；控制台日志（行）；控制台日志（行）；控制台日志（行）；…………………… 2. console.log(cursor); 3.console.log("流"); 请让我知道我在这里缺少什么。

const csv = require('csv-parser');
const fs = require('fs');

var mongodb = require("mongodb");

var client = mongodb.MongoClient;
var url = "mongodb://localhost:27017/";
var collection;
client.connect(url,{ useUnifiedTopology: true }, function (err, client) {

  var db = client.db("UKCompanies");
  collection = db.collection("company");
  startRead();
});
var cursor={};

async function insertRec(row){
  console.log(row);
  cursor = await collection.update({CompanyNumber:23}, row, {upsert: true});
  if(cursor){
    console.log(cursor);
  }else{
    console.log('not exist')
  }
  console.log("stream");
}



async function startRead() …

我正在实现一个由猫鼬光标进行管道传输的转换流（其工作方式类似于可读流，（或者可能是真正的可读流））

const myTransform = new Transform({
  readableObjectMode: true,
  writableObjectMode: true,
  transform(chunk: Document, encoding, callback) {
    this.push(chunk?.toObject() ?? null); // <- transforms data
    callback(); // <- calls callback
  },
});

MyMongooseModelWithHugeData.find()
  .cursor()
  .pipe(myTransform)
  .pipe(someWriteStream)

是的。我当前的代码工作正常。

但我发现callback内部transform实现，接收第二个参数（显然看起来像经典的nodejs异步回调风格）。

所以，我改变了我的代码，如下所示，发现它也工作得很好。

const myTransform = new Transform({
  readableObjectMode: true,
  writableObjectMode: true,
  transform(chunk: Document, encoding, callback) {
    callback(null, chunk?.toObject() ?? null); // <- transforms data and calls callback.
  },
});

我感到有些尴尬。我搜索了一些有关创建转换流的博客。我发现的所有文章都告诉我必须使用this.push()和调用callback()。但他们甚至没有提到 的第二个参数callback()。

所以，我的问题是：

• this.push()传递转换值时和传递转换值时有区别吗 …

setTimeout(()=>{
        console.log('time out')
    },3000)
}

go();
console.log('app')

这是异步代码，我想在延迟后打印应用程序，但是我们知道先打印“应用程序”，然后再打印“超时”。