如何关闭可读流（结束前）？答案

【问题标题】：How to close a readable stream (before end)?如何关闭可读流（结束前）？
【发布时间】：2013-10-17 02:39:47
【问题描述】：

如何在 Node.js 中关闭 readable stream？

var input = fs.createReadStream('lines.txt');

input.on('data', function(data) {
   // after closing the stream, this will not
   // be called again

   if (gotFirstLine) {
      // close this stream and continue the
      // instructions from this if
      console.log("Closed.");
   }
});

这会比：

input.on('data', function(data) {
   if (isEnded) { return; }

   if (gotFirstLine) {
      isEnded = true;
      console.log("Closed.");
   }
});

但这不会停止阅读过程...

【问题讨论】：

警告：这个问题只在fs模块的上下文中。 close 在Stream.Readable 中不存在。
好消息。节点版本 8 提供stream.destroy()
你不能打电话给readable.push(null) && readable.destroy();

标签： node.js stream fs

【解决方案1】：

调用input.close()。它不在文档中，但是

https://github.com/joyent/node/blob/cfcb1de130867197cbc9c6012b7e84e08e53d032/lib/fs.js#L1597-L1620

显然可以完成这项工作 :) 它实际上做了类似于您的 isEnded 的事情。

EDIT 2015-Apr-19 基于以下 cmets，并澄清和更新：

此建议是 hack，未记录在案。
虽然查看当前的 lib/fs.js，但 >1.5 年后仍然有效。
我同意下面关于致电destroy() 更可取的评论。
如下所述，这适用于fs ReadStreams，不适用于通用Readable

至于通用解决方案：至少从我对文档的理解和快速浏览 _stream_readable.js 来看，它似乎没有一个。

我的建议是将您的可读流置于暂停模式，至少可以防止在您的上游数据源中进行进一步处理。不要忘记unpipe() 并删除所有data 事件侦听器，以便pause() 实际上暂停，如the docs 中所述

【讨论】：

其实我更喜欢打电话给destroy。至少如果您将 autoClose 设置为 true，就会调用它。通过查看源代码（今天），差异很小（destroy 调用 close）但将来可能会改变
现在不记得了，但看起来像 :)
可读对象上没有close()，有没有永远的解决方案？我的数据交换总是不完整...
已更新以澄清、解决 cmets 并为通用案例提供（穷人的）建议。尽管不强制通用readable 实现close() 并提供特定于类的方式来执行此操作确实有意义（如fs 中的情况，可能还有其他实现Readable 的类）
暂停不会导致上游（发送者）由于背压而阻塞，或者导致缓冲区增长直到超过其限制？理想情况下，我们会告诉发件人它不再需要...

【解决方案2】：

编辑：好消息！从 Node.js 8.0.0 开始readable.destroy 正式可用：https://nodejs.org/api/stream.html#stream_readable_destroy_error

ReadStream.destroy

>

您可以随时调用ReadStream.destroy函数。

var fs = require('fs');

var readStream = fs.createReadStream('lines.txt');
readStream
    .on('data', function (chunk) {
        console.log(chunk);
        readStream.destroy();
    })
    .on('end', function () {
        // This may not been called since we are destroying the stream
        // the first time 'data' event is received
        console.log('All the data in the file has been read');
    })
    .on('close', function (err) {
        console.log('Stream has been destroyed and file has been closed');
    });

公共函数 ReadStream.destroy 没有文档记录（Node.js v0.12.2），但您可以查看 source code on GitHub (Oct 5, 2012 commit)。

destroy 函数在内部将ReadStream 实例标记为已销毁，并调用close 函数释放文件。

您可以收听close event 以准确了解文件何时关闭。除非数据完全消耗，否则end event 不会触发。

请注意，destroy（和close）函数特定于fs.ReadStream。没有部分通用的stream.readable“接口”。

【讨论】：

至少在最新版本的Node中（其他的没查过），文件描述符是closed automatically。也就是说，我没有进行任何彻底的测试来确保流最终会触发error，如果它从未被读取过。除此之外，我唯一担心的其他泄漏是事件处理程序——再一次，我不是 100% 确定这一点，但我们可能没问题，因为 2010 年艾萨克斯的福音确实说处理程序是当发射器被 gc'd 时被修剪：groups.google.com/d/msg/nodejs/pXbJVo0NtaY/BxUmF_jp9LkJ
如果数据太小，on('data')只会触发一次，所以不会有.close()，只是提醒别人。
您实际上可以使用this.destroy()，除非您使用的是箭头函数。词汇this我恨你：D

