Computer >> Hướng Dẫn Máy Tính >  >> Lập Trình >> Ruby

Khai thác Event Streaming trong Rails với Kafka

Các công ty muốn phản ứng nhanh chóng với nhu cầu xử lý và chia sẻ lượng lớn dữ liệu theo thời gian thực để có được thông tin chi tiết và tạo trải nghiệm khách hàng hấp dẫn hơn. Do đó, xử lý dữ liệu truyền thống không còn khả thi trong thế giới ngày nay.

Để đạt được điều đó, bạn cần xử lý nhiều dữ liệu càng nhanh càng tốt và gửi nó đến các dịch vụ khác để xử lý thêm. Nhưng giữa tất cả những hành động nhanh chóng này, cần phải thông báo cho consumer khi sự kiện xảy ra — và chúng ta có thể làm chính xác điều đó bằng event streaming.

Đây là repo trên GitHub mà chúng ta sẽ sử dụng.

Sự kiện

Trước khi nói về event streaming, hãy nói về event là gì. Một sự kiện xảy ra trong ứng dụng có thể liên quan đến quy trình người dùng hoặc đơn giản là các hành động ảnh hưởng đến kinh doanh.

Sự kiện đại diện cho một thay đổi trạng thái, không phải câu hỏi về cách sửa đổi ứng dụng. Xem xét các ví dụ sau:

  • Một người dùng đăng nhập vào dịch vụ
  • Một giao dịch thanh toán
  • Một tác giả xuất bản bài viết trong blog

Trong hầu hết các trường hợp, một sự kiện sẽ kích hoạt thêm sự kiện khác; ví dụ: khi người dùng đăng ký dịch vụ, ứng dụng gửi thông báo đến thiết bị của họ, chèn bản ghi vào database, và gửi email chào mừng.

Event streaming

Event Streaming là một pattern để thu thập dữ liệu theo thời gian thực từ các nguồn sự kiện như database. Các phần chính của event streaming bao gồm:

  • Broker: Hệ thống chịu trách nhiệm lưu trữ sự kiện
  • Topic: Một danh mục sự kiện
  • Producer: Gửi sự kiện đến broker trên một topic cụ thể
  • Consumer: Đọc các sự kiện
  • Events: Dữ liệu mà producer muốn truyền đạt cho consumer

Không thể tránh khỏi việc nói về publish and subscribe architecture pattern (pub/sub pattern) tại thời điểm này; event streaming là một triển khai của pattern đó nhưng với những thay đổi sau:

  • Sự kiện xảy ra thay vì message.
  • Sự kiện được sắp xếp, thường theo thời gian.
  • Consumer có thể đọc sự kiện từ một điểm cụ thể trong topic.
  • Sự kiện có độ bền theo thời gian.

Luồng bắt đầu khi producer xuất bản một sự kiện mới vào một topic (như chúng ta đã thấy, topic chỉ là phân loại cho một loại sự kiện cụ thể). Sau đó, consumer quan tâm đến sự kiện của danh mục cụ thể đăng ký topic đó. Cuối cùng, broker xác định consumer của topic và cung cấp các sự kiện mong muốn.

Lợi ích của event streaming

  • Giảm sự phụ thuộc (Decoupling) Không có sự phụ thuộc giữa publisher và consumer vì họ không cần biết nhau. Ngoài ra, sự kiện không chỉ định hành động của chúng, nên nhiều consumer có thể nhận cùng một sự kiện và thực hiện hành động khác nhau.

  • Độ trễ thấp (Low Latency) Sự kiện được tách biệt và cho phép consumer sử dụng chúng bất cứ lúc nào; có thể xảy ra trong mili-giây.

  • Độc lập (Independence) Như chúng ta biết, publisher và consumer độc lập, nên các team khác nhau có thể làm việc với chúng dùng cùng sự kiện cho hành động hoặc mục đích khác.

  • Chịu lỗi (Fault Tolerance) Một số platform event streaming giúp bạn xử lý lỗi consumer; ví dụ: consumer có thể lưu vị trí và bắt đầu lại từ đó nếu lỗi xảy ra.

  • Xử lý thời gian thực (Real-Time Handling) Phản hồi được nhận theo thời gian thực, nên người dùng không cần chờ phút hay giờ để thấy phản hồi của sự kiện.

  • Hiệu suất cao (High Performance) Các platform sự kiện có thể xử lý nhiều message nhờ độ trễ thấp — ví dụ: hàng nghìn sự kiện trong một giây.

