xkafka — วิธีใช้ Kafka ใน Go ให้ง่ายขึ้น

• xkafka เป็นไลบรารีโอเพนซอร์สที่ช่วยให้ใช้งาน Kafka ในสภาพแวดล้อม Go ได้ง่ายเหมือนบริการ HTTP
• เดิมทีการใช้ confluent-kafka-go ต้องมีลูปจัดการที่ซับซ้อนและโค้ด boilerplate จำนวนมาก แต่ xkafka ใช้โครงสร้าง Handler, Middleware, Message เพื่อให้โฟกัสกับ ตรรกะหลัก ได้มากขึ้น
• รองรับ การ publish และ consume message ได้อย่างตรงไปตรงมา คล้ายรูปแบบ request/response ของ HTTP และซ่อนความซับซ้อนของ Kafka ไว้มาก เช่น การจัดการ offset, การตั้งค่า concurrency, การจัดการ error
• รองรับแพตเทิร์นที่ต้องใช้ในระบบจริงได้ง่าย เช่น Streaming/Batch, การประมวลผลแบบลำดับ/แบบ asynchronous, การรับประกันแบบ At-most-once/At-least-once
• ใช้งานแพตเทิร์นที่จำเป็นในงานจริงได้ง่าย เช่น การจัดการ error แบบหลายชั้น, retry/logging/metrics ที่อิง middleware

---

Kafka แบบ HTTP-like

• xkafka เป็นไลบรารีที่ abstraction Kafka ใน Go ให้อยู่ในรูปแบบคล้ายบริการ HTTP
  • Message คล้ายกับ HTTP request โดยมี topic/partition/offset/key/value/header/callback เป็นต้น
  • Handler ทำหน้าที่ประมวลผล business logic คล้าย HTTP Handler
  • Middleware แยกความสามารถเสริมอย่าง logging, metrics, retry ออกจาก business logic และนำไปใช้ได้สะดวก

การ publish message (Publishing Messages)

• สร้าง Producer ด้วย xkafka.NewProducer จากนั้นสร้างอ็อบเจ็กต์ message แล้ว publish ด้วยฟังก์ชัน Publish
• รองรับการ publish แบบ asynchronous (AsyncPublish) และการลงทะเบียน callback จึงเหมาะกับงานประสิทธิภาพสูงหรือการประมวลผล event แบบ asynchronous
• มี goroutine เบื้องหลังคอยจัดการการส่ง message และสามารถติดตามสถานะการส่งผ่าน callback ได้

การ consume message (Consuming Messages)

• ตอนสร้าง Consumer สามารถกำหนด Handler ฟังก์ชัน รวมถึง topic/broker/config ต่าง ๆ ได้
• เพิ่ม middleware ได้ด้วย consumer.Use()
• เริ่ม consume message ได้ด้วย consumer.Run(ctx)

Streaming vs. Batch

• Streaming: ประมวลผลทีละ 1 message ทันทีเมื่อมาถึง เหมาะกับงาน throughput ไม่สูงมาก งานที่ต้องการประหยัดหน่วยความจำ หรือการรับประกันการประมวลผลที่เข้มงวด
• Batch: รวมตามจำนวนหรือช่วงเวลาแล้วค่อยประมวลผล เหมาะกับระบบ throughput สูง หรือช่วยลดภาระของ downstream

Sequential หรือ Async

• ค่าเริ่มต้นคือประมวลผลแบบลำดับ (Sequential) — ต้องประมวลผลอันหนึ่งเสร็จก่อนจึงจะอ่านข้อความถัดไป
• เมื่อใช้ xkafka.Concurrency(N) จะรองรับ โหมด asynchronous (Async) ที่ประมวลผล message (หรือ batch) พร้อมกันได้ N ชุด

การจัดการ offset

• พฤติกรรมพื้นฐานของ Kafka คือเลื่อน offset ทันทีที่ส่งมอบ message ทำให้มีโอกาสสูญเสีย message ได้เมื่อเกิดความขัดข้อง
• xkafka ตั้งค่า enable.auto.offset.store=false เพื่อให้บันทึก offset หลังจากประมวลผล message (หรือ batch) เสร็จเท่านั้น
• จึงสามารถรับประกันการประมวลผลผ่าน Kafka ได้โดยไม่ต้องจัดการสถานะ message ด้วย DB หรือคิวแยกต่างหาก

• การรับประกันแบบ At-Most-Once

  • โดยพื้นฐานแล้ว commit offset ในเบื้องหลังตาม enable.auto.commit=true ของ Kafka
  • ใช้ xkafka.ManualCommit(true) ร่วมกับการประมวลผลแบบลำดับ เพื่อ commit offset ก่อนอ่านแต่ละ message/batch และรับประกันแบบ At-most-once

• การรับประกันแบบ At-Least-Once

  • ใช้ xkafka.ManualCommit(true) ร่วมกับ concurrency (N>1) เพื่อ commit offset แบบ synchronous ตามลำดับ แม้จะประมวลผลแบบขนาน
  • จึงนำแพตเทิร์น At-least-once ไปใช้ได้ง่าย

การจัดการ error

• ระดับ Handler

  • ภายใน Handler สามารถจัดการ application error หรือส่งต่อไปยัง Dead Letter Queue ได้
  • ควบคุมได้อย่างชัดเจนด้วย msg.AckSuccess() เมื่อสำเร็จ, msg.AckSkip() เมื่อต้องการข้าม, msg.AckFail(err) เมื่อเกิดความล้มเหลว

• ระดับ Middleware

  • ภายใน middleware สามารถเขียนตรรกะร่วม เช่น retry หรือ error logging แล้วนำกลับมาใช้ซ้ำกับหลาย Handler ได้
  • ปรับนโยบาย retry หรือวิธีจัดการตามประเภทของ error ได้สะดวก

• ระดับ Global

  • error จาก Kafka broker/ไลบรารีจะถูกจัดการแบบศูนย์กลางผ่าน option บังคับ xkafka.ErrorHandler
  • หาก handler นี้คืนค่า error ที่ไม่เป็น nil จะหยุดการทำงานของ Consumer/Producer

บทสรุป

• xkafka เปลี่ยนประสบการณ์ใช้งาน Apache Kafka ที่ซับซ้อน ให้กลายเป็นโครงสร้างแบบ HTTP server ที่คุ้นเคยสำหรับนักพัฒนา Go
• ช่วยลด boilerplate ที่ไม่จำเป็น และมอบสภาพแวดล้อมที่ทำให้โฟกัสกับ business logic ได้เต็มที่
• เมื่อเทียบกับโค้ด confluent-kafka-go เดิมแล้ว มีความกระชับและเข้าใจง่ายกว่ามาก
• สามารถเริ่มต้นได้ทันทีจาก เอกสาร และ ตัวอย่าง อย่างเป็นทางการ

[penza1]

อืม ใน golang
เท่าที่รู้ เหมือนว่า sarama จะเป็นตัวที่คนชอบใช้มากกว่านะ..
ไคลเอนต์ Kafka นี่.. ซับซ้อนกว่าที่คิดมากเวลาเจอบรोकเกอร์ล่มหรือมีข้อยกเว้น
จะครอบคลุมทุกเคสได้ไหม ..