【解决方案3】：

你不能。从节点 5.3.0 开始，没有记录的方法可以关闭/关闭/中止/销毁通用 Readable 流。这是 Node 流架构的限制。

正如此处的其他答案所解释的，对于 Node 提供的 Readable 的特定实现，存在未记录的黑客攻击，例如 fs.ReadStream。不过，这些不是任何 Readable 的通用解决方案。

如果有人可以在这里证明我错了，请这样做。我希望能够做我所说的不可能的事情，并且很高兴得到纠正。

编辑：这是我的解决方法：implement .destroy() for my pipeline though a complex series of unpipe() calls. 在所有复杂性之后，它是doesn't work properly in all cases。

编辑：节点 v8.0.0 添加了destroy() api for Readable streams。

【讨论】：

现在有stream.pipeline，它声称可以处理“转发错误并正确清理并在管道完成时提供回调”。这有帮助吗？

【解决方案4】：

您可以使用yourstream.resume() 清除和关闭流，这将转储流中的所有内容并最终关闭它。

来自official docs：

可读的.resume():

返回：这个

此方法将导致可读流继续发射“数据”事件。

此方法会将流切换到流动模式。如果您不想使用流中的数据，但确实想获得其“结束”事件，则可以调用 stream.resume() 来打开数据流。

var readable = getReadableStreamSomehow();
readable.resume();
readable.on('end', () => {
  console.log('got to the end, but did not read anything');
});

【讨论】：

这可以称为“排干”流。在我们的例子中，当然我们有一个'data' 事件监听器，但是我们让它检查一个布尔值if (!ignoring) { ... }，所以当我们排空流时它不会处理数据。 ignoring = true; readable.resume();
当然这假设流在某个时候会'end'。并非所有流都会这样做！（例如，永远每秒发送日期的流。）

【解决方案5】：

在4.*.* 版本中，将空值推送到流中将触发EOF 信号。

来自nodejs docs

如果传入的不是null，push()方法会在队列中加入一块数据，供后续流处理器消费。如果传递了 null，则表示流结束 (EOF)，在此之后不能再写入数据。

在此页面上尝试了许多其他选项后，这对我有用。

【讨论】：

为我工作。但是，我需要避免在推送 null 后调用 done() 回调以获得预期的行为 - 即整个流停止。

【解决方案6】：

这是一个老问题，但我也在寻找答案，并找到了最适合我实施的答案。 end 和 close 事件都会发出，所以我认为这是最干净的解决方案。

这将在节点 4.4.* 中解决问题（撰写本文时的稳定版本）：

var input = fs.createReadStream('lines.txt');

input.on('data', function(data) {
   if (gotFirstLine) {
      this.end(); // Simple isn't it?
      console.log("Closed.");
   }
});

更详细的解释见： http://www.bennadel.com/blog/2692-you-have-to-explicitly-end-streams-after-pipes-break-in-node-js.htm

【讨论】：

【解决方案7】：

这里的代码可以很好地解决问题：

function closeReadStream(stream) {
    if (!stream) return;
    if (stream.close) stream.close();
    else if (stream.destroy) stream.destroy();
}

writeStream.end() 是关闭 writeStream 的首选方法...

【讨论】：

为什么你提到 .end() 是首选方式，但是你的代码使用 close 和 destroy 甚至不使用 end？
我正在关闭示例中的 readStream... writeStream -- 使用 .end

【解决方案8】：

这个destroy 模块旨在确保流被销毁，处理不同的 API 和 Node.js 错误。现在是最好的选择之一。

注意。从节点 10 开始，您可以使用 .destroy 方法而无需进一步依赖。

【讨论】：

【解决方案9】：

今天，在节点 10

readableStream.destroy()

是关闭可读流的官方方式

见https://nodejs.org/api/stream.html#stream_readable_destroy_error

【讨论】：

【解决方案10】：

在某些调用后停止回调执行，您必须使用具有特定 processID 的 process.kill

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

const filepath = "./demo.csv"
let readStream = fs.createReadStream(filepath, {
    autoClose: true,
});
let MAX_LINE = 0;


readStream.on('error', (e) => {
        console.log(e);
        console.log("error");
    })

    .pipe(csv())
    .on('data', (row) => {

        if (MAX_LINE == 2) {
            process.kill(process.pid, 'SIGTERM')
        }
        // console.log("not 2");
        MAX_LINE++
        console.log(row);
    })

    .on('end', () => {
        // handle end of CSV
        console.log("read done");
    }).on("close", function () {
        console.log("closed");
    })

【讨论】：