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

कुछ मामलों में तैयार HTTP सेवा बैकएंड का उपयोग करना एक संदेश कतार को संसाधित करने के लिए एक आदर्श समाधान है:

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

इस दृष्टिकोण के नुकसान भी हैं:

आप समाधान की लपट के बारे में भूल सकते हैं। HTTP एक भारी प्रोटोकॉल है, और उपभोक्ता के पक्ष में चौखटे का उपयोग तुरन्त विलंबता और भार को बढ़ाएगा।
हम पुश दृष्टिकोण की कमजोरियों को देखते हुए पोल दृष्टिकोण की ताकत खो देते हैं।
संदेशों को उसी सेवा इंस्टेंसेस द्वारा संसाधित करना जो क्लाइंट को प्रोसेस करते हैं, जवाबदेही को प्रभावित कर सकते हैं। यह महत्वपूर्ण नहीं है, क्योंकि यह संतुलन और अलगाव के साथ व्यवहार किया जाता है।

मैंने विचार को क्यू-ओवर-एचटीपी सेवा के रूप में लागू किया, जिस पर बाद में चर्चा की जाएगी। परियोजना को कोटलिन में स्प्रिंग बूट 2.1 का उपयोग करके लिखा गया है। एक दलाल के रूप में, केवल अपाचे काफ्का वर्तमान में उपलब्ध है।

लेख में आगे, यह माना जाता है कि पाठक काफ्का से परिचित है और संदेशों (समूहों) और उपभोक्ताओं (उपभोक्ता) के सिद्धांतों के कमिट (ऑफ़सेट) और ऑफ़सेट (ऑफ़सेट) के बारे में जानता है, और यह भी समझता है कि विभाजन (विभाजन) विषय (विषय) से कैसे भिन्न होता है । यदि अंतराल हैं, तो मैं आपको सलाह देता हूं कि जारी रखने से पहले काफ्का प्रलेखन के इस भाग को पढ़ें।

सामग्री

सिंहावलोकन

कतार-ओवर-एचटीपी एक सेवा है जो एक संदेश दलाल और अंतिम HTTP उपभोक्ता के बीच मध्यस्थ के रूप में कार्य करता है (सेवा किसी भी अन्य तरीके से उपभोगकर्ताओं को संदेश भेजने के लिए समर्थन को लागू करना आसान बनाती है, उदाहरण के लिए, विभिन्न * आरपीसी)। फिलहाल, उपभोक्ताओं की सूची को केवल सब्सक्राइब, अनसब्सक्राइब करना और देखना उपलब्ध है। निर्माता के विशेष समर्थन के बिना संदेशों के आदेश की गारंटी देने की अक्षमता के कारण, HTTP के माध्यम से ब्रोकर (उपज) को संदेश भेजना अभी तक लागू नहीं किया गया है।

सेवा का मुख्य आंकड़ा उपभोक्ता है, जो विशिष्ट विभाजन या केवल विषयों (विषय पैटर्न समर्थित है) की सदस्यता ले सकता है। पहले मामले में, विभाजन का स्वत: संतुलन बंद हो जाता है। सदस्यता लेने के बाद, निर्दिष्ट HTTP संसाधन को असाइन किए गए काफ्का विभाजन के संदेश प्राप्त होने लगते हैं। वास्तुकला में, प्रत्येक ग्राहक एक देशी काफ्का जावा क्लाइंट के साथ जुड़ा हुआ है।

कफ़्का कोंसुमेर के बारे में मनोरंजक कहानी

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

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

