Kafka का उपयोग करके Rails में इवेंट स्ट्रीमिंग का उपयोग

कंपनियाँ अंतर्दृष्टि प्राप्त करने और अधिक आकर्षक ग्राहक अनुभव बनाने के लिए बड़ी मात्रा में डेटा को रीयल-टाइम में प्रोसेस और साझा करने की आवश्यकता पर तुरंत प्रतिक्रिया करना चाहती हैं। इसलिए, पारंपरिक डेटा प्रोसेसिंग आज की दुनिया में अब व्यवहार्य नहीं है।

इसे प्राप्त करने के लिए, आपको जितनी जल्दी हो सके बहुत सारे डेटा को प्रोसेस करने और फिर इसे आगे प्रोसेसिंग के लिए अन्य सेवाओं को भेजने की आवश्यकता है। लेकिन इन सभी त्वरित कार्यों के बीच, जब इवेंट होता है तो उपभोक्ताओं को सूचित करना आवश्यक है—और हम इवेंट स्ट्रीमिंग का उपयोग करके ठीक ऐसा कर सकते हैं।

यह वह GitHub रेपो है जिसका हम उपयोग करेंगे।

इवेंट्स

इवेंट स्ट्रीमिंग के बारे में बात करने से पहले, आइए बात करते हैं कि इवेंट क्या है। एक एप्लिकेशन के भीतर होने वाला एक इवेंट उपयोगकर्ता प्रक्रिया से संबंधित हो सकता है या केवल ऐसी क्रियाएं जो व्यवसाय को प्रभावित करती हैं।

इवेंट्स एक स्थिति परिवर्तन का प्रतिनिधित्व करते हैं, न कि एप्लिकेशन को कैसे संशोधित किया जाए इसका प्रश्न। इन्हें उदाहरण मानें:

• एक उपयोगकर्ता किसी सेवा में लॉग इन कर रहा है
• एक भुगतान लेनदेन
• एक लेखक ब्लॉग में एक पोस्ट प्रकाशित कर रहा है

अधिकांश मामलों में, एक इवेंट अधिक इवेंट्स को ट्रिगर करेगा; उदाहरण के लिए, जब कोई उपयोगकर्ता किसी सेवा के लिए साइन अप करता है, तो ऐप उनके डिवाइस पर एक सूचना भेजता है, डेटाबेस में रिकॉर्ड डालता है, और एक स्वागत ईमेल भेजता है।

इवेंट स्ट्रीमिंग

इवेंट स्ट्रीमिंग डेटाबेस जैसे इवेंट स्रोतों से रीयल-टाइम में डेटा कैप्चर करने का एक पैटर्न है। इवेंट स्ट्रीमिंग के मुख्य भाग निम्न हैं:

• ब्रोकर: इवेंट्स स्टोर करने के लिए जिम्मेदार सिस्टम
• टॉपिक: इवेंट्स की एक श्रेणी
• प्रोड्यूसर: एक विशिष्ट टॉपिक पर ब्रोकर को इवेंट्स भेजता है
• कंज्यूमर: इवेंट्स पढ़ता है
• इवेंट्स: डेटा जो प्रोड्यूसर कंज्यूमर को संवाद करना चाहते हैं

इस बिंदु पर पब्लिश और सब्सक्राइब आर्किटेक्चर पैटर्न (पब/सब पैटर्न) के बारे में बात करना अपरिहार्य है; इवेंट स्ट्रीमिंग उस पैटर्न का एक कार्यान्वयन है लेकिन इन परिवर्तनों के साथ:

• मैसेज के बजाय इवेंट्स होते हैं।
• इवेंट्स क्रमबद्ध होते हैं, आमतौर पर समय के अनुसार।
• कंज्यूमर टॉपिक में एक विशिष्ट बिंदु से इवेंट्स पढ़ सकते हैं।
• इवेंट्स में अस्थायी स्थायित्व होता है।

फ्लो तब शुरू होता है जब प्रोड्यूसर एक नया इवेंट एक टॉपिक में प्रकाशित करता है (जैसा कि हमने पहले देखा, टॉपिक केवल एक विशिष्ट प्रकार के इवेंट के लिए वर्गीकरण है)। फिर, किसी विशेष श्रेणी के इवेंट्स में रुचि रखने वाले कंज्यूमर उस टॉपिक को सब्सक्राइब करते हैं। अंत में, ब्रोकर टॉपिक के कंज्यूमर्स की पहचान करता है और वांछित इवेंट्स को उपलब्ध कराता है।