Nhược điểm của event streaming

  • Giám sát (Monitoring) Một số công cụ event streaming không có công cụ giám sát hoàn chỉnh; chúng đòi hỏi công cụ bổ sung như Datadog hoặc New Relic.

  • Cấu hình (Configuration) Cấu hình trong một số công cụ có thể quá tải ngay cả với người có kinh nghiệm. Có nhiều tham số, và đôi khi bạn cần hiểu sâu về chủ đề để triển khai.

  • Thư viện client (Client libraries) Không dễ triển khai Kafka trong ngôn ngữ khác Java. Đôi khi thư viện client không cập nhật, không ổn định, hoặc không cung cấp nhiều lựa chọn.

Một trong những công cụ phổ biến nhất cho event streaming là Apache Kafka. Công cụ này cho phép người dùng gửi, lưu trữ và yêu cầu dữ liệu bất cứ khi và nơi họ cần; hãy nói về nó.

Apache Kafka

"Apache Kafka là một platform event streaming phân tán mã nguồn mở được hàng nghìn công ty sử dụng cho các pipeline dữ liệu hiệu suất cao, streaming analytics, tích hợp dữ liệu và ứng dụng quan trọng."

Được thiết kế đặc biệt cho truyền log thời gian thực, Apache Kafka lý tưởng cho các ứng dụng yêu cầu:

  • Trao đổi dữ liệu đáng tin cậy giữa các thành phần khác nhau
  • Khả năng chia nhỏ khối lượng messaging khi yêu cầu ứng dụng thay đổi
  • Truyền thời gian thực để xử lý dữ liệu

Hãy dùng Kafka trong ứng dụng Rails!

Sử dụng Kafka với Rails

Gem nổi tiếng nhất để dùng Kafka trong Ruby là ruby-kafka của Zendesk, và nó tuyệt vời! Tuy nhiên, bạn phải tự làm tất cả triển khai, đó là lý do chúng ta có một số "framework" được xây dựng trên ruby-kafka. Chúng cũng giúp chúng ta với tất cả các bước cấu hình và thực thi.

Karafka là một framework dùng để đơn giản hóa phát triển ứng dụng Ruby dựa trên Apache Kafka.

Để làm việc với Kafka, cần cài đặt Java. Vì Kafka là ứng dụng Scala và Java, cài đặt Zookeeper cũng sẽ được yêu cầu.

Trước khi cài đặt, tôi muốn giải thích một chút về Zookeeper. Zookeeper là dịch vụ tập trung thiết yếu cho Kafka; nó gửi thông báo trong trường hợp thay đổi như tạo topic mới, crash broker, xóa broker, xóa topic, v.v.

Nhiệm vụ chính của nó là quản lý Kafka broker, duy trì danh sách với metadata tương ứng, và tạo cơ chế health-check. Ngoài ra, nó giúp chọn broker leader cho các partition khác nhau của topic.

Yêu cầu

Cho MacOS:

Bây giờ, hãy cài đặt Java và Zookeeper với các lệnh sau:

brew install java
brew install zookeeper

Sau đó, tiếp tục cài đặt Kafka bằng lệnh:

brew install kafka

Khi đã cài Kafka và Zookeeper, cần khởi động dịch vụ như sau:

brew services start zookeeper
brew services start kafka

Cho Windows và Linux:

Hướng dẫn:

  1. Cài đặt Java
  2. Tải Zookeeper

Thiết lập Rails

Chỉ cần tạo ứng dụng Rails đơn giản như bình thường:

rails new karafka_example

và thêm gem karafka vào Gemfile:

gem 'karafka'

Sau đó, chạy bundle install để cài gem vừa thêm, và đừng quên chạy lệnh sau để lấy mọi thứ của Karafka:

bundle exec karafka install

Lệnh đó sẽ tạo một số file thú vị: đầu tiên là karafka.rb trong thư mục gốc, app/consumers/application_consumer.rb, và app/responders/application_responder.rb.

Khởi tạo Karafka

