如何通过node-rdkafka跟踪主题队列状态并实现优先级消费

针对你的场景，这里提供几个实用方案来跟踪主题待消费消息状态，进而实现高优先级主题优先消费的逻辑：

1. 通过queryWatermarkOffsets计算分区待消费消息数 #

node-rdkafka的消费者实例提供了queryWatermarkOffsets方法，可获取指定分区的高水位偏移量（HW），结合消费者已提交的偏移量，就能算出该分区的待消费消息数。将所有分区的数值累加，即可得到整个主题的待消费总量。

示例代码：

const Kafka = require('node-rdkafka');

async function getTopicPendingMessages(consumer, topic) {
  // 获取主题的所有分区
  const metadata = consumer.getMetadata({ topic });
  const partitions = metadata.topics.find(t => t.name === topic).partitions.map(p => p.id);
  
  let totalPending = 0;
  for (const partition of partitions) {
    // 获取当前分区的已提交偏移量
    const committed = await new Promise((resolve, reject) => {
      consumer.committed([{ topic, partition }], 5000, (err, offsets) => {
        if (err) reject(err);
        else resolve(offsets[0].offset);
      });
    });
    
    // 获取分区的水位偏移量
    const { high } = consumer.queryWatermarkOffsets(topic, partition, 5000);
    
    // 计算当前分区待消费消息数（高水位 - 已提交偏移量，负数取0）
    const pending = Math.max(high - committed, 0);
    totalPending += pending;
  }
  return totalPending;
}

// 使用示例
const consumer = new Kafka.KafkaConsumer({
  'group.id': 'your-group-id',
  'metadata.broker.list': 'kafka-broker:9092'
});

consumer.connect();
consumer.on('ready', async () => {
  consumer.subscribe(['high-priority-topic']);
  // 定期检查待消费消息数
  setInterval(async () => {
    const pending = await getTopicPendingMessages(consumer, 'high-priority-topic');
    if (pending === 0) {
      // 高优先级主题无消息，切换到低优先级主题
      consumer.unsubscribe();
      consumer.subscribe(['low-priority-topic']);
    }
  }, 1000); // 按需调整检查间隔
  
  consumer.consume({
    callback: (err, messages) => {
      if (err) console.error(err);
      // 处理消息逻辑
      messages.forEach(msg => {
        console.log('处理消息:', msg.value.toString());
        // 手动提交偏移量（如果配置了手动提交）
        consumer.commitMessage(msg);
      });
    }
  });
});

2. 监听stats事件获取实时Lag数据 #

node-rdkafka消费者会定期触发stats事件，返回的统计数据中包含每个主题分区的lag（待消费消息数），可直接解析这个字段判断主题是否有未处理消息。

需要先在消费者配置中开启统计功能：

const consumer = new Kafka.KafkaConsumer({
  'group.id': 'your-group-id',
  'metadata.broker.list': 'kafka-broker:9092',
  'statistics.interval.ms': 1000 // 每1秒触发一次stats事件
});

然后监听stats事件解析数据：

let highPriorityTopic = 'high-priority-topic';
let currentTopic = highPriorityTopic;

consumer.on('stats', (stats) => {
  const statsObj = JSON.parse(stats);
  // 找到当前订阅主题的所有分区lag
  const topicPartitions = statsObj.consumer.assignment[currentTopic];
  if (!topicPartitions) return;
  
  const totalLag = Object.values(topicPartitions).reduce((sum, partition) => {
    return sum + partition.lag;
  }, 0);
  
  if (totalLag === 0 && currentTopic === highPriorityTopic) {
    // 切换到低优先级主题
    consumer.unsubscribe();
    currentTopic = 'low-priority-topic';
    consumer.subscribe([currentTopic]);
  } else if (totalLag > 0 && currentTopic !== highPriorityTopic) {
    // 高优先级主题有新消息，切回去（可选逻辑）
    consumer.unsubscribe();
    currentTopic = highPriorityTopic;
    consumer.subscribe([currentTopic]);
  }
});

// 启动消费
consumer.on('ready', () => {
  consumer.subscribe([highPriorityTopic]);
  consumer.consume({
    callback: (err, messages) => {
      if (err) console.error(err);
      // 处理消息逻辑
      messages.forEach(msg => {
        console.log('处理消息:', msg.value.toString());
        consumer.commitMessage(msg);
      });
    }
  });
});

consumer.connect();

注意事项 #

• queryWatermarkOffsets是同步方法，频繁调用可能影响性能，建议合理设置检查间隔。
• stats事件的统计数据存在一定延迟，适合做非实时的状态判断。
• 如果采用多实例单主题部署，可能需要借助外部调度组件（如K8s的HPA、自定义调度服务）来根据主题消息量调整实例数量，而非单个实例切换订阅。

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