无需第三方库实现NodeJS直接调用Kafka完成poll请求的步骤有哪些

核心前提 #

Kafka 所有客户端与broker的交互都基于TCP端口（默认9092）的私有二进制协议，无需依赖HTTP或第三方库，核心工作是手动实现协议的序列化/反序列化、TCP流处理、请求响应匹配即可。

完整实现步骤 #

• 步骤1：建立TCP基础连接 #

  用NodeJS原生net模块对接broker的TCP端口，TCP是流式协议，需要自行处理粘包问题：Kafka所有请求/响应的前4字节是整包的长度（大端序），每次读取时先读4字节获取整包大小，再读取对应长度的字节作为完整的协议包。如果是SSL加密的集群，将net替换为tls模块加载对应证书即可。

• 步骤2：API版本协商（可选但推荐） #

  发送ApiVersions请求（协议ApiKey=18），获取broker支持的所有API的版本范围，避免后续请求使用broker不兼容的协议版本导致报错。

• 步骤3：身份验证 #

  常用认证方式的实现逻辑：
  • 无认证集群：直接跳过该步骤
  • SASL/PLAIN认证：先发送SaslHandshake请求（ApiKey=17）指定认证机制为PLAIN，再发送格式为\0用户名\0密码的认证串，收到broker的成功响应即完成认证
  • SASL/SCRAM/ Kerberos认证：需要按照对应机制实现多轮挑战-响应交互流程

• 步骤4：获取集群元数据 #

  发送Metadata请求（ApiKey=3），指定要消费的topic名称，拿到该topic的分区分布、每个分区的leader broker地址，后续Fetch请求必须发往对应分区的leader节点才能拿到数据。

• 步骤5：消费者组流程（仅使用消费者组时需要，独立消费者可跳过） #

  如果要加入消费者组实现负载均衡、位移托管，需要完成以下流程：
  1. 发送JoinGroup请求（ApiKey=11），指定消费者组ID、分区分配策略，从broker拿到成员ID、分配到的分区列表
  2. 发送SyncGroup请求（ApiKey=14），确认分区分配结果
  3. 定期发送Heartbeat请求（ApiKey=12）保持会话，避免被broker判定为离线踢出消费组

• 步骤6：订阅分区与Poll实现 #

  我们常说的poll本质就是定期发送Fetch请求（ApiKey=1）的封装：
  1. 订阅阶段：自主指定要消费的分区和起始offset（独立消费者），或用消费者组分配到的分区列表
  2. 构造Fetch请求体，填入要拉取的分区、对应分区的起始offset、最大拉取字节数、max_wait_ms（长轮询等待时间，broker没有足够消息时最多等待的时长）
  3. 发送请求拿到响应后，反序列化消息集，处理完成后更新本地消费offset，需要提交位移的话额外发送OffsetCommit请求即可

最小演示示例（无认证、独立消费者场景） #

注意：本示例仅做底层原理演示，省略了CRC校验、错误重试、多分区兼容、协议版本适配等生产级能力，仅供学习使用

const net = require('net');
const { Buffer } = require('buffer');

// 配置项
const KAFKA_HOST = '127.0.0.1';
const KAFKA_PORT = 9092;
const TOPIC_NAME = 'test-topic';
const PARTITION = 0;
const CLIENT_ID = 'native-test-consumer';
const CORRELATION_ID = 1; // 请求匹配ID，每次请求递增即可

// 构造Metadata请求（查询topic元数据）
function buildMetadataRequest() {
  // 请求头：ApiKey=3(Metadata), ApiVersion=0, CorrelationId, ClientId
  const clientIdBuf = Buffer.from(CLIENT_ID);
  const header = Buffer.alloc(2 + 2 + 4 + 2 + clientIdBuf.length);
  let offset = 0;
  header.writeInt16BE(3, offset); offset +=2; // ApiKey=Metadata
  header.writeInt16BE(0, offset); offset +=2; // ApiVersion=0
  header.writeInt32BE(CORRELATION_ID, offset); offset +=4; // CorrelationId
  header.writeInt16BE(clientIdBuf.length, offset); offset +=2; // ClientId长度
  clientIdBuf.copy(header, offset); offset += clientIdBuf.length;

  // 请求体：topic数量 + topic名
  const body = Buffer.alloc(4 + 2 + Buffer.from(TOPIC_NAME).length);
  let bodyOffset = 0;
  body.writeInt32BE(1, bodyOffset); bodyOffset +=4; // 1个topic
  const topicBuf = Buffer.from(TOPIC_NAME);
  body.writeInt16BE(topicBuf.length, bodyOffset); bodyOffset +=2;
  topicBuf.copy(body, bodyOffset); bodyOffset += topicBuf.length;

  // 整包长度 = 头长度 + 体长度
  const totalSize = header.length + body.length;
  const fullReq = Buffer.alloc(4 + totalSize);
  fullReq.writeInt32BE(totalSize, 0);
  header.copy(fullReq, 4);
  body.copy(fullReq, 4 + header.length);
  return fullReq;
}

