Kafka คืออะไร? พร้อมตัวอย่างการใช้งานด้วยภาษา Golang

Kafka เป็น Message Queue ที่ทำหน้าที่เป็นตัวกลางสำหรับการรับส่งข้อมูลจาก Producer ไปยัง Consumer หรือเปรียบเสมือนพนักงานส่งของที่รับข้อมูลจากหลายแหล่ง (Producer) รวบรวม จัดระเบียบ และส่งต่อข้อมูลไปยังผู้รับปลายทางที่ต้องการ (Consumer) โดยที่ Kafka มีจุดเด่นในการจัดการข้อมูลขนาดใหญ่ที่เข้ามาอย่างรวดเร็วและต่อเนื่องโดยทำงานในลักษณะของ Distributed System ที่ช่วยให้สามารถขยายระบบได้ตามต้องการ นอกจากนี้ Kafka ยังมีคุณสมบัติที่ช่วยให้ข้อมูลมีความคงทนและเชื่อถือได้ โดยการเก็บข้อมูลในรูปแบบของ Log ที่สามารถเรียกใช้ย้อนหลังได้ตามต้องการ

(ที่มา : Part 1: Apache Kafka for beginners – What is Apache Kafka? – CloudKarafka, Apache Kafka Message streaming as a Service)

องค์ประกอบหลักของ Kafka

  • Producer: เป็นผู้ที่ส่งข้อมูลเข้าสู่ Kafka อาจจะเป็นแอปพลิเคชันหรือบริการที่ผลิตข้อมูล
  • Consumer: เป็นผู้ที่รับข้อมูลจาก Kafka และนำข้อมูลไปใช้งานต่อ อาจจะเป็นแอปพลิเคชันหรือบริการที่ต้องการข้อมูลจาก Kafka
  • Broker: เป็นเซิร์ฟเวอร์ที่ทำหน้าที่จัดการและเก็บข้อมูลจาก Producer และส่งข้อมูลไปให้ Consumer
  • Topic: เป็นช่องทางหรือหมวดหมู่สำหรับการจัดเก็บและส่งข้อมูลใน Kafka ผู้ใช้งานสามารถกำหนด Topic ได้ตามต้องการ

การทำงานเริ่มต้นจากการที่ Producer ส่งข้อมูลไปยัง Topic ที่ได้กำหนดไว้ใน Kafka หลังจากนั้น ตัว Kafka Broker จะทำการเก็บข้อมูลที่ได้จาก Producer เป็นรูปแบบ Log ไฟล์ที่จัดเก็บอยู่ใน Topic นั้น ๆ หากมีการขอดึงข้อมูลจาก Consumer จาก Topic ที่ต้องการ Kafka Broker จะเป็นคนส่งข้อมูลให้ Consumer โดยสามารถรองรับการส่งข้อมูลแบบ Real-Time หรือแบบ Batch ก็ได้

การใช้งาน Kafka ในงานต่าง ๆ

  • เว็บไซต์ขายสินค้า: เมื่อลูกค้าสั่งซื้อสินค้า ข้อมูลการสั่งซื้อจะถูกส่งไปยัง Kafka ในฐานะ Producer จากนั้นระบบ Back-End ของเว็บไซต์จะทำหน้าที่เป็น Consumer ดึงข้อมูลการสั่งซื้อจาก Kafka ไปประมวลผลต่อ เช่น การจัดส่งสินค้า การอัปเดตสต็อก และการแจ้งเตือนการชำระเงิน
  • แพลตฟอร์มโซเชียลมีเดีย: โพสต์และข้อความต่าง ๆ บนโซเชียลมีเดียสามารถถูกส่งไปยัง Kafka ในฐานะ Producer จากนั้นระบบวิเคราะห์ข้อมูลจะทำหน้าที่เป็น Consumer ดึงข้อมูลจาก Kafka ไปวิเคราะห์เพื่อดู Sentiment ของผู้ใช้ และติดตามเทรนด์ต่าง ๆ