while (!Thread.interrupted()) { var hasWork = false for (consumer in kafkaConsumers) { val queueGroup = consumers[consumer] ?: continue invalidateSubscription(consumer, queueGroup) val records = consumer.poll(Duration.ZERO) /*      */ if (!records.isEmpty) { hasWork = true } } val committed = doCommit() if (!hasWork && committed == 0) { // ,    Thread.sleep(1) } }

ऐसा लगता है कि सब कुछ अद्भुत है, दर्जनों उपभोक्ताओं के साथ भी विलंबता न्यूनतम है। व्यवहार में, यह पता चला कि KafkaConsumer ऑपरेशन के इस मोड के लिए KafkaConsumer और निष्क्रिय समय में लगभग 1.5 एमबी / एस की आवंटन दर देता है। 100 कोरियर के साथ, आवंटन दर 150 एमबी / एस तक पहुंच जाती है और जीसी को अक्सर आवेदन के बारे में सोचता है। बेशक, यह सभी कचरा युवा क्षेत्र में है, जीसी इसे संभालने में काफी सक्षम है, लेकिन फिर भी, समाधान सही नहीं है।

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

मैं ऊपर से कोड को फिर से लिखता हूं, आंतरिक लूप को हटाकर Duration.ZERO को Duration.ofMillis(100) बदल देता हूं। यह अच्छी तरह से पता चला है, आवंटन की दर स्वीकार्य प्रति उपभोक्ता 80-150 केबी / एस तक गिरती है। हालाँकि, 100ms के समय के साथ मतदान, इन्हीं 100ms पर आने वाली संपूर्ण कतार को विलंबित कर देता है, और यह अस्वीकार्य है।

समस्या के समाधान खोजने की प्रक्रिया में, मुझे KafkaConsumer::wakeup याद है, जो एक WakeupException फेंकता है और उपभोक्ता पर किसी भी अवरुद्ध ऑपरेशन को बाधित करता है। इस पद्धति के साथ, कम-विलंबता का मार्ग सरल है: जब प्रतिबद्ध के लिए एक नया अनुरोध आता है, तो हम इसे कतार में रख देते हैं, और मूल उपभोक्ता पर हम wakeup कहते हैं। कार्य चक्र में, WakeupException को WakeupException और कमिट करें कि क्या जमा हुआ है। अपवादों की मदद से नियंत्रण के हस्तांतरण के लिए, आपको तुरंत इसे अपने हाथों में देना चाहिए, लेकिन चूंकि और कुछ नहीं ...

यह पता चला है कि यह विकल्प सही से बहुत दूर है, क्योंकि देशी उपभोक्ता पर कोई भी ऑपरेशन अब WakeupException , WakeupException सहित। इस स्थिति को संसाधित करना कोड को फ्लैग के साथ अव्यवस्था की अनुमति देगा।

मैं इस नतीजे पर KafkaConsumer::poll कि KafkaConsumer::poll पद्धति को संशोधित करना अच्छा होगा ताकि अतिरिक्त ध्वज के अनुसार इसे सामान्य रूप से बाधित किया जा सके। नतीजतन, फ्रेंकस्टीन प्रतिबिंब से पैदा हुआ था, जो मूल मतदान विधि की प्रतिलिपि बनाता है, ध्वज द्वारा लूप से बाहर निकलने को जोड़ता है। यह ध्वज एक अलग से व्यवधान विधि द्वारा सेट किया गया है, जो, इसके अलावा, ग्राहक चयनकर्ता को I / O संचालन पर थ्रेड लॉक जारी करने के लिए जगाता है।

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

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

सेवा थ्रूपुट को बढ़ाने के लिए बैच संदेश प्रसंस्करण का समर्थन करती है। जब सदस्यता लेते हैं, तो आप प्रत्येक विषय के concurrencyFactor निर्दिष्ट कर सकते हैं (स्वतंत्र रूप से प्रत्येक निर्दिष्ट विभाजन पर लागू होता है)। उदाहरण के लिए, concurrencyFactor=1000 अर्थ है कि HTTP अनुरोधों के रूप में 1000 संदेश उसी समय उपभोक्ता को भेजे जा सकते हैं। जैसे ही पैक से सभी संदेशों को उपभोक्ता द्वारा स्पष्ट रूप से काम किया गया, सेवा काफका में अंतिम संदेश की ऑफसेट की अगली प्रतिबद्ध पर निर्णय लेती है। इसलिए, कंफ़ेक्टफैक्टर का दूसरा मूल्य कफ़्का या क्यू-ओवर-एचटीपी दुर्घटना की स्थिति में उपभोक्ता द्वारा संसाधित संदेशों की अधिकतम संख्या है।

देरी को कम करने के लिए, कतार में loadFactor = concurrencyFactor * 2 , जो आपको ब्रोकर के दो संदेशों को पढ़ने के लिए दो बार पढ़ने की अनुमति देता है। चूंकि ऑटोकॉमिट देशी क्लाइंट पर अक्षम है, इसलिए ऐसी योजना एट-लिस्ट-वन्स गारंटी का उल्लंघन नहीं करती है।
एक उच्च concurrencyFactor मान कतार के थ्रूपुट को बढ़ाता है सबसे बुरे मामले में कम से कम 10 एमएस तक ले जाता है। इसी समय, उपभोक्ता पर भार बढ़ता है।

बंडल के भीतर संदेश भेजने के आदेश की गारंटी नहीं है, लेकिन इसे concurrencyFactor=1 निर्धारित करके प्राप्त किया जा सकता है concurrencyFactor=1 ।

प्रतिबद्ध

कमिट सेवा का एक महत्वपूर्ण हिस्सा हैं। जब डेटा का अगला पैकेट तैयार हो जाता है, तो पैकेट से अंतिम संदेश की भरपाई तुरंत काफ्का को कर दी जाती है, और एक सफल प्रतिबद्धता के बाद ही अगला पैकेट प्रसंस्करण के लिए उपलब्ध होता है। अक्सर यह पर्याप्त नहीं होता है और एक ऑटो कमिट की आवश्यकता होती है। ऐसा करने के लिए, पैरामीटर autoCommitPeriodMs , जिसमें मूल क्लाइंट के लिए शास्त्रीय ऑटोकॉमिट अवधि के साथ सामान्य रूप से कम है जो विभाजन से पढ़े गए अंतिम संदेश को प्रतिबद्ध करते हैं। कल्पना कीजिए concurrencyFactor=10 । सेवा ने सभी 10 संदेश भेजे हैं और उनमें से प्रत्येक के तैयार होने की प्रतीक्षा है। संदेश 3 का प्रसंस्करण पहले पूरा हो गया है, फिर संदेश 1, और फिर, संदेश 10। इस बिंदु पर, यह ऑटो-कमिट का समय है। यह महत्वपूर्ण है कि एट-लिस्ट-वन्स शब्दार्थ का उल्लंघन न करें। इसलिए, आप केवल पहला संदेश दे सकते हैं, वह है, ऑफसेट 2, क्योंकि केवल उस समय इसे सफलतापूर्वक संसाधित किया गया था। इसके अलावा, अगले ऑटोकॉमिट तक, संदेश 2, 5, 6, 4, और 8 संसाधित किए जाते हैं। अब आपको केवल 7 ऑफसेट करने की आवश्यकता है, और इसी तरह। ऑटोकॉमिट का थ्रूपुट पर लगभग कोई प्रभाव नहीं है।

हैंडलिंग में त्रुटि

ऑपरेशन के सामान्य मोड में, सेवा पर्यवेक्षक को एक बार संदेश भेजती है। यदि किसी कारण से इसमें 4xx या 5xx त्रुटि हुई, तो सेवा संदेश को फिर से भेज देगी, जो सफल प्रसंस्करण की प्रतीक्षा कर रही है। प्रयासों के बीच के समय को एक अलग पैरामीटर के रूप में कॉन्फ़िगर किया जा सकता है।

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

चिपके रहने के बारे में

आमतौर पर, HTTP सर्वर, 4xx या 5xx को रिस्पांस का दर्जा देते हुए, Connection: close हेडर भी भेजता है। इस तरह से बंद किया गया एक टीसीपी कनेक्शन TIME_WAITED स्थिति में रहता है जब तक कि इसे कुछ समय बाद ऑपरेटिंग सिस्टम द्वारा साफ़ नहीं किया जाता है। समस्या यह है कि इस तरह के कनेक्शन पूरे पोर्ट पर कब्जा कर लेते हैं, जो जारी होने तक पुन: उपयोग नहीं किए जा सकते यह टीसीपी कनेक्शन स्थापित करने के लिए मशीन पर नि: शुल्क बंदरगाहों की अनुपस्थिति के परिणामस्वरूप हो सकता है और प्रत्येक भेजने के बाद लॉग में अपवाद के साथ सेवा को फेंक दिया जाएगा। व्यवहार में, विंडोज 10 पर, पोर्ट 10-20 हजार के बाद समाप्त हो जाते हैं और 1-2 मिनट के भीतर गलत संदेश भेजते हैं। मानक मोड में, यह कोई समस्या नहीं है।

संदेशों

ब्रोकर से निकाले गए प्रत्येक संदेश को सलाहकार के पास HTTP के माध्यम से सदस्यता के दौरान निर्दिष्ट संसाधन पर भेजा जाता है। डिफ़ॉल्ट रूप से, शरीर में POST अनुरोध द्वारा एक संदेश भेजा जाता है। किसी अन्य विधि को निर्दिष्ट करके इस व्यवहार को बदला जा सकता है। यदि विधि शरीर में डेटा भेजने का समर्थन नहीं करती है, तो आप उस स्ट्रिंग पैरामीटर का नाम निर्दिष्ट कर सकते हैं जिसमें संदेश भेजा जाएगा। इसके अलावा, सदस्यता लेते समय, आप अतिरिक्त हेडर निर्दिष्ट कर सकते हैं जो प्रत्येक संदेश में जोड़ा जाएगा, जो टोकन का उपयोग करके मूल प्राधिकरण के लिए सुविधाजनक है। उपभोक्ता, विषय और विभाजन के पहचानकर्ता के साथ प्रत्येक संदेश में हेडर जोड़े जाते हैं, जहां से संदेश पढ़ा गया था, संदेश संख्या, विभाजन कुंजी, यदि लागू हो, साथ ही साथ दलाल का नाम।

उत्पादकता

प्रदर्शन का मूल्यांकन करने के लिए, मैंने एक पीसी (विंडोज 10, ओपनजेडके -11 (ट्यूनिंग के बिना जी 1), i7-6700K, 16GB) का उपयोग किया, जो सेवा और एक लैपटॉप चलाता है (विंडोज 10, i5-8250U, 8GB), जिस पर संदेश निर्माता, HTTP घूम रहा था। डिफ़ॉल्ट सेटिंग्स के साथ संसाधन उपभोक्ता और काफ्का। पीसी 1Gb / s वायर्ड कनेक्शन, 802.11ac के माध्यम से लैपटॉप के माध्यम से राउटर से जुड़ा हुआ है। निर्माता प्रत्येक 110 एमएस को प्रत्येक 100 एमएस पर 110 बाइट्स के लिए निर्दिष्ट विषयों के लिए संदेशों को लिखता है, जिसके लिए अनुयायियों को सदस्यता दी जाती है (अलग-अलग समूहों से concurrencyFactor=500 ( concurrencyFactor=500 , ऑटो-कमिट बंद कर दिया जाता है)। स्टैंड आदर्श से बहुत दूर है, लेकिन आप कुछ चित्र प्राप्त कर सकते हैं।

एक प्रमुख माप पैरामीटर विलंबता पर सेवा का प्रभाव है।

करते हैं:
- मूल ग्राहक से संदेश प्राप्त करने वाली सेवा का टाइम _q - टाइमस्टैम्प
- d _t0 , t _{q के} बीच का समय है और संदेश को स्थानीय कतार से अधिकारियों के पूल में भेजा जाता है
- d _t , t _q और HTTP अनुरोध भेजने के समय के बीच का समय है। यह d _t संदेश के विलंबता पर सेवा का प्रभाव है।

माप के दौरान, निम्नलिखित परिणाम प्राप्त किए गए (सी - उपभोक्ता, टी - विषय, एम - संदेश):

मानक ऑपरेटिंग मोड में, सेवा स्वयं लगभग विलंबता को प्रभावित नहीं करती है, और मेमोरी की खपत न्यूनतम है। D _t (लगभग 60ms) के अधिकतम मूल्यों को विशेष रूप से इंगित नहीं किया गया है, क्योंकि वे जीसी के संचालन पर निर्भर करते हैं, और सेवा पर ही नहीं। जीसी की विशेष ट्यूनिंग या शेनडोनह के साथ जी 1 की जगह अधिकतम मूल्यों के प्रसार को सुचारू बनाने में मदद कर सकती है।

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

दुर्भाग्य से, उच्च भार पर परीक्षण करना संभव नहीं है, क्योंकि लैपटॉप 1300 आरपीएस पर पहले से ही झुकता है। यदि कोई उच्च भार पर माप के संगठन के साथ मदद कर सकता है, तो मैं खुशी से परीक्षणों के लिए एक विधानसभा प्रदान करूंगा।

प्रदर्शन

अब हम प्रदर्शन के लिए आगे बढ़ते हैं। इसके लिए हमें चाहिए:

काफ्का दलाल, जाने को तैयार। मैं बिटनमी से 192.168.99.100:9092 पर उठाया गया उदाहरण लूंगा।
एक HTTP संसाधन जो संदेश प्राप्त करेगा। स्पष्टता के लिए, मैंने स्लैक से वेब-हुक लिया।

सबसे पहले, आपको स्वयं क्यू-ओवर-एचटीपी सेवा को बढ़ाने की आवश्यकता है। ऐसा करने के लिए, एक खाली application.yml निर्देशिका में निम्नलिखित सामग्री बनाएँ:

 spring: profiles: default logging: level: com: viirrtus: queueOverHttp: DEBUG app: persistence: file: storageDirectory: "persist" brokers: - name: "Kafka" origin: "kafka" config: bootstrap.servers: "192.168.99.100:9092"

यहां हम सेवा को एक विशिष्ट ब्रोकर के कनेक्शन मापदंडों के साथ-साथ ग्राहकों को संग्रहीत करने के लिए कहां से शुरू करते हैं ताकि वे शुरू होने के बीच खो न जाएं। `App.brokers []। Config में, आप देशी काफ्का क्लाइंट द्वारा समर्थित किसी भी कनेक्शन पैरामीटर को निर्दिष्ट कर सकते हैं, एक पूरी सूची यहां पाई जा सकती है ।

चूंकि कॉन्फ़िगरेशन फ़ाइल स्प्रिंग द्वारा संसाधित की जाती है, इसलिए आप वहां बहुत सारी दिलचस्प चीजें लिख सकते हैं। सहित, लॉगिंग कॉन्फ़िगर करें।

अब सेवा ही चलाओ। हम सबसे आसान तरीका उपयोग करते हैं - docker-compose.yml :

 version: "2" services: app: image: viirrtus/queue-over-http:0.1.3 restart: unless-stopped command: --debug ports: - "8080:8080" volumes: - ./application.yml:/application.yml - ./persist:/persist

यदि यह विकल्प आपको सूट नहीं करता है, तो आप स्रोत से सेवा संकलित कर सकते हैं। रीडमी परियोजना में विधानसभा निर्देश, एक लिंक जो लेख के अंत में दिया गया है।

अगला चरण पहले ग्राहक को पंजीकृत कर रहा है। ऐसा करने के लिए, आपको उपभोक्ता के विवरण के साथ सेवा के लिए HTTP अनुरोध करने की आवश्यकता है:

 POST localhost:8080/broker/subscription Content-Type: application/json { "id": "my-first-consumer", "group": { "id": "consumers" }, "broker": "Kafka", "topics": [ { "name": "slack.test", "config": { "concurrencyFactor": 10, "autoCommitPeriodMs": 100 } } ], "subscriptionMethod": { "type": "http", "delayOnErrorMs": 1000, "retryBeforeCommit": 10, "uri": "<slack-wh-uri>", "additionalHeaders": { "Content-Type": "application/json" } } }

यदि सब कुछ ठीक रहा, तो प्रतिक्रिया लगभग एक ही भेजी गई सामग्री होगी।

चलो प्रत्येक पैरामीटर के माध्यम से चलते हैं:

Consumer.id - हमारे ग्राहक की आईडी
Consumer.group.id - समूह पहचानकर्ता
Consumer.broker - इंगित करें कि आपको कौन से सेवा दलालों की सदस्यता लेने की आवश्यकता है
Consumer.topics[0].name - उस विषय का नाम जिससे हम संदेश प्राप्त करना चाहते हैं
Consumer.topics[0].config. concurrencyFactor Consumer.topics[0].config. concurrencyFactor - एक साथ भेजे गए संदेशों की अधिकतम संख्या
Consumer.topics[0].config. autoCommitPeriodMs Consumer.topics[0].config. autoCommitPeriodMs - तैयार संदेशों के लिए प्रतिबद्ध अवधि
Consumer.subscriptionMethod.type - सदस्यता का प्रकार। वर्तमान में केवल HTTP उपलब्ध है।
Consumer.subscriptionMethod.delayOnErrorMs - एक त्रुटि में समाप्त होने वाले संदेश को संशोधित करने से पहले का समय
Consumer.subscriptionMethod.retryBeforeCommit - त्रुटि संदेश को पुनः भेजने के प्रयासों की संख्या। यदि 0 - संदेश सफल प्रसंस्करण तक स्पिन होगा। हमारे मामले में, पूर्ण वितरण की गारंटी प्रवाह की गति के रूप में महत्वपूर्ण नहीं है।
Consumer.subscriptionMethod.uri - वह संसाधन जिस पर संदेश भेजे जाएंगे
Consumer.subscriptionMethod.additionalHeader - प्रत्येक संदेश के साथ भेजे जाने वाले अतिरिक्त शीर्ष लेख। ध्यान दें कि प्रत्येक संदेश के शरीर में JSON होगा ताकि स्लैक अनुरोध को सही ढंग से व्याख्या कर सके।

इस अनुरोध में, HTTP विधि को छोड़ दिया गया है, क्योंकि डिफ़ॉल्ट, POST, Slack काफी ठीक है।

इस पल से, सेवा नए संदेशों के लिए slack.test विषय के निर्दिष्ट विभाजन की निगरानी करती है।

विषय को संदेश लिखने के लिए, मैं काफ्का में निर्मित उपयोगिताओं का उपयोग करूंगा जो कि लॉन्च /opt/bitnami/kafka/bin गई काफ्का छवि के /opt/bitnami/kafka/bin (अन्य काफ्का उदाहरणों में उपयोगिताओं का स्थान भिन्न हो सकता है):

 kafka-console-producer.sh --broker-list localhost:9092 --topic slack.test > {“text”: “Hello!”}

उसी समय, स्लैक आपको एक नए संदेश की सूचना देगा:

किसी उपभोक्ता को अनसब्सक्राइब करने के लिए, 'ब्रोकर / अनसब्सक्राइब' के लिए उसी सामग्री के साथ POST अनुरोध करना पर्याप्त है जो सब्सक्रिप्शन के दौरान था।

निष्कर्ष

फिलहाल, केवल बुनियादी कार्यक्षमता को लागू किया गया है। इसके अलावा, बैचिंग में सुधार करने की योजना बनाई गई है, सटीक-एक बार शब्दार्थ को लागू करने की कोशिश करें, HTTP के माध्यम से ब्रोकर को संदेश भेजने की क्षमता जोड़ें और सबसे महत्वपूर्ण बात यह है कि अन्य लोकप्रिय पब-सब के लिए समर्थन जोड़ें।

कतार-ओवर-एचटीपी सेवा वर्तमान में सक्रिय विकास के तहत है। संस्करण 0.1.3 देव और मंच के स्टैंड पर परीक्षण के लिए पर्याप्त स्थिर है। प्रदर्शन का परीक्षण विंडोज 10, डेबियन 9 और उबंटू 18.04 पर किया गया है। आप अपने जोखिम पर ठेस का उपयोग कर सकते हैं। यदि आप विकास में मदद करना चाहते हैं या सेवा पर कोई प्रतिक्रिया देना चाहते हैं - तो जीथब परियोजना में आपका स्वागत है।

HTTP के माध्यम से काफ्का की सदस्यता लें या अपने वेब हुक को कैसे सरल बनाएं