// 构造Fetch请求（即Poll的核心请求）
function buildFetchRequest(offset = 0) {
  const clientIdBuf = Buffer.from(CLIENT_ID);
  // 请求头：ApiKey=1(Fetch), ApiVersion=0
  const header = Buffer.alloc(2 + 2 + 4 + 2 + clientIdBuf.length);
  let offset = 0;
  header.writeInt16BE(1, offset); offset +=2;
  header.writeInt16BE(0, offset); offset +=2;
  header.writeInt32BE(CORRELATION_ID + 1, offset); offset +=4;
  header.writeInt16BE(clientIdBuf.length, offset); offset +=2;
  clientIdBuf.copy(header, offset); offset += clientIdBuf.length;

  // 简化版Fetch请求体
  const body = Buffer.alloc(4 + 4 + 4 + 4 + 2 + Buffer.from(TOPIC_NAME).length + 4 + 4 + 8 + 4);
  let bodyOff = 0;
  body.writeInt32BE(-1, bodyOff); bodyOff +=4; // replica_id=-1
  body.writeInt32BE(500, bodyOff); bodyOff +=4; // max_wait_ms=500ms
  body.writeInt32BE(1024, bodyOff); bodyOff +=4; // min_bytes=1KB
  body.writeInt32BE(1, bodyOff); bodyOff +=4; // 1个topic
  const topicBuf = Buffer.from(TOPIC_NAME);
  body.writeInt16BE(topicBuf.length, bodyOff); bodyOff +=2;
  topicBuf.copy(body, bodyOff); bodyOff += topicBuf.length;
  body.writeInt32BE(1, bodyOff); bodyOff +=4; // 1个分区
  body.writeInt32BE(PARTITION, bodyOff); bodyOff +=4; // 分区ID
  body.writeBigInt64BE(BigInt(offset), bodyOff); bodyOff +=8; // 起始offset
  body.writeInt32BE(1024 * 1024, bodyOff); bodyOff +=4; // 最大拉取字节

  const totalSize = header.length + body.length;
  const fullReq = Buffer.alloc(4 + totalSize);
  fullReq.writeInt32BE(totalSize, 0);
  header.copy(fullReq, 4);
  body.copy(fullReq, 4 + header.length);
  return fullReq;
}

// 建立TCP连接
const client = net.createConnection(KAFKA_PORT, KAFKA_HOST, () => {
  console.log('已连接到Kafka broker');
  // 先发送元数据请求
  client.write(buildMetadataRequest());
});

let remaining = 0;
let buffer = Buffer.alloc(0);
client.on('data', (data) => {
  buffer = Buffer.concat([buffer, data]);
  while (true) {
    if (remaining === 0) {
      if (buffer.length < 4) break;
      remaining = buffer.readInt32BE(0);
      buffer = buffer.slice(4);
    }
    if (buffer.length < remaining) break;
    const res = buffer.slice(0, remaining);
    buffer = buffer.slice(remaining);
    remaining = 0;

    // 处理响应
    const correlationId = res.readInt32BE(0);
    if (correlationId === CORRELATION_ID) {
      console.log('元数据请求成功，开始发送Fetch请求(Poll)');
      // 从offset 0开始拉取消息
      client.write(buildFetchRequest(0));
    } else if (correlationId === CORRELATION_ID + 1) {
      console.log('拿到Poll响应，消息原始字节长度：', res.length);
      // 此处可添加消息反序列化逻辑解析消息内容
    }
  }
});

client.on('error', (err) => {
  console.error('连接错误：', err);
});

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