इवेंट स्ट्रीमिंग के लाभ

• डिकपलिंग पब्लिशर्स और कंज्यूमर्स के बीच कोई निर्भरता नहीं है क्योंकि उन्हें एक-दूसरे को जानने की आवश्यकता नहीं है। इसके अलावा, इवेंट्स अपनी क्रियाओं को निर्दिष्ट नहीं करते हैं, इसलिए कई कंज्यूमर्स एक ही इवेंट प्राप्त कर सकते हैं और अलग-अलग क्रियाएं कर सकते हैं।

• कम विलंबता इवेंट्स डिकपल्ड हैं और कंज्यूमर को उन्हें कभी भी उपयोग करने देते हैं; यह मिलीसेकंड में हो सकता है।

• स्वतंत्रता जैसा कि हम जानते हैं, पब्लिशर्स और कंज्यूमर्स स्वतंत्र हैं, इसलिए अलग-अलग टीमें उन्हीं इवेंट्स का उपयोग अन्य क्रियाओं या उद्देश्यों के लिए कर सकती हैं।

• फॉल्ट टॉलरेंस कुछ इवेंट स्ट्रीमिंग प्लेटफॉर्म कंज्यूमर विफलताओं से निपटने में आपकी मदद करते हैं; उदाहरण के लिए, कंज्यूमर अपनी स्थिति सहेज सकते हैं और त्रुटि होने पर वहीं से फिर से शुरू कर सकते हैं।

• रीयल-टाइम हैंडलिंग फीडबैक रीयल-टाइम में प्राप्त होता है, इसलिए उपयोगकर्ताओं को अपने इवेंट्स की प्रतिक्रिया देखने के लिए मिनटों या घंटों तक इंतजार करने की आवश्यकता नहीं होती।

• उच्च प्रदर्शन इवेंट प्लेटफॉर्म कम विलंबता के कारण कई मैसेज संभाल सकते हैं—उदाहरण के लिए, एक सेकंड में हजारों इवेंट्स।

इवेंट स्ट्रीमिंग के नुकसान

• मॉनिटरिंग कुछ इवेंट स्ट्रीमिंग टूल्स में पूर्ण मॉनिटरिंग टूल नहीं होता; वे अतिरिक्त टूल्स को लागू करने के लिए कहते हैं, जैसे Datadog या New Relic।

• कॉन्फ़िगरेशन कुछ टूल्स में कॉन्फ़िगरेशन अनुभवी लोगों के लिए भी भारी हो सकता है। कई पैरामीटर हैं, और कभी-कभी, आपको उन्हें लागू करने के लिए विषय के बारे में गहराई से जानने की आवश्यकता होती है।

• क्लाइंट लाइब्रेरीज़ Java के अलावा अन्य भाषाओं में Kafka को लागू करना आसान नहीं है। कभी-कभी, क्लाइंट लाइब्रेरीज़ अपडेट नहीं होतीं, अस्थिरता दिखाती हैं, या चुनने के लिए कई विकल्प नहीं देतीं।

इवेंट स्ट्रीमिंग के लिए सबसे लोकप्रिय टूल्स में से एक Apache Kafka है। यह टूल उपयोगकर्ताओं को डेटा भेजने, स्टोर करने और जब भी और जहाँ भी उन्हें आवश्यकता हो, अनुरोध करने की अनुमति देता है; आइए इसके बारे में बात करते हैं।

Apache Kafka

"Apache Kafka हाई-परफॉरमेंस डेटा पाइपलाइन्स, स्ट्रीमिंग एनालिटिक्स, डेटा इंटीग्रेशन, और मिशन-क्रिटिकल एप्लिकेशन के लिए हजारों कंपनियों द्वारा उपयोग किया जाने वाला एक ओपन-सोर्स डिस्ट्रिब्यूटेड इवेंट स्ट्रीमिंग प्लेटफॉर्म है।"

विशेष रूप से रीयल-टाइम लॉग ट्रांसमिशन के लिए डिज़ाइन किया गया, Apache Kafka उन एप्लिकेशन के लिए आदर्श है जिन्हें निम्न की आवश्यकता होती है:

• विभिन्न घटकों के बीच विश्वसनीय डेटा एक्सचेंज
• एप्लिकेशन आवश्यकताओं के बदलने पर मैसेजिंग वर्कलोड को विभाजित करने की क्षमता
• डेटा प्रोसेसिंग के लिए रीयल-टाइम ट्रांसमिशन

