home

หน้าแรก
 Computer >> บทช่วยสอนคอมพิวเตอร์ >  >> การเขียนโปรแกรม >> Ruby

การใช้ Event Streaming ใน Rails ด้วย Kafka

บริษัทต้องการ React เร็วต่อความต้องการในการ Process และ Share ข้อมูลปริมาณมาก Real-time เพื่อ Insights และ Customer Experiences ที่ดียิ่งขึ้น Traditional Data Processing จึงไม่ Viable อีกต่อไป

ต้อง Process Data เร็วที่สุดแล้วส่งต่อ Services อื่นๆ แต่ระหว่างกระบวนการเร็วๆ นี้ จำเป็นต้อง Notify Consumers เมื่อ Event เกิดขึ้น — และเราทำได้ด้วย Event Streaming

นี่คือ GitHub Repo ที่เราจะใช้

Events (เหตุการณ์)

ก่อนพูดถึง Event Streaming มาทำความเข้าใจ Event ก่อน Event ใน Application อาจเกี่ยวกับ User Process หรือ Actions ที่ส่งผลต่อ Business

Events แทน State Change ไม่ใช่ "วิธี Modify Application" ตัวอย่าง:

  • User Login
  • Payment Transaction
  • Writer Publish Blog Post

โดยทั่วไป Event จะ Trigger Events อื่นๆ เช่น User Sign Up -> App ส่ง Notification, Insert DB Record, ส่ง Welcome Email

Event Streaming (การสตรีมเหตุการณ์)

Event Streaming เป็น Pattern สำหรับ Capture Data Real-time จาก Event Sources เช่น Databases ส่วนประกอบหลัก:

  • Broker: System ที่เก็บ Events
  • Topic: Category ของ Events
  • Producer: ส่ง Events ไป Broker บน Topic เฉพาะ
  • Consumer: อ่าน Events
  • Events: Data ที่ Producers ต้องการ Communicate ให้ Consumers

ต้องพูดถึง Publish/Subscribe Pattern (Pub/Sub) Event Streaming เป็น Implementation ของ Pattern นี้แต่มีความแตกต่าง:

  • มี Events แทน Messages
  • Events Ordered โดยทั่วไปตามเวลา
  • Consumers อ่าน Events จากจุดเฉพาะใน Topic ได้
  • Events มี Temporal Durability

Flow: Producer Publish Event -> Topic -> Consumers Subscribe Topic -> Broker Identify Consumers และทำ Events ให้ Available

Advantages (ข้อดี)

  • Decoupling: ไม่มี Dependency ระหว่าง Publishers/Consumers เพราะไม่ต้องรู้จักกัน Events ไม่ Specify Actions ดังนั้นหลาย Consumers รับ Event เดียวกันทำ Actions ต่างกันได้
  • Low Latency: Events Decoupled ให้ Consumer ใช้ได้ทุกเมื่อ ใน Milliseconds
  • Independence: Publishers/Consumers Independent ทีมต่างๆ ทำงานพร้อมกัน用 Same Events สำหรับ Purposes ต่างกัน
  • Fault Tolerance: Platforms บางตัวจัดการ Consumer Failures เช่น Save Position แล้ว Resume จากนั้นได้
  • Real-Time Handling: Feedback Real-time User ไม่ต้องรอ นาที/ชั่วโมง
  • High Performance: Platforms Handle Messages มากเพราะ Low Latency เช่น พัน Events/Second

Disadvantages (ข้อเสีย)

  • Monitoring: Tools บางตัวไม่มี Monitoring สมบูรณ์ ต้อง Tool เพิ่ม เช่น Datadog, New Relic
  • Configuration: Config อาจ Overwhelming แม้คนมี Experience Parameters เยอะ ต้องรู้ลึกถึงหัวข้อ
  • Client Libraries: Implement Kafka ในภาษา Non-Java ยาก Libraries บ้าง Outdated, Unstable, หรือ Alternatives น้อย

Tool ยอดนิยมที่สุดคือ Apache Kafka ส่ง เก็บ Request Data ได้ทุกเมื่อ ทุกที่

Apache Kafka

"Apache Kafka เป็น Open-source Distributed Event Streaming Platform ใช้โดยบริษัทพันรายสำหรับ High-performance Data Pipelines, Streaming Analytics, Data Integration, Mission-critical Applications."

ออกแบบมาเฉพาะ Real-time Log Transmission Kafka เหมาะสำหรับ Apps ที่ต้องการ:

  • Reliable Data Exchange ระหว่าง Components
  • Divide Messaging Workloads ตาม Requirements ที่เปลี่ยน
  • Real-time Transmission สำหรับ Data Processing

Using Kafka with Rails (ใช้ Kafka กับ Rails)

Gem ยอดนิยมคือ ruby-kafka โดย Zendesk ดีมาก แต่ต้อง Implement เองทั้งหมด จึงมี "Frameworks" สร้างบน ruby-kafka ช่วย Config และ Execution

Karafka = Framework ลดความซับซ้อน Apache Kafka-based Ruby Apps Development

ต้อง Install Java (Kafka เขียน Scala/Java) และ Zookeeper

