如何将蓝牙设备的JSON数组对象持续推入Node.js Readable对象流？

首先，先明确你的核心需求：每次蓝牙收到包含10个对象的JSON数组后，把每个独立对象推到对象模式的可读流，供下游流消费。不用特意用PassThrough流，直接用Readable流（对象模式）就可以搞定，你的问题主要出在可读流的_read实现、索引处理和流的推送逻辑上。

先分析你之前的错误点 #

你第一次的代码里chunk.toString()报错，原因有两个：

1. 你的s1stream是对象模式的流，chunk是JavaScript对象，不是Buffer/字符串，所以没有toString()方法；
2. 索引n1++的顺序错了，n1++ < minLength会先判断再自增，导致你推的是s1[n1]（此时n1已经自增了，第一次循环时n1变成1，取的是索引1的元素，而不是0，甚至最后会越界拿到undefined，这就是chunk未定义的根源）。

优化后的完整实现 #

我们可以简化逻辑，同时解决这些问题，还能完美适配蓝牙事件触发的场景（不是固定的setInterval）：

1. 基础可读流封装（适配蓝牙事件触发） #

首先，创建一个可复用的对象可读流，当蓝牙收到数据包时，把数组里的每个对象逐个推入：

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

// 创建一个对象模式的可读流
class SensorReadableStream extends Readable {
  constructor(options = {}) {
    super({ ...options, objectMode: true });
    // 缓存待推送的对象队列
    this.queue = [];
  }

  // 实现_read方法，对象模式下可以是空实现，因为我们主动push数据
  _read() {}

  // 接收蓝牙传来的JSON数组，把对象加入队列并推送
  pushBatch(dataArray) {
    this.queue.push(...dataArray);
    this._processQueue();
  }

  // 处理队列，处理流的背压问题（避免推送过快导致内存溢出）
  _processQueue() {
    while (this.queue.length > 0) {
      // push返回false表示流已满，暂停推送
      if (!this.push(this.queue.shift())) {
        // 当流可以继续接收时，再恢复处理
        this.once('drain', () => this._processQueue());
        break;
      }
    }
  }

  // 结束流的方法
  endStream() {
    this.push(null);
  }
}

2. 蓝牙事件绑定与流消费 #

然后，绑定蓝牙接收事件，每当收到数据包时调用pushBatch，同时用through2处理对象：

// 初始化传感器流
const sensorStream = new SensorReadableStream();

// 替换成你实际的蓝牙接收回调函数
function onBluetoothDataReceived(rawData) {
  // 解析JSON数组（假设rawData是字符串格式的JSON）
  const dataArray = JSON.parse(rawData);
  // 把数组中的对象批量推入流
  sensorStream.pushBatch(dataArray);
}

// 下游流处理：比如把对象的record和timestamp字段转成字符串输出
const transformStream = through2.obj((obj, enc, callback) => {
  this.push(`记录ID: ${obj.record} | 时间戳: ${obj.timestamp}\n`);
  callback();
});

// 管道连接，把处理后的内容输出到控制台
sensorStream.pipe(transformStream).pipe(process.stdout);

// 示例：模拟接收一次蓝牙数据包（实际场景中删除这段，替换成真实的蓝牙事件）
const mockBluetoothData = `[{"record":0,"sensor":1,"timestamp":26566,"date":{"day":7,"hour":10,"minute":45,"month":5,"second":38,"year":18}}, {"record":1,"sensor":1,"timestamp":26567,"date":{"day":7,"hour":10,"minute":45,"month":5,"second":38,"year":18}}, {"record":2,"sensor":1,"timestamp":26568,"date":{"day":7,"hour":10,"minute":45,"month":5,"second":38,"year":18}}, {"record":3,"sensor":1,"timestamp":26569,"date":{"day":7,"hour":10,"minute":45,"month":5,"second":38,"year":18}}, {"record":4,"sensor":1,"timestamp":26570,"date":{"day":7,"hour":10,"minute":45,"month":5,"second":38,"year":18}}, {"record":5,"sensor":1,"timestamp":26571,"date":{"day":7,"hour":10,"minute":45,"month":5,"second":38,"year":18}}, {"record":6,"sensor":1,"timestamp":26572,"date":{"day":7,"hour":10,"minute":45,"month":5,"second":38,"year":18}}, {"record":7,"sensor":1,"timestamp":26573,"date":{"day":7,"hour":10,"minute":45,"month":5,"second":38,"year":18}}, {"record":8,"sensor":1,"timestamp":26574,"date":{"day":7,"hour":10,"minute":45,"month":5,"second":38,"year":18}}, {"record":9,"sensor":1,"timestamp":26575,"date":{"day":7,"hour":10,"minute":45,"month":5,"second":38,"year":18}}]`;
onBluetoothDataReceived(mockBluetoothData);

针对你原有代码的修复说明 #

你后来的可运行代码解决了_read的问题（用read() {}空实现），但还有索引错误：

• n2++ < minLength会导致你推送的是s2[n2]（此时n2已经自增，第一次取的是索引1的元素，而不是0），正确的写法应该是先取元素再自增：

  var n2 = 0;
  const timer = setInterval(function() {
    if (n2 < minLength) {
      s2stream.push(s2[n2]);
      n2++; // 推送后再自增索引
    } else if (n2 === minLength) {
      s2stream.push(null);
      clearInterval(timer); // 记得清除定时器，避免重复执行
    }
  }, 1000);
  

• 另外，对象模式的流不要直接调用chunk.toString()，因为chunk是对象，你可以提取对象的属性再转字符串，比如chunk.record.toString()就没问题。

关键要点总结 #

• 一定要开启objectMode: true，这样可读流可以直接推送JavaScript对象，不用转成Buffer；
• _read方法在对象模式下可以是空实现，因为我们是主动推送数据，不是等待流拉取；
• 处理背压：当push()返回false时，要等待drain事件再继续推送，避免内存溢出；
• 索引操作要注意顺序，先取元素再自增，避免越界或取到undefined。

内容的提问来源于stack exchange，提问作者apeman