आइए Rails एप्लिकेशन में Kafka का उपयोग करें!

Rails के साथ Kafka का उपयोग

Ruby में Kafka का उपयोग करने के लिए सबसे प्रसिद्ध जेम Zendesk द्वारा ruby-kafka कहलाता है, और यह बहुत अच्छा है! फिर भी, आपको सभी कार्यान्वयन मैन्युअल रूप से करने की आवश्यकता है, यही कारण है कि हमारे पास ruby-kafka के साथ बने कुछ "फ्रेमवर्क" हैं। वे कॉन्फ़िगरेशन और निष्पादन चरणों में भी हमारी मदद करते हैं।

Karafka, Apache Kafka-आधारित Ruby एप्लिकेशन विकास को सरल बनाने के लिए उपयोग किया जाने वाला एक फ्रेमवर्क है।

Kafka के साथ काम करने के लिए, Java इंस्टॉल करना आवश्यक है। क्योंकि Kafka एक Scala और Java एप्लिकेशन भी है, Zookeeper इंस्टॉल करना भी आवश्यक होगा।

इंस्टॉलेशन से पहले, मैं Zookeeper के बारे में थोड़ा समझाना चाहता हूँ। Zookeeper, Kafka के लिए आवश्यक एक केंद्रीकृत सेवा है; यह नए टॉपिक के निर्माण, ब्रोकर के क्रैश, ब्रोकर के हटाने, टॉपिक्स के विलोपन आदि जैसे परिवर्तनों के मामले में सूचनाएं भेजता है।

इसका मुख्य कार्य Kafka ब्रोकर्स को प्रबंधित करना, उनके संबंधित मेटाडेटा के साथ एक सूची बनाए रखना, और हेल्थ-चेकिंग तंत्र की सुविधा प्रदान करना है। इसके अलावा, यह टॉपिक्स के विभिन्न पार्टीशन्स के लिए अग्रणी ब्रोकर का चयन करने में मदद करता है।

आवश्यकताएँ

MacOS के लिए:

अब, निम्न कमांड के साथ Java और Zookeeper इंस्टॉल करें:

brew install java
brew install zookeeper

फिर, हम इसे चलाकर Kafka इंस्टॉल करना जारी रख सकते हैं:

brew install kafka

एक बार हमारे पास Kafka और Zookeeper इंस्टॉल हो जाने के बाद, सेवाओं को इस तरह शुरू करना आवश्यक है:

brew services start zookeeper
brew services start kafka

Windows और Linux के लिए:

निर्देश:

1. Java इंस्टॉल करना
2. Zookeeper डाउनलोड करना

Rails की सेटिंग

बस हमेशा की तरह एक साधारण Rails एप्लिकेशन बनाएँ:

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 इनिशियलाइज़र

karafka.rb फाइल Rails कॉन्फिग से अलग एक इनिशियलाइज़र एप्लिकेशन की तरह है। यह आपको Karafka एप्लिकेशन को कॉन्फ़िगर करने और कुछ रूट्स बनाने की अनुमति देता है, API के संदर्भ में Rails एप्लिकेशन रूट्स के समान। लेकिन यहाँ, यह टॉपिक्स और कंज्यूमर्स के लिए है।

प्रोड्यूसर

प्रोड्यूसर इवेंट्स बनाने के प्रभारी होते हैं, और हम उन्हें 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 का उपयोग करते हैं। लेकिन यदि आप इवेंट्स को बैचों में पढ़ेंगे और आपके पास कॉन्फिग config.batch_fetching true के रूप में है, तो आपको params_batch का उपयोग करना चाहिए।

टेस्टिंग

हमारी Karafka सेवा चलाने के लिए (जो इवेंट्स सुनेगी), कंसोल पर जाएँ, एक नया टैब खोलें, Rails प्रोजेक्ट पर जाएँ, और चलाएँ:

bundle exec karafka server

सफल इवेंट

अब, एक और कंसोल टैब खोलें, Rails प्रोजेक्ट पर जाएँ, और यह टाइप करें:

rails c

वहाँ, आइए अपने रेस्पॉन्डर के साथ एक इवेंट बनाएँ:

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

यदि आप Rails कंसोल देखें, तो इवेंट बनने के बाद हमें यह मैसेज प्राप्त होगा:

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

