javascript - 卡夫卡 : Reading off __consumer_offsets with node-rdkafka

Question

有没有办法使用 javascript 解码主题 __consumer_offsets 中的数据？

现在我有这样的东西:

consumer.on('data: ', function(data) {
  if(data.value) {
    console.log('data.value', data.value.toString())
  }
})

并打印类似的东西数据值:消费者范围，rdkafka-78bfd5c4-2a48-4573-ae6b-5f5957332b0b，rdkafka-78bfd5c4-2a48-4573-ae6b-5f5957332b0brdkafka/192.168.7.190u0u0__consumer_offsets__consum er_offsets2

Answer 1

它看起来不像 node-rdkafka(或 kafka-node 那样)包含必要的消息格式化程序，所以你自己写，可能不会直接。

您最好的选择可能是简单地从 child_process 中的命令行读取它。

要运行的命令如下所示:

kafka-console-consumer --consumer.config /tmp/consumer.config \
  --formatter "kafka.coordinator.GroupMetadataManager\$OffsetsMessageFormatter" \
  --zookeeper localhost:2181 \
  --topic __consumer_offsets

它可以作为子进程产生，看起来像这样:

import { spawn } from 'child_process';

const child = spawn('kafka-console-consumer', ['--formatter "kafka.coordinator.GroupMetadataManager\$OffsetsMessageFormatter"', '--zookeeper', 'localhost:2181', '--topic', '__consumer_offsets'], { stdio: 'pipe' });

// create a handleData and handleError function
child.stdout.on('data', handleData);
child.stderr.on('data', handleError);

如果不编写您自己的消息格式化程序来转换该主题的二进制消息，这可能是最简单的方法。

这里有更多关于读取 Kafka 消费者偏移量的信息:https://www.ctheu.com/2017/08/07/looking-at-kafka-s-consumers-offsets/#consuming-consumer-offsets