File karafka.rb giống như một initializer ứng dụng tách biệt khỏi config Rails. Nó cho phép bạn cấu hình ứng dụng Karafka và vẽ một số route, tương tự về API như route ứng dụng Rails. Nhưng ở đây, nó dành cho topic và consumer.

Producer

Producer chịu trách nhiệm tạo sự kiện, và chúng ta có thể thêm chúng vào thư mục app/responders. Bây giờ, hãy tạo một producer đơn giản cho users:

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

Consumer

Consumer chịu trách nhiệm đọc tất cả sự kiện/message gửi từ producer. Đây chỉ là consumer ghi log message nhận được.

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

Chúng ta dùng params để lấy sự kiện. Nhưng nếu bạn đọc sự kiện theo batch và config config.batch_fetching là true, bạn nên dùng params_batch.

Kiểm thử

Để chạy service Karafka (sẽ lắng nghe sự kiện), vào console, mở tab mới, vào project Rails, và chạy:

bundle exec karafka server

Sự kiện thành công

Mở tab console khác, vào project Rails, và gõ:

rails c

Đó, hãy tạo sự kiện với responder:

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

Nếu kiểm tra Rails console, chúng ta sẽ nhận được message sau sau khi sự kiện được tạo:

Successfully appended 1 messages to users/0 on 192.168.1.77:9092 (node_id=0)
=> {"users"=>[["{\"event_name\":\"user_created\",\"payload\":{\"user_id\":1}}", {:topic=>"users"}]]}

Và trong tab Karafka service, bạn sẽ thấy như sau:

New [User] event: #<Karafka::Params::Params:0x00007fa76f0316c8>
Inline processing of topic users with 1 messages took 0 ms
1 message on users topic delegated to UsersConsumer
[[karafka_example] {}:] Marking users/0:1 as processed
[[karafka_example] {}:] Committing offsets: users/0:2
[[karafka_example] {}:] [offset_commit] Sending offset_commit API request 28 to 192.168.1.77:9092

Nhưng nếu bạn chỉ muốn message payload, bạn có thể thêm params.payload trong consumer và bạn sẽ có như sau:

Params deserialization for users topic successful in 0 ms
New [User] event: {"event_name"=>"user_created", "payload"=>{"user_id"=>1}}
Inline processing of topic users with 1 messages took 1 ms
1 message on users topic delegated to UsersConsumer

Sự kiện thất bại

Bạn có thể tạo model User với các thuộc tính như email, first_name và last_name bằng lệnh:

rails g model User email first_name last_name

Sau đó, chạy migration:

rails db:migrate

Bây giờ, thêm validation như sau:

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

Cuối cùng, thay đổi consumer:

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

Vậy, hãy tạo hai sự kiện với cùng email:

UsersResponder.call({ event_name: "user_created", user: { user_id: 1, email: 'batman@mail.com', first_name: 'Bruce', last_name: 'Wayne' } } )
UsersResponder.call({ event_name: "user_created", user: { user_id: 2, email: 'batman@mail.com', first_name: 'Bruce', last_name: 'Wayne' } } )

Với này, sự kiện đầu tiên được tạo trong database:

New [User] event: {"event_name"=>"user_created", "user"=>{"user_id"=>1, "email"=>"batman@mail.com", "first_name"=>"Bruce", "last_name"=>"Wayne"}}
[[karafka_example] {users: 0}:] [fetch] Received response 2 from 192.168.1.77:9092
[[karafka_example] {users: 0}:] Fetching batches
[[karafka_example] {users: 0}:] [fetch] Sending fetch API request 3 to 192.168.1.77:9092
[[karafka_example] {users: 0}:] [fetch] Waiting for response 3 from 192.168.1.77:9092
TRANSACTION (0.1ms) BEGIN ↳ app/consumers/users_consumer.rb:14:in `consume'
User Create (9.6ms) INSERT INTO "users" ("user_id", "email", "first_name", "last_name", "created_at", "updated_at") VALUES ($1, $2, $3, $4, $5, $6) RETURNING "id" [["user_id", "1"], ["email", "batman@mail.com"], ["first_name", "Bruce"], ["last_name", "Wayne"], ["created_at", "2021-03-10 04:29:14.827778"], ["updated_at", "2021-03-10 04:29:14.827778"]] ↳ app/consumers/users_consumer.rb:14:in `consume'
TRANSACTION (5.0ms) COMMIT ↳ app/consumers/users_consumer.rb:14:in `consume'
Inline processing of topic users with 1 messages took 70 ms
1 message on users topic delegated to UsersConsumer

Nhưng sự kiện thứ hai sẽ thất bại, vì validation yêu cầu email unique. Nếu thử thêm bản ghi khác với email đã tồn tại, bạn sẽ thấy như sau:

New [User] event: {"event_name"=>"user_created", "user"=>{"user_id"=>2, "email"=>"batman@mail.com", "first_name"=>"Bruce", "last_name"=>"Wayne"}}
[[karafka_example] {users: 0}:] [fetch] Received response 2 from 192.168.1.77:9092
[[karafka_example] {users: 0}:] Fetching batches
[[karafka_example] {users: 0}:] [fetch] Sending fetch API request 3 to 192.168.1.77:9092
[[karafka_example] {users: 0}:] [fetch] Waiting for response 3 from 192.168.1.77:9092
TRANSACTION (0.2ms) BEGIN ↳ app/consumers/users_consumer.rb:14:in `consume'
User Exists? (0.3ms) SELECT 1 AS one FROM "users" WHERE "users"."email" = $1 LIMIT $2 [["email", "batman@mail.com"], ["LIMIT", 1]] ↳ app/consumers/users_consumer.rb:14:in `consume'
TRANSACTION (0.2ms) ROLLBACK ↳ app/consumers/users_consumer.rb:14:in `consume'
[[karafka_example] {users: 0}:] Exception raised when processing users/0 at offset 42 -- ActiveRecord::RecordInvalid: Validation failed: Email has already been taken

Bạn thấy lỗi ở dòng cuối ActiveRecord::RecordInvalid: Validation failed: Email has already been taken. Nhưng điều thú vị là Kafka sẽ thử xử lý sự kiện, lại và lại. Ngay cả nếu bạn restart Karafka server, nó sẽ cố xử lý sự kiện cuối cùng. Kafka biết bắt đầu từ đâu?

Nếu bạn xem console, sau lỗi, bạn sẽ thấy:

[[karafka_example] {users: 0}:] Exception raised when processing users/0 at offset 42

Nó sẽ nói offset nào được xử lý: trong trường hợp này là offset 42. Vì vậy, nếu restart service Karafka, nó sẽ bắt đầu ở offset đó.

[[karafka_example] {}:] Committing offsets with recommit: users/0:42
[[karafka_example] {users: 0}:] Fetching batches

Nó vẫn sẽ thất bại vì validation email trong model User. Lúc này, dừng Karafka server, xóa hoặc comment validation đó, và khởi động server lại; bạn sẽ thấy sự kiện được xử lý thành công:

