Node.js Transform流数据未转换问题排查（基于through2）

首先，最关键的错误在这里：你对pipe()的返回值理解错了！

看你的代码：

const stream = through2.obj(/* 第一个转换逻辑 */)
  .pipe(through2.obj(/* 第二个转换逻辑 */))
  .pipe(process.stdout);

pipe()方法返回的是目标流，而不是源流。所以这里的stream变量最后指向的是process.stdout，而不是你第一个创建的through2转换流！当你调用stream.write(data)时，相当于直接把原始数据写到了标准输出，完全绕过了前面两个转换流——这就是为什么数据看起来“直接穿过”的原因！

修正方法： #

把第一个转换流单独保存，后续的pipe链挂在它后面，然后往第一个转换流写数据：

// 先创建第一个转换流，保存为sourceStream
const sourceStream = through2.obj(function (chunk, enc, callback) {
  callback(null, parseString(chunk))
});
// 然后链式pipe后续流
sourceStream
  .pipe(through2.obj(function (chunk, enc, callback) {
    callback(null, JSON.stringify(chunk));
  }))
  .pipe(process.stdout);

// 往sourceStream写数据，而不是最后一个pipe的结果
socket.on('m', data => sourceStream.write(data));

其他需要排查的点： #

解决了上面的问题后，还要注意parseString函数的this指向问题：
你的parseString里用到了this.FIELDS，但当你在through2的回调里直接调用parseString(chunk)时，this的上下文是through2的Transform对象，而不是你定义parseString时的原始上下文（比如如果parseString是某个类的方法，或者某个对象的属性）。这会导致this.FIELDS为undefined，进而让parseString返回一个空对象，最终输出{}。

修正this指向的方法：

1. 如果parseString是某个对象的方法，用bind绑定上下文：

// 假设parseString是obj的方法
const boundParse = obj.parseString.bind(obj);
// 在through2里调用boundParse
through2.obj(function (chunk, enc, callback) {
  callback(null, boundParse(chunk))
})

2. 或者把FIELDS作为参数传入parseString，改成纯函数：

function parseString(value, FIELDS) {
  var valuesArray = value.split("~");
  var valuesArrayLenght = valuesArray.length;
  var mask = valuesArray[valuesArrayLenght - 1];
  var maskInt = parseInt(mask, 16);
  var unpackedCurrent = {};
  var currentField = 0;
  for (var property in FIELDS) {
    if (FIELDS[property] === 0) {
      unpackedCurrent[property] = valuesArray[currentField];
      currentField++;
    } else if (maskInt & FIELDS[property]) {
      if (property === 'LASTMARKET') {
        unpackedCurrent[property] = valuesArray[currentField];
      } else {
        unpackedCurrent[property] = parseFloat(valuesArray[currentField]);
      }
      currentField++;
    }
  }
  return unpackedCurrent;
};
// 调用时传入FIELDS
through2.obj(function (chunk, enc, callback) {
  callback(null, parseString(chunk, yourFieldsObject))
})

额外调试技巧： #

在每个转换流里加日志，确认数据是否经过处理：

const sourceStream = through2.obj(function (chunk, enc, callback) {
  console.log('第一个转换流收到的chunk:', chunk);
  const parsed = parseString(chunk);
  console.log('转换后的数据:', parsed);
  callback(null, parsed);
});

这样可以直观看到每个步骤的数据变化，快速定位哪里出了问题。

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