विफल इवेंट

आप निम्न कमांड चलाकर email, first_name और last_name जैसे कुछ एट्रिब्यूट्स के साथ एक User मॉडल बना सकते हैं:

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। लेकिन यहाँ दिलचस्प बात यह है कि Kafka उस इवेंट को बार-बार प्रोसेस करने का प्रयास करेगा। यहां तक कि यदि आप Karafka सर्वर को रीस्टार्ट करते हैं, तो यह अंतिम इवेंट को प्रोसेस करने का प्रयास करेगा। Kafka को कैसे पता चलता है कि कहाँ से शुरू करना है?

यदि आप अपना कंसोल देखते हैं, तो त्रुटि के बाद, आप यह देखेंगे:

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

यह आपको बताएगा कि कौन सा ऑफसेट प्रोसेस किया गया था: इस मामले में, यह ऑफसेट 42 था। तो, यदि आप Karafka सर्विस को रीस्टार्ट करते हैं, तो यह उस ऑफसेट से शुरू होगा।

[[karafka_example] {}:] Committing offsets with recommit: users/0:42
[[karafka_example] {users: 0}:] 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।

कॉलबैक्स

यह कुछ अच्छा है जो Karafka प्रदान करता है: आप अपने कंज्यूमर में कॉलबैक्स का उपयोग कर सकते हैं। ऐसा करने के लिए, आपको केवल मॉड्यूल आयात करना होगा और उनका उपयोग करना होगा। फिर, अपना 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

पोल वह माध्यम है जिसके माध्यम से हम वर्तमान पार्टीशन ऑफसेट के आधार पर रिकॉर्ड्स प्राप्त करते हैं। तो, वे कॉलबैक्स before_poll और after_poll, जैसा कि उनके नाम से पता चलता है, उस क्षण निष्पादित होते हैं। हम केवल एक मैसेज लॉग कर रहे हैं, और आप उन्हें अपने Karafka सर्वर में देख सकते हैं—एक फेचिंग से पहले और दूसरा उसके बाद:

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

हार्टबीट्स

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

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... के साथ, Kafka जानता है कि हम जीवित हैं और हम इसके कंज्यूमर ग्रुप के एक वैध सदस्य हैं। इसके अलावा, हम अधिक रिकॉर्ड्स का उपभोग कर सकते हैं।

कमिट

एक ऑफसेट को उपभोग के रूप में चिह्नित करना ऑफसेट कमिट करना कहलाता है। Kafka में, हम आंतरिक Kafka टॉपिक में लिखकर ऑफसेट कमिट रिकॉर्ड करते हैं जिसे ऑफसेट्स टॉपिक कहा जाता है। एक मैसेज को केवल तभी उपभोग किया हुआ माना जाता है जब उसका ऑफसेट ऑफसेट्स टॉपिक में कमिट हो जाता है।

Karafka के पास इसे हर बार स्वचालित रूप से करने के लिए एक कॉन्फिग है; कॉन्फिग है kafka.offset_commit_interval, और इसका मान डिफ़ॉल्ट रूप से 10 सेकंड है। इसके साथ, Karakfa हर 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 हमें बताता है कि वह कौन सा ऑफसेट कमिट कर रहा है; मेरे मामले में, इसने Kafka को बताया कि वह टॉपिक 0 से ऑफसेट नंबर 44 कमिट कर सकता है। इस तरह, यदि हमारी सर्विस के साथ कुछ होता है, तो Karafka उस ऑफसेट से फिर से इवेंट्स प्रोसेस करना शुरू कर सकता है।

निष्कर्ष

इवेंट स्ट्रीमिंग हमें तेज होने, डेटा का बेहतर उपयोग करने, और बेहतर उपयोगकर्ता अनुभव डिजाइन करने में मदद करती है। तथ्य की बात के रूप में, कई कंपनियाँ अपनी सभी सेवाओं को संचार करने और विभिन्न इवेंट्स पर रीयल-टाइम में प्रतिक्रिया करने में सक्षम होने के लिए इस पैटर्न का उपयोग कर रही हैं। जैसा कि मैंने पहले उल्लेख किया था, Karafka के अलावा अन्य विकल्प हैं जिनका आप Rails के साथ उपयोग कर सकते हैं। आपके पास पहले से ही मूल बातें हैं; अब, उनके साथ प्रयोग करने के लिए स्वतंत्र महसूस करें।

संदर्भ

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