Node中的可读流是什么

_readableState:ReadableState{
objectMode:false,//操作除了string、Buffer、null之外的其他类型的数据需要把这个模式打开
highWaterMark:16384,//水位限制，1024*16，默认16kb，超过这个限制则会停止调用_read()读数据到buffer中
buffer:BufferList{head:null,tail:null,length:0},//Buffer链表，用于保存数据
length:0,//整个可读流数据的大小，如果是objectMode则与buffer.length相等
pipes:[],//保存监听了该可读流的所有管道队列
flowing:null,//可独流的状态null、false、true
ended:false,//所有数据消费完毕
endEmitted:false,//结束事件收否已发送
reading:false,//是否正在读取数据
constructed:true,//流在构造好之前或者失败之前，不能被销毁
sync:true,//是否同步触发'readable'/'data'事件，或是等到下一个tick
needReadable:false,//是否需要发送readable事件
emittedReadable:false,//readable事件发送完毕
readableListening:false,//是否有readable监听事件
resumeScheduled:false,//是否调用过resume方法
errorEmitted:false,//错误事件已发送
emitClose:true,//流销毁时，是否发送close事件
autoDestroy:true,//自动销毁，在'end'事件触发后被调用
destroyed:false,//流是否已经被销毁
errored:null,//标识流是否报错
closed:false,//流是否已经关闭
closeEmitted:false,//close事件是否已发送
defaultEncoding:'utf8',//默认字符编码格式
awaitDrainWriters:null,//指向监听了'drain'事件的writer引用，类型为null、Writable、Set
multiAwaitDrain:false,//是否有多个writer等待drain事件
readingMore:false,//是否可以读取更多数据
dataEmitted:false,//数据已发送
decoder:null,//解码器
encoding:null,//编码器
[Symbol(kPaused)]:null
},

constStream=require('stream');
constreadableStream=newS免费云主机、域名tream.Readable({
objectMode:true,
read(){},
});

readableStream.push({name:'lisa'});
console.log(readableStream._readableState.buffer.tail);
readableStream.push(true);
console.log(readableStream._readableState.buffer.tail);
readableStream.push('lisa');
console.log(readableStream._readableState.buffer.tail);
readableStream.push(666);
console.log(readableStream._readableState.buffer.tail);
readableStream.push(()=>{});
console.log(readableStream._readableState.buffer.tail);
readableStream.push(Symbol(1));
console.log(readableStream._readableState.buffer.tail);
readableStream.push(BigInt(123));
console.log(readableStream._readableState.buffer.tail);

constStream=require('stream');

constreadableStream=newStream.Readable({

objectMode:false,

read(){},

});

readableStream.push({name:'lisa'});

constStream=require('stream');

constreadableStream=newStream.Readable();

RS._read=function(size){}

constStream=require('stream');

constreadableStream=newStream.Readable({

read(size){}

});

源码地址：BufferList.consume
https://github.com/nodejs/node/blob/d5e94fa7121c9d424588f0e1a388f8c72c784622/lib/internal/streams/buffer_list.js#L80

