রেইলসে কাফকা (Kafka) ব্যবহার করে ইভেন্ট স্ট্রিমিং হারনেস করা

কোম্পানিগুলো চায় রিয়েল-টাইমে বিশাল পরিমাণ ডেটা প্রসেস ও শেয়ার করে ইনসাইট পাওয়া এবং আরও আকর্ষণীয় কাস্টমার এক্সপেরিয়েন্স তৈরি করা। তাই ট্রাডিশনাল ডেটা প্রসেসিং আজকের বিশ্বে আর কাজে লাগে না।

এটি অর্জন করতে, আপনাকে খুব দ্রুত প্রচুর ডেটা প্রসেস করতে হবে এবং তারপর অন্য সার্ভিসে পাঠাতে হবে আরও প্রসেসিং-এর জন্য। কিন্তু এই সব দ্রুত کارর্যের মাঝপথে, কনসিউমারদের নোটিফাই করা জরুরি যখন ইভেন্টটি ঘটে—এবং আমরা ইভেন্ট স্ট্রিমিং ব্যবহার করে ঠিক এটি করতে পারি।

এটি গিটহাব রিপো যা আমরা ব্যবহার করব।

ইভেন্ট সমূহ

ইভেন্ট স্ট্রিমিং সম্পর্কে বলার আগে, চলো জানি ইভেন্ট কী। একটি অ্যাপ্লিকেশনের ভিতরে একটি ইভেন্ট ব্যবহারকারী প্রসেস বা কেবল বিজনেসকে প্রভাবিত করার অ্যাকশন থেকে সংক্রান্ত হতে পারে।

ইভেন্টগুলো একটি স্টেট চেঞ্জকে প্রতীকী করে, অ্যাপ্লিকেশন কীভাবে মডিফাই করবেন সেটা নয়। উদাহরণস্বরূপ:

• একজন ব্যবহারকারী সার্ভিসে লগইন করল
• একটি পেমেন্ট ট্রানজ্যাকশন
• একজন লেখক ব্লগে একটি পোস্ট পাবলিশ করল

বেশিরভাগ কেসে, একটি ইভেন্ট আরও ইভেন্ট ট্রিগার করে; উদাহরণস্বরূপ, যখন একজন ব্যবহারকারী সার্ভিসে সাইন আপ করে, অ্যাপ তাদের ডিভাইসে নোটিফিকেশন পাঠায়, ডেটাবেসে রেকর্ড ইনসার্ট করে, এবং একটি স্বাগত ইমেইল পাঠায়।

ইভেন্ট স্ট্রিমিং

ইভেন্ট স্ট্রিমিং হলো ইভেন্ট সোর্স (যেমন ডেটাবেস) থেকে রিয়েল-টাইমে ডেটা ক্যাপচার করার একটি প্যাটার্ন। ইভেন্ট স্ট্রিমিং-এর মূল অংশগুলো:

• ব্রোকার (Broker): ইভেন্ট সংরক্ষণ করার দায়িত্বপূর্ণ সিস্টেম
• টপিক (Topic): ইভেন্টের একটি ক্যাটাগরি
• প্রোডিউসার (Producer): একটি নির্দিষ্ট টপিকে ব্রোকারে ইভেন্ট পাঠায়
• কনসিউমার (Consumer): ইভেন্টগুলো পড়ে
• ইভেন্ট (Events): ডেটা যা প্রোডিউসার কনসিউমারদের যোগাযোগ করতে চায়

এই পয়েন্টে পাবলিশ এবং সাবস্ক্রাইব আর্কিটেকচার প্যাটার্ন (pub/sub pattern) নিয়ে কথা বলা অপরিহার্য; ইভেন্ট স্ট্রিমিং সেই প্যাটার্নের একটি ইম্প্লিমেন্টেশন কিন্তু এই পরিবর্তনներով:

• মেসেজের বদলে ইভেন্ট ঘটে।
• ইভেন্টগুলো অর্ডারড থাকে, সাধারণত সময় অনুযায়ী।
• কনসিউমাররা টপিকের একটি নির্দিষ্ট পয়েন্ট থেকে ইভেন্ট পড়তে পারে।
• ইভেন্টগুলোর টেম্পোরাল ড্যুরাবিলিটি (temporal durability) থাকে।