ข้อดีของการใช้ Kafka

  • Kafka ถูกออกแบบมาให้สามารถจัดการกับข้อมูลปริมาณมากและรวดเร็วได้
  • Kafka มีระบบการจัดการข้อมูลที่ทำให้มั่นใจได้ว่าข้อมูลจะไม่สูญหาย และสามารถส่งข้อมูลไปยัง Consumer ได้อย่างถูกต้อง
  • การใช้ Kafka ช่วยให้การสื่อสารระหว่างบริการเป็นแบบ Asynchronous ซึ่งทำให้ระบบไม่ต้องรอการตอบสนองทันที ลดปัญหาการรอคอยและเพิ่มประสิทธิภาพในการทำงานของระบบโดยรวม

ตัวอย่างการใช้ Kafka ในภาษา Golang

การวางโครงสร้างของ Project

เตรียมความพร้อม

ก่อนอื่นเลย Kafka จำเป็นต้องใช้ Zookeeper หรือ Distributed Service ตัวอื่น ๆ ร่วมด้วย เนื่องจาก Zookeeper ทำหน้าที่สำคัญในการจัดการและประสานงานระหว่าง Broker ต่าง ๆ ในคลัสเตอร์ของ Kafka โดยในการใช้งานสามารถสร้างไฟล์  docker-compose.yml  แล้วก็ run ผ่าน Docker Compose

สร้างไฟล์ docker-compose.yml

version: '3.9'
services:
  zookeeper:
    image: zookeeper
    container_name: zookeeper
    volumes:
      - .zookeeper:/data

  kafka:
    image: bitnami/kafka
    container_name: kafka
    ports:
      - 9092:9092
    volumes:
      - .kafka:/bitnami/kafka/data
    environment:
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
    depends_on:
      - zookeeper
การเริ่มต้น Docker

docker compose up
ติดตั้ง Lib ที่จำเป็นในทั้งส่วนของ Producer และ Consumer

go get github.com/Shopify/sarama

ตัวอย่างการส่งข้อความ (Producer)

สร้างไฟล์ main.go ใน Folder Producer

package main

import (
"log"
"github.com/Shopify/sarama"
)

func main() {
    // กำหนด Kafka Brokers
    brokers := []string{"localhost:9092"}
    
    // สร้าง Producer
    producer, err := sarama.NewSyncProducer(brokers, nil)
    if err != nil {
        log.Fatalf("Failed to start producer: %v", err)
    }
    defer producer.Close()
    
    // สร้างข้อความที่จะส่ง
    message := &sarama.ProducerMessage{
        Topic: "example-topic",
        Value: sarama.StringEncoder("Hello, Kafka!"),
    }
    
    // ส่งข้อความ
    partition, offset, err := producer.SendMessage(message)
    if err != nil {
        log.Fatalf("Failed to send message: %v", err)
    }
    
    log.Printf("Message sent to partition %d with offset %d\n", partition, offset)
}

ตัวอย่างการรับข้อความ (Consumer)

สร้างไฟล์ main.go ใน Folder Consumer

package main
import (
    "log"
    "github.com/Shopify/sarama"
    )

func main() {
    // กำหนด Kafka brokers
    brokers := []string{"localhost:9092"}
    
    // สร้าง Consumer
    consumer, err := sarama.NewConsumer(brokers, nil)
    if err != nil {
        log.Fatalf("Failed to start consumer: %v", err)
    }
    defer consumer.Close()
    
    // เลือกพาร์ติชั่นที่ต้องการอ่าน
    partitionConsumer, err := consumer.ConsumePartition("example-topic", 0, sarama.OffsetNewest)
    if err != nil {
        log.Fatalf("Failed to start partition consumer: %v", err)
    }
    defer partitionConsumer.Close()
    
    // อ่านข้อความจากพาร์ติชั่น
    for message := range partitionConsumer.Messages() {
        log.Printf("Message received: %s\n", string(message.Value))
    }
}

ทดสอบผลลัพธ์

Producer ส่งข้อความ “Hello, Kafka!” ไปยัง Kafka topic และข้อความนี้ถูกบันทึกในพาร์ติชั่นที่ 0 ของ topic นี้
Consumer ได้รับข้อความที่ถูกส่งมาและแสดงผลว่า “Hello, Kafka!”