Node.js 环境下向 ksqlDB 发起 HTTP2 请求的实现问题咨询

你编写的简化版代码读取不到 chunk 是 Node.js 流的工作模式导致的，核心遗漏点如下：

• http2 返回的 ClientHttp2Stream 默认处于暂停模式，你同步执行 while 循环调用 stream.read() 时，HTTP 请求还未收到响应，底层没有可用数据，read() 直接返回 null，循环直接退出，不会等待数据返回。
• 没有监听流的 readable 事件，只有该事件触发时调用 read() 才能获取到已到达的响应数据。
• 响应拼接逻辑错误：parseChunk 返回的是行数组，直接用 += 拼接会把数组转为逗号分隔的字符串，破坏原始的行结构，后续无法正常解析每行JSON数据。
• 没有监听 end 事件判断单次响应是否全部接收完成。

针对单次请求的轻量需求，我们可以用 Promise 封装整个请求流程，不需要复杂的异步生成器逻辑：

const http2 = require('http2');

const DELIMITED_CONTENT_TYPE = 'application/vnd.ksqlapi.delimited.v1';

const queryKsqlDB = (queryOptions) => {
  return new Promise((resolve, reject) => {
    const session = http2.connect('http://localhost:8088');
    const stream = session.request({
      [http2.constants.HTTP2_HEADER_PATH]: '/query-stream',
      [http2.constants.HTTP2_HEADER_METHOD]: 'POST',
      [http2.constants.HTTP2_HEADER_CONTENT_TYPE]: DELIMITED_CONTENT_TYPE,
    });

    const reqPayload = Buffer.from(JSON.stringify(queryOptions));
    stream.end(reqPayload);

    const rows = [];
    let buffer = ''; // 存储未拆分完的半行数据，适配网络分包场景

    stream
      .on('error', (error) => {
        session.destroy();
        reject(error);
      })
      .on('close', () => session.destroy())
      .on('abort', () => {
        session.destroy();
        reject(new Error('请求被中断'))
      })
      .on('timeout', () => {
        session.destroy();
        reject(new Error('请求超时'))
      })
      // 有可用数据时触发，此时读取流才能拿到有效内容
      .on('readable', () => {
        let chunk;
        while ((chunk = stream.read()) !== null) {
          // 拼接上一次残留的半行数据
          buffer += chunk.toString();
          const lines = buffer.split('\n');
          // 最后一行可能不完整，留到下一次拼接
          buffer = lines.pop() || '';
          // 过滤空行后加入结果集
          rows.push(...lines.filter(l => l.trim()));
        }
      })
      // 所有数据接收完成
      .on('end', () => {
        // 处理最后残留的半行数据
        if (buffer.trim()) {
          rows.push(buffer.trim());
        }
        session.destroy();
        // 可按需在这里添加元数据解析、行数据格式化逻辑
        resolve(rows);
      });
  });
};

// 调用示例
const main = async () => {
  try {
    const result = await queryKsqlDB({
      sql: `SELECT * FROM test_view where name='john';`,
    });
    // ksqlDB返回的第一行是查询元数据，后续是行数据，按需解析即可
    console.log('查询元数据：', JSON.parse(result[0]));
    console.log('查询结果行：', result.slice(1).map(r => JSON.parse(r)));
  } catch (e) {
    console.error('请求失败：', e);
  }
};

main();

关键改动说明 #

1. 用 Promise 封装整个请求流程，适配异步调用习惯，接收完所有数据后统一返回结果。
2. 监听 readable 事件触发时才读取流数据，保证能读到已到达的 chunk。
3. 增加了半行数据的兼容处理：网络传输中可能会把一行拆成多个 chunk 发送，通过 buffer 存储未拆分完成的半行数据，避免解析错误。
4. 流结束时处理最后残留的半行数据，保证不会丢失结果。
5. 所有异常场景都会主动销毁 session 避免资源泄漏，同时抛出错误方便上层捕获。

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