ফ্লো শুরু হয় যখন প্রোডিউসার একটি নতুন ইভেন্ট পাবলিশ করে একটি টপিকে (যেমন আমরা আগে দেখেছি, টপিক হলো একটি নির্দিষ্ট টাইপের ইভেন্টের ক্যাটাগরাইজেশন)। তারপর, সেই ক্যাটাগরির ইভেন্টে আগ্রহী কনসিউমাররা সে টপিক সাবস্ক্রাইব করে। অবশেষে, ব্রোকার টপিকের কনসিউমারদের শনাক্ত করে এবং কাঙ্খিত ইভেন্টগুলো উপলব্ধ করায়।

ইভেন্ট স্ট্রিমিং-এর সুবিধা

• ডিকাপলিং (Decoupling) পাবলিশার এবং কনসিউমারের মধ্যে কোনো ডিপেন্ডেন্সি নেই কারণ তারা একে অপরকে জানতে হয় না। এর পরাশ্রয়, ইভেন্টগুলো তাদের অ্যাকশন স্পেসিফাই করে না, তাই অনেক কনসিউমার একই ইভেন্ট পেয়ে ভিন্ন অ্যাকশন পারফর্ম করতে পারে।

• লো ল্যাটেন্সি (Low Latency) ইভেন্টগুলো ডিকাপলড থাকে এবং কনসিউমার যেকোনো সময় সেগুলো ইউটিলাইজ করতে পারে; এটি মিলিসেকেন্ডের মধ্যে হয়ে যেতে পারে।

• স্বাধীনতা (Independence) আমরা জানি, পাবলিশার এবং কনসিউমার স্বাধীন, তাই ভিন্ন টিমরা একই ইভেন্ট ব্যবহার করে ভিন্ন অ্যাকশন বা উদ্দেশ্যে কাজ করতে পারে।

• ফল্ট টলারেন্স (Fault Tolerance) কিছু ইভেন্ট স্ট্রিমিং প্ল্যাটফর্ম কনসিউমার ফেইলিউর হ্যান্ডেল করতে সাহায্য করে; উদাহরণস্বরূপ, কনসিউমার তাদের পজিশন সেভ করতে পারে এবং এরর হলে সেখান থেকে আবার শুরু করতে পারে।

• রিয়েল-টাইম হ্যান্ডলিং (Real-Time Handling) ফিডব্যাক রিয়েল-টাইমে পাওয়া যায়, তাই ব্যবহারকারীদের মিনিট বা ঘন্টা অপেক্ষা করতে হয় না তাদের ইভেন্টের রেসপন্স देखने के लिए।

• হাই পারফরম্যান্স (High Performance) ইভেন্ট প্ল্যাটফর্মগুলো অনেক মেসেজ হ্যান্ডেল করতে পারে লো ল্যাটেন্সির কারণে—উদাহরণস্বরূপ, প্রতি সেকেন্ডে হাজার হাজার ইভেন্ট।

ইভেন্ট স্ট্রিমিং-এর অসুবিধা

• মনিটরিং (Monitoring) কিছু ইভেন্ট স্ট্রিমিং টুলে কমপ্লিট মনিটরিং টুল নেই; তারা ড্যাটাডগ (Datadog) বা নিউ রেলিক (New Relic) এর মতো অডিশনাল টুল ইম্প্লিমেন্ট করার ডিমান্ড করে।

• কনফিগারেশন (Configuration) কিছু টুলের কনফিগারেশন অভিজ্ঞ লোকজনো জন্যও ওভারওয়েল্মিং হতে পারে। অনেক প্যারামিটার রয়েছে, এবং কখনো কখনো আপনাকে গভীরভাবে বিষয়টি জানতে হয় ইম্প্লিমেন্ট করার জন্য।

• ক্লায়েন্ট লাইব্রেরি (Client libraries) জাভা ছাড়া অন্যান্য ল্যাঙ্গুয়েজে কাফকা ইম্প্লিমেন্ট করা সহজ নয়। কখনো ক্লায়েন্ট লাইব্রেরি আপ-টু-ডেট নয়, অস্থিরতা দেখায়, বা অনেক অপশন দেয় না।

ইভেন্ট স্ট্রিমিং-এর সবচেয়ে জনপ্রিয় টুলগুলির একটি হলো Apache Kafka। এই টুল ব্যবহারকারীদের ডেটা পাঠানোর, সংরক্ষণের, এবং অনুরোধ করার সুবিধা দেয় যখন এবং যেখানে তাদের দরকার; চলুন এই সম্পর্কে কথা বলি।

Apache Kafka

