Node.js：SAX解析流数据管道传输至Kafka写入流的实现问题

需求 #

• 从S3获取文件流
• 通过htmlparser2的SaxWritableStream做SAX式XML解析
• 将解析后的结构化数据批量异步写入Kafka，需避免流溢出，正确处理背压

初始实现（无背压控制） #

const content: Readable = getFileFromS3()

const chunkSize: number = 50
const handler = async (data) => ...

const writableKafkaStream = new KafkaWritableStream(chunksSize, handler)

const currentParsingObject = {}

const parser = new SaxWritableStream(
    {
        onopentag(name, attributes) {
            // 根据标签构造数据
            currentParsingObject.someProp = name
        },

        ontext(text) {
            // 根据文本内容构造数据
            currentParsingObject.text = text
        },

        onclosetag(tagname) {
            // 完成数据构造后写入Kafka流
            currentParsingObject.someProp2 = tagname
            writableKafkaStream.write(currentParsingObject)
        },
    },
    { xmlMode: true },
)

return new Promise(resolve => {
    content.pipe(parser).on('finish', () => {
        writableKafkaStream.end()
        resolve(count)
    })
})

初始问题 #

直接调用writableKafkaStream.write()不会触发原始S3流的暂停，当Kafka写入速度跟不上解析速度时，会导致writableKafkaStream内部缓存溢出，内存占用飙升。

改进尝试（手动处理背压） #

为了控制背压，添加了流暂停/恢复逻辑：

onclosetag(tagname) {
    if (shouldAddToWritableKafkaStream) {
        if (!writableKafkaStream.write(data)) {
            content.pause()
            writableKafkaStream.once('drain', () => {
                content.resume()
            })
        }
        count += 1
    }
},

新问题 #

内存占用得到控制，但出现大量警告：

(node:26490) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 drain listeners added to [KafkaTransporterTransformStream]. Use emitter.setMaxListeners() to increase limit

原因是每次write返回false时，都会给writableKafkaStream绑定一个新的drain事件监听，多次触发后监听数量超过Node.js默认的10个上限。

解决方案 #

方案1：复用背压监听逻辑 #

维护一个暂停状态标记，避免重复绑定drain事件：

// 外部维护暂停状态和统一的drain处理函数
let isSourcePaused = false;
const handleDrain = () => {
  content.resume();
  isSourcePaused = false;
};

// 修改SAX解析的onclosetag逻辑
onclosetag(tagname) {
  if (shouldAddToWritableKafkaStream) {
    const canWrite = writableKafkaStream.write(data);
    if (!canWrite && !isSourcePaused) {
      content.pause();
      isSourcePaused = true;
      writableKafkaStream.once('drain', handleDrain);
    } else if (canWrite && isSourcePaused) {
      // 写入恢复且之前处于暂停状态，重置标记
      isSourcePaused = false;
    }
    count += 1;
  }
}

这样只会在流真正需要暂停时绑定一次drain监听，避免监听数量超标。

方案2：使用Transform流作为中间层（推荐） #

通过stream.Transform将SAX解析的事件转换成流数据，利用Node.js流的管道机制自动处理背压：

const { Transform } = require('stream');

// 创建objectMode的Transform流，接收解析后的结构化数据
const saxTransform = new Transform({
  objectMode: true,
  transform(parsedData, _, callback) {
    // 将数据传递给下游Kafka流
    callback(null, parsedData);
  }
});

// 管道连接：Transform流 -> Kafka写入流
saxTransform.pipe(writableKafkaStream);

// 修改SAX解析的回调逻辑
const parser = new SaxWritableStream(
  {
    // ...其他回调逻辑不变
    onclosetag(tagname) {
      // 完成数据构造后写入Transform流
      currentParsingObject.someProp2 = tagname;
      saxTransform.write(currentParsingObject);
    }
  },
  { xmlMode: true }
);

// 原始流管道连接到SAX解析流
return new Promise(resolve => {
  content.pipe(parser).on('finish', () => {
    saxTransform.end();
    resolve(count);
  });
});

这种方式下，当Kafka流无法写入时，Transform流的transform回调会自动暂停，间接阻止SAX解析的速度，无需手动控制原始流的暂停/恢复，从根本上解决背压和监听溢出问题。

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