The Consumer API

ไอเดียในหนึ่งประโยค

Kafka consumer จะ subscribe topic แล้ว รัน consume loop — KafkaJS เรียก handler eachMessage ของคุณด้วย record จาก partition ที่ถูก assign ให้ และ loop ที่รันอยู่นี้เองที่คอยทำให้ consumer ยังมีชีวิตอยู่ใน group

subscribe ก่อน แล้วค่อย run

คุณสร้าง consumer จาก Kafka client พร้อม groupId, connect(), subscribe() ไปยัง topic ที่ต้องการ แล้วเริ่ม run() KafkaJS จะส่งแต่ละ record ให้ callback eachMessage ของคุณ โดย key และ value มาเป็น Buffer ดิบ — คุณ deserialize เอง (เช่น .toString() หรือ JSON.parse) เป็นภาพสะท้อนของฝั่ง producer ที่ serialize

import { Kafka } from 'kafkajs';

const kafka = new Kafka({ clientId: 'order-processor', brokers: ['localhost:9092'] });
const consumer = kafka.consumer({ groupId: 'order-processor' }); // consumer group membership

async function main() {
  await consumer.connect();
  await consumer.subscribe({ topic: 'orders', fromBeginning: false }); // false = only new records (latest)

  // run() drives the poll loop for you and calls eachMessage per record
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      // key/value are Buffers — deserialize them yourself
      console.log(
        `p=${partition} off=${message.offset} key=${message.key?.toString()} val=${message.value?.toString()}`,
      );
    },
  });
}

main();

ทำไม run loop ถึงสำคัญ

run() ทำมากกว่าแค่ดึงข้อมูล เบื้องหลังจะคอย poll Kafka, ส่ง heartbeat, เข้าร่วม group coordination และดึง partition assignment ให้ ถ้า eachMessage ของคุณ block นานเกินไป — process record หนึ่งนานกว่า session timeout — group coordinator จะถือว่า consumer ตายแล้ว และสั่ง rebalance ย้าย partition ไปให้ตัวอื่น กฎคือ คุมเวลา process ต่อ record ให้จำกัด

flowchart LR
  sub["subscribe(topics)"] --> run["consumer.run(...)"]
  run --> each["eachMessage per record"]
  each --> proc["process the record"]
  proc --> run
  run -->|also| hb["send heartbeats + coordinate group"]

The consumer run loop

แต่ละ record พก coordinate ของตัวเองมา

message ทุกตัวเปิดเผย topic, partition, offset, key, value, timestamp และ headers ออกมา coordinate เหล่านี้คือสิ่งที่คุณใช้ commit ความคืบหน้า และใช้คิดเรื่องลำดับ — consumer จะเห็น record เรียงตาม offset ภายในแต่ละ partition แม้ว่า record จาก partition ต่างกันอาจสลับกันได้

🌊 สรุปหนึ่งบรรทัด

consumer จะ subscribe() ไปยัง topic แล้วเริ่ม consumer.run({ eachMessage }) ซึ่งขับ poll loop ให้คุณ — heartbeat, group coordination, ดึง assignment — ดังนั้นถ้า handler ของคุณ block นานเกินไป group จะ rebalance ย้าย partition ของคุณออกไป record มาถึง เรียงตาม offset ภายใน partition โดยแต่ละตัวติดป้าย topic, partition และ offset และคุณ deserialize key กับ value เอง

consumer.run() (poll loop) ทำอะไรอีกนอกจากคืน record?

ไม่ทำอะไรอีก — แค่ดึงข้อมูลอย่างเดียว ยังส่ง heartbeat, ประสาน group membership และดึง partition assignment ด้วย commit offset ลงฐานข้อมูล สร้าง partition ใหม่

ถ้า consumer หยุด consume (handler block) นานเกินไปจะเกิดอะไรขึ้น?

Kafka จะเร่งความเร็วในการส่ง group coordinator จะถือว่าตายแล้ว และ rebalance ย้าย partition ไปให้ consumer ตัวอื่น consumer จะ scale up อัตโนมัติ ไม่มีอะไรเกิดขึ้น การ consume เป็นทางเลือก

จะได้ key หรือ value ที่มี type จาก message ของ KafkaJS ได้ยังไง?

บีบอัด record ก่อนส่ง key และ value มาเป็น Buffer — คุณ deserialize เอง (เช่น toString หรือ JSON.parse) assign partition เอง commit offset ก่อน

consumer เห็น record เรียงตามลำดับแบบไหน?

สุ่มลำดับทั่วทั้ง topic เรียงตาม offset ภายในแต่ละ partition แม้ partition ต่าง ๆ อาจสลับกันข้าม batch ตัวใหม่สุดก่อนเสมอ เรียงตาม key

Submit

Mark as complete