"Apache Kafka হলো একটি ওপেন-सোর্স ডিস্ট্রিবিউটেড ইভেন্ট স্ট্রিমিং প্ল্যাটফর্ম যা হাজার হাজার কোম্পানি ব্যবহার করে হাই-পারফরম্যান্স ডেটা পাইপলাইন, স্ট্রিমিং অ্যানালিটিক্স, ডেটা ইন্টিগ্রেশন, এবং মিশন-ক্রিটিক্যাল অ্যাপ্লিকেশনের জন্য।"

বিশেষভাবে রিয়েল-টাইম লগ ট্রান্সমিশনের জন্য ডিজাইন করা হলে, Apache Kafka অ্যাপ্লিকেশনের জন্য আদর্শ যেগুলো আপেক্ষা করে:

• বিভিন্ন কম্পোনেন্টের মধ্যে विश्वস্ত ডেটা এক্সচেঞ্জ
• অ্যাপ্লিকেশন রিকোয়ারমেন্ট পরিবর্তন অনুযায়ী মেসেজিং ওয়ার্কলোড ভাগ করার ক্ষমতা
• ডেটা প্রসেসিং-এর জন্য রিয়েল-টাইম ট্রান্সমিশন

চলুন রেইলস অ্যাপ্লিকেশনে কাফকা ব্যবহার করি!

রেইলসে কাফকা ব্যবহার

রুবিতে কাফকা ব্যবহার করার সবচেয়ে বিখ্যাত জেম হলো Zendesk-এর ruby-kafka, এবং এটি দুর্দান্ত! তবুও, আপনাকে সকল ইম্প্লিমেন্টেশন ম্যানুয়ালি করতে হয়, তাই আমাদের কিছু "ফ্রেমওয়ার্ক" রয়েছে যা ruby-kafka দিয়ে তৈরি। তারা কনফিগারেশন এবং এক্সিকিউশন স্টেপসেও সাহায্য করে।

Karafka হলো একটি ফ্রেমওয়ার্ক যা Apache Kafka-ভিত্তিক রুবি অ্যাপ্লিকেশন ডেভেলপমেন্ট সিম্পলিফাই করে।

কাফকার সাথে কাজ করার জন্য জাভা ইনস্টল করা জরুরি। কারণ কাফকা একটি স্কালা (Scala) এবং জাভা অ্যাপ্লিকেশন, তাই Zookeeper ইনস্টল করাও আবশ্যক।

ইনস্টলেশনের আগে, Zookeeper সম্পর্কে একটু জানি। Zookeeper একটি সেন্ট্রালাইজড সার্ভিস যা কাফকার জন্য অপরিহার্য; এটি নোটিফিকেশন পাঠায় পরিবর্তন হলে—যেমন নতুন টপিক তৈরি, ব্রোকার ক্র্যাশ, ব্রোকার রিমুভ, টপিক ডিলিশন, ইত্যাদি।

এর মূল কাজ হলো কাফকা ব্রোকার ম্যানেজ করা, তাদের মেটাডেটার সাথে একটি লিস্ট মেইনটেইন করা, এবং হেলথ-চেকিং মেকানিজম ফ্যাসিলিটেট করা। এর পাশাপাশি, এটি টপিকের ভিন্ন পার্টিশনের জন্য লিডিং ব্রোকার সিলেক্ট করতে সাহায্য করে।

প্রয়োজনীয়তা

macOS-এর জন্য:

এখন, জাভা এবং Zookeeper ইনস্টল করুন নিচের কমান্ডগুলো দিয়ে:

brew install java
brew install zookeeper

তারপর, কাফকা ইনস্টল করুন:

brew install kafka

একবার কাফকা এবং Zookeeper ইনস্টল হলে, সার্ভিসগুলো শুরু করতে হবে এইভাবে:

brew services start zookeeper
brew services start kafka

Windows এবং Linux-এর জন্য:

निর্দেশনা:

1. জাভা ইনস্টল করা
2. Zookeeper ডাউনলোড করা

রেইলস সেটআপ

সামান্যভাবে একটি সিম্পল রেইলস অ্যাপ্লিকেশন তৈরি করুন:

rails new karafka_example

এবং Gemfile-এ karafka জেম যোগ করুন:

gem 'karafka'

তারপর bundle install চালান জেমটি ইনস্টল করার জন্য, এবং ভুলবেন না নিচের কমান্ড চালাতে যাতে সকল Karafka জিনিসপত্র পাওয়া যায়:

bundle exec karafka install

এই কমান্ড কিছু আকর্ষণীয় ফাইল জেনারেট করবে: প্রথমটি হলো রুট ডিরেক্টরিতে karafka.rb, app/consumers/application_consumer.rb, এবং app/responders/application_responder.rb।