[[karafka_example] {}:] Committing offsets with recommit: users/0:42
[[karafka_example] {}:] [offset_commit] Sending offset_commit API request 5 to 192.168.1.77:9092
[[karafka_example] {}:] [offset_commit] Waiting for response 5 from 192.168.1.77:9092
[[karafka_example] {}:] [offset_commit] Received response 5 from 192.168.1.77:9092
Params deserialization for users topic successful in 0 ms
New [User] event: {"event_name"=>"user_created", "user"=>{"user_id"=>2, "email"=>"batman@mail.com", "first_name"=>"Bruce", "last_name"=>"Wayne"}}
TRANSACTION (0.2ms) BEGIN ↳ app/consumers/users_consumer.rb:14:in `consume'
User Create (3.8ms) INSERT INTO "users" ("user_id", "email", "first_name", "last_name", "created_at", "updated_at") VALUES ($1, $2, $3, $4, $5, $6) RETURNING "id" [["user_id", "2"], ["email", "batman@mail.com"], ["first_name", "Bruce"], ["last_name", "Wayne"], ["created_at", "2021-03-10 04:49:37.832452"], ["updated_at", "2021-03-10 04:49:37.832452"]] ↳ app/consumers/users_consumer.rb:14:in `consume'
TRANSACTION (5.5ms) COMMIT ↳ app/consumers/users_consumer.rb:14:in `consume'
Inline processing of topic users with 1 messages took 69 ms
1 message on users topic delegated to UsersConsumer
[[karafka_example] {}:] Marking users/0:43 as processed

Cuối cùng, bạn thấy message dòng cuối: Marking users/0:43 as processed.

Callbacks

Đây là điều thú vị Karafka cung cấp: bạn có thể dùng callbacks trong Consumer. Để làm vậy, chỉ cần import module và dùng chúng. Mở UserConsumer và thêm:

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

Poll là trung gian lấy bản ghi dựa trên offset partition hiện tại. Vì vậy callbacks before_poll và after_poll, như tên gọi, được thực thi tại thời điểm đó. Chúng ta chỉ log message, và bạn thấy chúng trong Karafka server — một trước khi fetch và cái kia sau đó:

*** Checking something new for users
[[karafka_example] {}:] No batches to process
[[karafka_example] {users: 0}:] [fetch] Received response 325 from 192.168.1.77:9092
[[karafka_example] {users: 0}:] Fetching batches
[[karafka_example] {users: 0}:] [fetch] Sending fetch API request 326 to 192.168.1.77:9092
[[karafka_example] {users: 0}:] [fetch] Waiting for response 326 from 192.168.1.77:9092
*** We just checked for new messages!

Heartbeats

Heartbeat chỉ là cách chúng ta, với tư cách consumer, nói với Kafka rằng chúng ta còn sống; nếu không, Kafka sẽ cho rằng consumer đã chết.

Trong Karafka, có config mặc định để làm điều này theo một khoảng thời gian; đó là kafka.heartbeat_interval và mặc định là 10 giây. Bạn thấy heartbeat này trong Karafka server.

*** Checking something new for users
[[karafka_example_example] {}:] Sending heartbeat...
[[karafka_example_example] {}:] [heartbeat] Sending heartbeat API request 72 to 192.168.1.77:9092
[[karafka_example_example] {}:] [heartbeat] Waiting for response 72 from 192.168.1.77:9092
[[karafka_example_example] {}:] [heartbeat] Received response 72 from 192.168.1.77:9092
*** We just checked for new messages!

Với Sending heartbeat..., Kafka biết chúng ta còn sống và là thành viên hợp lệ của consumer group. Cũng như vậy, chúng ta có thể consume thêm bản ghi.

Commit

Đánh dấu một offset đã được consume gọi là commit offset. Trong Kafka, chúng ta ghi nhận offset commit bằng cách ghi vào topic Kafka nội bộ gọi là offsets topic. Một message được coi là đã consume chỉ khi offset của nó được commit vào offsets topic.

Karafka có config để thực hiện commit này tự động mỗi lần; config là kafka.offset_commit_interval, giá trị mặc định 10 giây. Với này, Karafka sẽ commit offset mỗi 10 giây, và bạn thấy message đó trong Karafka server:

*** Checking something new for users
[[karafka_example] {}:] No batches to process
[[karafka_example] {users: 0}:] [fetch] Received response 307 from 192.168.1.77:9092
[[karafka_example] {users: 0}:] Fetching batches
[[karafka_example] {users: 0}:] [fetch] Sending fetch API request 308 to 192.168.1.77:9092
[[karafka_example] {users: 0}:] [fetch] Waiting for response 308 from 192.168.1.77:9092
[[karafka_example] {}:] Committing offsets: users/0:44
[[karafka_example] {}:] [offset_commit] Sending offset_commit API request 69 to 192.168.1.77:9092
[[karafka_example] {}:] [offset_commit] Waiting for response 69 from 192.168.1.77:9092
[[karafka_example] {}:] [offset_commit] Received response 69 from 192.168.1.77:9092
*** We just checked for new messages!

Committing offsets: users/0:44 cho biết offset nào đang commit; trong trường hợp của tôi, nó báo cho Kafka rằng có thể commit offset số 44 từ topic 0. Theo cách đó, nếu có sự cố với service, Karafka có thể bắt đầu lại xử lý sự kiện từ offset đó.

Kết luận

Event streaming giúp chúng ta nhanh hơn, tận dụng dữ liệu tốt hơn, và thiết kế trải nghiệm người dùng tốt hơn. Thực tế, nhiều công ty dùng pattern này để giao tiếp giữa tất cả service và phản ứng với các sự kiện khác nhau theo thời gian thực. Như đã đề cập, có các lựa chọn khác ngoài Karafka bạn có thể dùng với Rails. Bạn đã có cơ bản; hãy tự do thử nghiệm với chúng.

Tham khảo

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