Zookeeper = Centralized Service สำคัญสำหรับ Kafka ส่ง Notifications เมื่อ Changes (New Topic, Broker Crash, Remove Broker, Delete Topics, etc.) จัดการ Kafka Brokers, Maintain Metadata List, Facilitate Health-checking, ช่วย Select Leading Broker สำหรับ Partitions

Requirements (ข้อกำหนด)

For MacOS:

brew install java
brew install zookeeper
brew install kafka
brew services start zookeeper
brew services start kafka

For Windows and Linux:

  1. Installing Java
  2. Download Zookeeper

Setting Up Rails

rails new karafka_example
# Gemfile
gem 'karafka'
bundle install
bundle exec karafka install

คำสั่งสุดท้าย Generate Files: karafka.rb (Root), app/consumers/application_consumer.rb, app/responders/application_responder.rb

Karafka Initializer

karafka.rb เหมือน Initializer แยกจาก Rails Config Configure Karafka App และ Draw Routes (คล้าย Rails Routes แต่สำหรับ Topics และ Consumers)

Producer (ผู้ผลิต)

Producer สร้าง Events วางใน app/responders สร้าง Producer สำหรับ Users:

# app/responders/users_responder.rb
class UsersResponder < ApplicationResponder
  topic :users
  def respond(event_payload)
    respond_to :users, event_payload
  end
end

Consumer (ผู้บริโภค)

Consumer อ่าน Events จาก Producer ตัวอย่าง Log Received Message:

# app/consumers/users_consumer.rb
class UsersConsumer < ApplicationConsumer
  def consume
    Karafka.logger.info "New [User] event: #{params}"
  end
end

ใช้ params ได้ Event หาก Batch Fetching (config.batch_fetching = true) ใช้ params_batch

Testing (การทดสอบ)

รัน Karafka Service (Listener):

bundle exec karafka server

Successful Event (เหตุการณ์สำเร็จ)

Console ใหม่ -> Rails Project -> rails c สร้าง Event:

> UsersResponder.call({ event_name: "user_created", payload: { user_id: 1 } })

Rails Console จะได้ Message Success และ Karafka Tab จะเห็น Log Processed

ถ้าอยากแค่ Payload ใช้ params.payload ใน Consumer

Failed Event (เหตุการณ์ล้มเหลว)

สร้าง User Model:

rails g model User email first_name last_name
rails db:migrate

Add Validation:

class User < ApplicationRecord
  validates :email, uniqueness: true
end

เปลี่ยน Consumer:

class UsersConsumer < ApplicationConsumer
  def consume
    Karafka.logger.info "New [User] event: #{params.payload}"
    User.create!(params.payload['user'])
  end
end

สร้าง 2 Events Email เดียวกัน Event แรกสำเร็จ Event ที่ 2 Fail (Validation Unique) Kafka จะ Retry ซ้ำๆ แม้ Restart Server ก็ยัง Retry จาก Offset เดิม

Offset บอกว่า Process ที่ไหน (เช่น Offset 42) Restart แล้วเริ่มจาก Offset นั้น ต้อง Stop Server, Remove Validation, Start ใหม่ จะ Process สำเร็จและเห็น Marking users/0:43 as processed

Callbacks

Karafka มี Callbacks ใน Consumer ใช้ได้โดย Include Module:

class UsersConsumer < ApplicationConsumer
  include Karafka::Consumers::Callbacks
  before_poll do
    Karafka.logger.info "*** Checking something new for #{topic.name}"
  end
  after_poll do
    Karafka.logger.info '*** We just checked for new messages!'
  end
  def consume
    Karafka.logger.info "New [User] event: #{params.payload}"
    User.create!(params.payload['user'])
  end
end

before_poll/after_poll รันก่อน/หลัง Fetch Records ตาม Partition Offset

Heartbeats

Heartbeat = Consumer บอก Kafka ว่าตัวเอง Alive ไม่งั้น Kafka ถือว่า Dead Default Config kafka.heartbeat_interval = 10 วินาที เห็นใน Karafka Server Logs

Commit (การ Commit Offset)

Mark Offset As Consumed = Commit Offset Kafka บันทึกใน Internal Topic ชื่อ Offsets Topic Message ถือว่า Consumed เมื่อ Offset Committed แล้ว

Karafka Config kafka.offset_commit_interval = 10 วินาที Default Auto-commit ทุก 10 วิ Logs จะแสดง Committing offsets: users/0:44 บอกว่า Commit Offset 44 ของ Topic 0 หาก Service Crash Karafka Resume จาก Offset นั้น

Conclusion (สรุป)

Event Streaming ช่วยทำให้เร็วขึ้น ใช้ Data ดีกว่า Design UX ดีกว่า บริษัทหลายรายใช้ Pattern นี้ Communicate Services และ React Events Real-time มี Alternatives อื่นๆ สำหรับ Rails ตอนนี้มี Basics แล้ว ลอง Experiment ได้เลย

References

  • https://kafka.apache.org/
  • https://github.com/karafka/karafka
  • https://en.wikipedia.org/wiki/Publish%E2%80%93subscribe_pattern