Karafka Initializer

karafka.rb ফাইলটি রেইলস কনফিগ থেকে আলাদা একটি ইনিশিয়ালাইজার অ্যাপ্লিকেশনের মতো। এটি আপনাকে Karafka অ্যাপ্লিকেশন কনফিগার করতে দেয় এবং কিছু রুট ড্র করার অনুমতি দেয়, API-তে রেইলস অ্যাপ্লিকেশন রুটের মতো। কিন্তু এখানে এটি টপিক এবং কনসিউমারের জন্য।

প্রোডিউসার

প্রোডিউসার ইভেন্ট তৈরি করার দায়িত্বপূর্ণ, এবং আমরা এগুলো app/responders ফোল্ডারে যোগ করতে পারি। এখন, ইউজারদের জন্য একটি সিম্পল প্রোডিউসার বানাই:

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

কনসিউমার

কনসিউমার প্রোডিউসার থেকে পাঠানো সকল ইভেন্ট/মেসেজ পড়ার দায়িত্বপূর্ণ। এটি শুধু একটি কনসিউমার যা রিসিভ করা মেসেজ লগ করে।

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

ইভেন্ট পাওয়ার জন্য params ব্যবহার করি। কিন্তু আপনি যদি ব্যাচে (batches) ইভেন্ট পড়েন এবং কনফিগ config.batch_fetching true থাকে, তবে params_batch ব্যবহার করুন।

টেস্টিং

আমাদের Karafka সার্ভিস চালানোর জন্য (যেটি ইভেন্ট শোনবে), কনসোলে যান, নতুন ট্যাব খুলুন, রেইলস প্রজেক্টে যান, এবং চালান:

bundle exec karafka server

সফল ইভেন্ট

এখন আরেকটি কনসোল ট্যাব খুলুন, রেইলস প্রজেক্টে যান, এবং টাইপ করুন:

rails c

এখানে, আমাদের রেসপন্ডার দিয়ে একটি ইভেন্ট তৈরি করি:

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

রেইলস কনসোল চেক করলে, ইভেন্ট তৈরি হওয়ার পর এই মেসেজ পাওয়া যাবে:

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"}]]} 

এবং Karafka সার্ভিস ট্যাবে আপনি এমন কিছু দেখবেন:

New [User] event: #
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

যদি আপনি শুধু মেসেজ পেলোড চান, আপনার কনসিউমারে params.payload যোগ করুন এবং আপনি এমন কিছু পাবেন:

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

ব্যর্থ ইভেন্ট

আপনি একটি User মডেল তৈরি করতে পারেন কিছু অ্যাট্রিবিউট যেমন email, first_name এবং last_name নিচের কমান্ড দিয়ে:

rails g model User email first_name last_name

তারপর মাইগ্রেশন চালান:

rails db:migrate

এখন কিছু ভ্যালিডেশন যোগ করুন:

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

অবশেষে, কনসিউমার পরিবর্তন করুন:

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

তাহলে, একই ইমেইল দিয়ে দুটি ইভেন্ট তৈরি করি:

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' } } )

এটি হলে, প্রথম ইভেন্ট ডেটাবেসে তৈরি হবে:

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

কিন্তু দ্বিতীয়টি ফেল করবে, কারণ আমাদের একটি ভ্যালিডেশন আছে যে ইমেইল ইউনিক হতে হবে। যদি আপনি বিদ্যমান ইমেইল দিয়ে আরেকটি রেকর্ড যোগ করার চেষ্টা করেন, আপনি এমন কিছু দেখবেন:

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

আপনি লাস্ট লাইনে এরর দেখতে পাচ্ছেন: ActiveRecord::RecordInvalid: Validation failed: Email has already been taken। কিন্তু মজার কথা হলো কাফকা বারবার ইভেন্টটি প্রসেস করার চেষ্টা করবে। এমনকি আপনি Karafka সার্ভার রিস্টার্ট করলেও, এটি লাস্ট ইভেন্ট প্রসেস করার চেষ্টা করবে। কাফকা কীভাবে জানে কোথা থেকে শুরু করতে হবে?

আপনার কনসোল দেখলে, এররের পর আপনি এটি দেখবেন:

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

এটি আপনাকে বলে কোন অফসেট প্রসেস করা হয়েছিল: এই কেসে, এটি ছিল অফসেট ৪২। তাই, আপনি যদি Karafka সার্ভিস রিস্টার্ট করেন, এটি সেই অফসেট থেকে শুরু করবে।

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