//Consumesaspecifiedamountofbytesorcharactersfromthebuffereddata.
consume(n,hasStrings){
constdata=this.head.data;
if(n

源码地址：BufferList._getBuffer
https://github.com/nodejs/node/blob/d5e94fa7121c9d424588f0e1a388f8c72c784622/lib/internal/streams/buffer_list.js#L137

//Consumesaspecifiedamountofbytesfromthebuffereddata.
_getBuffer(n){
constret=Buffer.allocUnsafe(n);
constretLen=n;
letp=this.head;
letc=0;
do{
constbuf=p.data;
if(n>buf.length){
TypedArrayPrototypeSet(ret,buf,retLen-n);
n-=buf.length;
}else{
if(n===buf.length){
TypedArrayPrototypeSet(ret,buf,retLen-n);
++c;
if(p.next)
this.head=p.next;
else
this.head=this.tail=null;
}else{
TypedArrayPrototypeSet(ret,
newUint8Array(buf.buffer,buf.byteOffset,n),
retLen-n);
this.head=p;
p.data=buf.slice(n);
}
break;
}
++c;
}while((p=p.next)!==null);
this.length-=c;
returnret;
}

源码地址：BufferList._getString
https://github.com/nodejs/node/blob/d5e94fa7121c9d424588f0e1a388f8c72c784622/lib/internal/streams/buffer_list.js#L106

//Consumesaspecifiedamountofcharactersfromthebuffereddata.
_getString(n){
letret='';
letp=this.head;
letc=0;
do{
conststr=p.data;
if(n>str.length){
ret+=str;
n-=str.length;
}else{
if(n===str.length){
ret+=str;
++c;
if(p.next)
this.head=p.next;
else
this.head=this.tail=null;
}else{
ret+=StringPrototypeSlice(str,0,n);
this.head=p;
p.data=StringPrototypeSlice(str,n);
}
break;
}
++c;
}while((p=p.next)!==null);
this.length-=c;
returnret;
}

这部分源码在这儿：stream/legacy
https://github.com/nodejs/node/blob/d5e94fa7121c9d424588f0e1a388f8c72c784622/lib/internal/streams/legacy.js#L10

functionStream(opts){
EE.call(this,opts);
}
ObjectSetPrototypeOf(Stream.prototype,EE.prototype);
ObjectSetPrototypeOf(Stream,EE);

这部分源码在这儿：readable
https://github.com/nodejs/node/blob/d5e94fa7121c9d424588f0e1a388f8c72c784622/lib/internal/streams/readable.js#L77

ObjectSetPrototypeOf(Readable.prototype,Stream.prototype);
ObjectSetPrototypeOf(Readable,Stream);

Stream{
destroy:[Function:destroy],
_undestroy:[Function:undestroy],
_destroy:[Function(anonymous)],
push:[Function(anonymous)],
unshift:[Function(anonymous)],
isPaused:[Function(anonymous)],
setEncoding:[Function(anonymous)],
read:[Function(anonymous)],
_read:[Function(anonymous)],
pipe:[Function(anonymous)],
unpipe:[Function(anonymous)],
on:[Function(anonymous)],
addListener:[Function(anonymous)],
removeListener:[Function(anonymous)],
off:[Function(anonymous)],
removeAllListeners:[Function(anonymous)],
resume:[Function(anonymous)],
pause:[Function(anonymous)],
wrap:[Function(anonymous)],
iterator:[Function(anonymous)],
[Symbol(nodejs.rejection)]:[Function(anonymous)],
[Symbol(Symbol.asyncIterator)]:[Function(anonymous)]
}

Readable.prototype.push=function(chunk,encoding){
returnreadableAddChunk(this,chunk,encoding,false);
};

functionreadableAddChunk(stream,chunk,encoding,addToFront){
conststate=stream._readableState;
if(chunk===null){//pushnull流结束信号，之后不能再写入数据
state.reading=false;
onEofChunk(stream,state);
}elseif(!state.objectMode){//如果不是对象模式
if(typeofchunk==='string'){
chunk=Buffer.from(chunk);
}elseif(chunkinstanceofBuffer){//如果是Buffer
//处理一下编码
}elseif(Stream._isUint8Array(chunk)){
chunk=Stream._uint8ArrayToBuffer(chunk);
}elseif(chunk!=null){
err=newERR_INVALID_ARG_TYPE('chunk',['string','Buffer','Uint8Array'],chunk);
}
}

if(state.objectMode||(chunk&&chunk.length>0)){//是对象模式或者chunk是Buffer
//这里省略几种数据的插入方式的判断
addChunk(stream,state,chunk,true);
}
}

functionaddChunk(stream,state,chunk,addToFront){
if(state.flowing&&state.length===0&&!state.sync&&
stream.listenerCount('data')>0){//如果处于流动模式，有监听data的订阅者
stream.emit('data',chunk);
}else{//否则保存数据到缓冲区中
state.length+=state.objectMode?1:chunk.length;
if(addToFront){
state.buffer.unshift(chunk);
}else{
state.buffer.push(chunk);
}
}
maybeReadMore(stream,state);//尝试多读一点数据
}

//如果read的长度大于hwm，则会重新计算hwm
if(n>state.highWaterMark){
state.highWaterMark=computeNewHighWaterMark(n);
}
//调用用户实现的_read方法
try{
constresult=this._read(state.highWaterMark);
if(result!=null){
constthen=result.then;
if(typeofthen==='function'){
then.call(
result,
nop,
function(err){
errorOrDestroy(this,err);
});
}
}
}catch(err){
errorOrDestroy(this,err);
}

functionfromList(n,state){
//nothingbuffered.
if(state.length===0)
returnnull;
letret;
if(state.objectMode)
ret=state.buffer.shift();
elseif(!n||n>=state.length){//处理n为空或者大于缓冲区的长度的情况
//Readitall,truncatethelist.
if(state.decoder)//有解码器，则将结果序列化为字符串
ret=state.buffer.join('');
elseif(state.buffer.length===1)//只有一个数据，返回头节点数据
ret=state.buffer.first();
else//将所有数据存储到一个Buffer中
ret=state.buffer.concat(state.length);
state.buffer.clear();//清空缓冲区
}else{
//处理读取长度小于缓冲区的情况
ret=state.buffer.consume(n,state.decoder);
}
returnret;
}

constStream=require('stream');
constreadableStream=newStream.Readable({
read(hwm){
this.push(String.fromCharCode(this.currentCharCode++));
if(this.currentCharCode>122){
this.push(null);
}
},
});
readableStream.currentCharCode=97;
readableStream.pipe(process.stdout);
//abcdefghijklmnopqrstuvwxyz%

Readable.prototype.pipe=function(dest,pipeOpts){
constsrc=this;
conststate=this._readableState;
state.pipes.push(dest);//收集Writable流

src.on('data',ondata);
functionondata(chunk){
constret=dest.write(chunk);
if(ret===false){
pause();
}
}
//Tellthedestthatit'sbeingpipedto.
dest.emit('pipe',src);
//启动流，如果流处于暂停模式
if(dest.writableNeedDrain===true){
if(state.flowing){
pause();
}
}elseif(!state.flowing){
src.resume();
}
returndest;
}

Readable.prototype.resume=function(){
conststate=this._readableState;
if(!state.flowing){
state.flowing=!state.readableListening;//是否处于流动模式取决于是否设置了'readable'监听句柄
resume(this,state);
}
};

functionresume(stream,state){
if(!state.resumeScheduled){//开关，使resume_方法仅在同一个Tick中调用一次
state.resumeScheduled=true;
process.nextTick(resume_,stream,state);
}
}

functionresume_(stream,state){
if(!state.reading){
stream.read(0);
}
state.resumeScheduled=false;
stream.emit('resume');
flow(stream);
}

functionflow(stream){//当流处于流模式该方法会不断的从buffer中读取数据，直到缓冲区为空
conststate=stream._readableState;
while(state.flowing&&stream.read()!==null);
//因为这里会调用read方法，设置了'readable'事件监听器的stream，也有可能会调用read方法，
//从而导致数据不连贯（不影响data，仅影响在'readable'事件回调中调用read方法读取数据）
}

Readable.prototype.pause=function(){
if(this._readableState.flowing!==false){
debug('pause');
this._readableState.flowing=false;
this.emit('pause');
}
returnthis;
};