এটি এখনও ফেল করবে কারণ আমাদের User মডেলে ইমেইল ভ্যালিডেশন আছে। এই পয়েন্টে, Karafka সার্ভার বন্ধ করুন, সেই ভ্যালিডেশন রিমুভ বা কমেন্ট আউট করুন, এবং আবার সার্ভার শুরু করুন; আপনি দেখবেন ইভেন্টটি সফলভাবে প্রসেস হয়েছে:

[[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

শেষে, আপনি লাস্ট লাইনে এই মেসেজ দেখবেন: Marking users/0:43 as processed।

Callbacks

Karafka যা অফার করে সেটি কয়েক্টা: আপনি আপনার Consumer-এ কলব্যাক ব্যবহার করতে পারেন। এটি করতে, আপনাকে শুধু মডিউল ইম্পোর্ট করতে হবে এবং সেগুলো ব্যবহার করতে হবে। তারপর, আপনার UserConsumer খুলুন এবং এটি যোগ করুন:

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 হলো मीডিয়াম जिसके মাধ্যমে हम करंट पार्टिशन ऑफसेट के आधार पर रिकॉर्ड्स लाते हैं। तो वे callbacks before_poll এবং after_poll, उनके नाम के अनुसार, उस क्षण निष्पादित होते हैं। हम सिर्फ एक मैसेज लॉग कर रहे हैं, और आप उन्हें अपने Karafka सर्वर में देख सकते हैं—एक fetching से पहले और दूसरा उसके बाद:

*** 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

हार्टबीट सिर्फ वह तरीका है जिससे हम, उपभोक्ताओं के रूप में, काफ्का को कहते हैं कि हम जीवित हैं; अन्यथा, काफ्का मान लेगा कि उपभोक्ता मर चुका है।

Karafka में, हमारे पास एक डिफ़ॉल्ट कॉन्फिग है जो इसे एक अवधि में करता है; यह kafka.heartbeat_interval है और डिफ़ॉल्ट 10 सेकंड है। आप इसे अपने Karafka सर्वर में देख सकते हैं।

*** 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!

Sending heartbeat... के साथ, काफ्का जानता है कि हम जीवित हैं और हम उसके कंज्यूमर ग्रुप के एक वैध सदस्य हैं। साथ ही, हम अधिक रिकॉर्ड उपभोग कर सकते हैं।

Commit (অফসেট কমিট)

एक ऑफसेट को कन्ज्यूमড के रूप में मार्क करना ऑफसेट कमिट कहलाता है। काफ्का में, हम ऑफसेट कमिट को एक इंटरनल काफ्का टॉपिक में लिखकर रिकॉर्ड करते हैं जिसे offsets टॉपिक कहा जाता है। एक मैसेज को तभी कन्ज्यूमड माना जाता है जब उसका ऑफसेट offsets टॉपिक में कमिट हो जाता है।

Karafka में एक कॉन्फिग है जो हर बार इसे ऑटोमेटिकली करता है; कॉन्फিগ है kafka.offset_commit_interval, और इसका डिफ़ॉल्ट मान 10 सेकंड है। इसके साथ, Karafka हर 10 सेकंड में एक ऑफसेट कमिट करेगा, और आप अपने Karafka सर्वर में वह मैसेज देख सकते हैं:

*** 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 हमें बताता है कि कौन सा ऑफसेट कमिट हो रहा है; मेरे केस में, इसने काफ्का को बताया कि वह टॉपিক 0 से ऑफसेट নম্বর ৪৪ কমিট করতে পারে। इस तरह, अगर हमारे सर्विस के साथ कुछ होता है, Karafka उस ऑफসেট থেকে আবার ইভেন্ট প্রসেস করতে পারে।

উপসংহার

ইভেন্ট স্ট্রিমিং আমাদের দ্রুত হওয়া, ডেটা ভালো ব্যবহার করা, এবং ভালো ইউজার এক্সপেরিয়েন্স ডিজাইন করা সহায়তা করে। ফ্যাক্ট 그것은, অনেক কোম্পানি এই প্যাটার্ন ব্যবহার করে তাদের সকল সার্ভিস কমিউনিকেট করতে এবং রিয়েল-টাইমে ভিন্ন ইভেন্টে রিয়াক্ট করতে। যেনে 언급 করা হয়েছে, রেইলসের সাথে ব্যবহার করার জন্য Karafka ছাড়া অন্যান্য অল্টারনেটিভও রয়েছে। আপনার বেসিকস ক্লিয়ার; এখন বিনিশ্চিন্তে এক্সপেরিমেন্ট করুন।

References

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