नमस्कार, हेब्र! इस पोस्ट में, हम JRE को एक काम कर रहे माइक्रोसैसवर्क एप्लिकेशन को स्थापित करने से, लिनक्स के तहत अपाचे काफ्का का उपयोग करके स्प्रिंग बूट 2 पर एक आवेदन लिखेंगे।

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

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



पोस्ट में दो खंड होंगे। पहले हम एक लिनक्स मशीन पर अपाचे काफ्का को कॉन्फ़िगर करते हैं और चलाते हैं, दूसरे में हम जावा में दो माइक्रोसिस्ट्रिस लिखते हैं।

स्टार्टअप में, जिसमें मैंने एक प्रोग्रामर के रूप में अपने पेशेवर करियर की शुरुआत की, काफ्का पर माइक्रोसर्विस थे, और मेरे एक माइक्रोसेवा ने भी काफ्का के माध्यम से दूसरों के साथ काम किया था, लेकिन मुझे नहीं पता था कि सर्वर ने कैसे काम किया, चाहे वह एक एप्लीकेशन के रूप में लिखा गया हो या यह पहले से ही पूरी तरह से बॉक्सिंग हो। उत्पाद। मेरा आश्चर्य और निराशा क्या था जब यह पता चला कि काफ्का अभी भी एक बॉक्सिंग उत्पाद था, और मेरा काम न केवल जावा (जो मुझे करना पसंद है) में एक क्लाइंट लिखना होगा, साथ ही साथ तैयार एप्लिकेशन को devOps के रूप में तैनात करना और कॉन्फ़िगर करना होगा (जो मैं) घृणा करना)। हालांकि, यहां तक ​​कि अगर मैं एक दिन से भी कम समय में इसे काफ्का वर्चुअल सर्वर पर उठा सकता था, तो यह वास्तव में ऐसा करने के लिए काफी सरल है। So.

हमारे आवेदन में निम्नलिखित बातचीत संरचना होगी:


पोस्ट के अंत में, हमेशा की तरह, वर्किंग कोड के साथ git के लिंक होंगे।

एक वर्चुअल मशीन पर Apache Kafka + Zookeeper तैनात करें

मैंने काफ्का को स्थानीय लिनक्स पर, खसखस ​​पर और दूरस्थ लिनक्स पर उठाने की कोशिश की। दो मामलों में (लिनक्स), मैं बहुत जल्दी सफल हुआ। खसखस के साथ अभी तक कुछ नहीं हुआ। इसलिए, हम लिनक्स पर काफ्का बढ़ाएंगे। मैंने उबंटू को 18.04 चुना।

काफ्का को काम करने के लिए उसे एक ज़ूकीपर की जरूरत है। ऐसा करने के लिए, आपको काफ्का लॉन्च करने से पहले इसे डाउनलोड और चलाना होगा।

So.

0. जेआरई स्थापित करें

यह निम्नलिखित आदेशों द्वारा किया जाता है:

sudo apt-get update sudo apt-get install default-jre 

यदि सब कुछ ठीक रहा, तो आप कमांड दर्ज कर सकते हैं

 java -version 

और सुनिश्चित करें कि जावा स्थापित है।

1. डाउनलोड Zookeeper

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

मैंने नीचे कमांड का इस्तेमाल किया। यदि आप अन्य लिनक्स कमांड जानते हैं, जो आपकी राय में, आपको यह अधिक नस्लीय रूप से सही ढंग से करने की अनुमति देगा, तो उनका उपयोग करें। मैं एक डेवलपर हूं, एक भक्त नहीं हूं, और मैं "बकरी ही" के स्तर पर सर्वर के साथ संवाद करता हूं। तो, आवेदन डाउनलोड करें:

 wget -P /home/xpendence/downloads/ "http://apache-mirror.rbc.ru/pub/apache/zookeeper/zookeeper-3.4.12/zookeeper-3.4.12.tar.gz" 

एप्लिकेशन को आपके द्वारा निर्दिष्ट फ़ोल्डर में डाउनलोड किया जाता है, मैंने उन सभी एप्लिकेशनों को डाउनलोड करने के लिए फ़ोल्डर / होम / xpendence / डाउनलोड बनाया है जिनकी मुझे आवश्यकता है।

2. अनपैक ज़ुकाइपर

मैंने कमांड का उपयोग किया:

 tar -xvzf /home/xpendence/downloads/zookeeper-3.4.12.tar.gz 

यह कमांड उस फ़ोल्डर में संग्रह को अनपैक करता है जिसमें आप स्थित हैं। फिर आपको एप्लिकेशन को / ऑप्ट / ज़ुकीपर को स्थानांतरित करने की आवश्यकता हो सकती है। और आप तुरंत इसमें जा सकते हैं और वहां से पहले से ही संग्रह को अनपैक कर सकते हैं।

3. सेटिंग्स संपादित करें

फ़ोल्डर में / zookeeper / conf / एक फ़ाइल zoo-sample.cfg है, मैं इसे zoo.conf में नाम बदलने का प्रस्ताव देता हूं, यह वह फाइल है जिसे JVM स्टार्टअप के लिए देखेगा। निम्नलिखित को इस फ़ाइल में अंत में जोड़ा जाना चाहिए:

 tickTime=2000 dataDir=/var/zookeeper clientPort=2181 

इसके अलावा, / var / zookeeper निर्देशिका बनाएं।

4. ज़ूकीपर लॉन्च करें

/ Opt / zookeeper फ़ोल्डर में जाएं और कमांड के साथ सर्वर शुरू करें:

 bin/zkServer.sh start 

"शुरू" दिखाई देना चाहिए।

जिसके बाद, मैं यह जांचने का प्रस्ताव करता हूं कि सर्वर काम कर रहा है। हम लिखते हैं:

 telnet localhost 2181 

एक संदेश दिखाई देना चाहिए कि कनेक्शन एक सफलता थी। यदि आपके पास एक कमजोर सर्वर है और संदेश दिखाई नहीं दिया, तो फिर से प्रयास करें - जब भी STARTED दिखाई देता है, तो एप्लिकेशन बहुत बाद में पोर्ट को सुनना शुरू करता है। जब मैंने कमजोर सर्वर पर यह सब करने की कोशिश की, तो यह हर बार मेरे साथ हुआ। यदि सब कुछ जुड़ा हुआ है, तो कमांड दर्ज करें

 ruok 

इसका क्या मतलब है: "क्या आप ठीक हैं?" सर्वर को जवाब देना चाहिए:

 imok ( !) 

और डिस्कनेक्ट करें। तो, सब कुछ योजना के अनुसार है। हम अपाचे काफ्का लॉन्च करने के लिए आगे बढ़ें।

5. काफ्का के तहत एक उपयोगकर्ता बनाएं

काफ्का के साथ काम करने के लिए हमें एक अलग उपयोगकर्ता की आवश्यकता है।

 sudo adduser --system --no-create-home --disabled-password --disabled-login kafka 

6. अपाचे काफ्का डाउनलोड करें

दो वितरण हैं - द्विआधारी और स्रोत। हमें एक बाइनरी की आवश्यकता है। उपस्थिति में, बाइनरी के साथ संग्रह आकार में अलग है। बाइनरी का वजन 59 एमबी, और 6.5 एमबी का वजन है।

नीचे दी गई लिंक का उपयोग करके, बाइनरी को डाउनलोड करें

 wget -P /home/xpendence/downloads/ "http://mirror.linux-ia64.org/apache/kafka/2.1.0/kafka_2.11-2.1.0.tgz" 

7. अपाचे काफ्का

अनपैकिंग प्रक्रिया ज़ुकाइपर के लिए समान से अलग नहीं है। हम आर्क को / ऑप्ट डायरेक्टरी में भी अनपैक करते हैं और इसे काकाका का नाम देते हैं ताकि / बिन फोल्डर का रास्ता / ऑप्ट / काफ्का / बिन हो जाए

 tar -xvzf /home/xpendence/downloads/kafka_2.11-2.1.0.tgz 

8. सेटिंग्स संपादित करें

सेटिंग्स में / अपनाने /kafka/config/server.properties हैं। एक पंक्ति जोड़ें:

 delete.topic.enable = true 

यह सेटिंग वैकल्पिक प्रतीत होती है, यह इसके बिना काम करती है। यह सेटिंग आपको विषयों को हटाने की अनुमति देती है। अन्यथा, आप बस कमांड लाइन के माध्यम से विषयों को हटा नहीं सकते हैं।

9. हम उपयोगकर्ता काफ्का निर्देशिका काफ्का को एक्सेस देते हैं

 chown -R kafka:nogroup /opt/kafka chown -R kafka:nogroup /var/lib/kafka 

10. अपाचे काफ्का का लंबे समय से प्रतीक्षित लॉन्च

हम कमांड दर्ज करते हैं, जिसके बाद काफ्का शुरू होना चाहिए:

 /opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties 

यदि सामान्य क्रियाओं (काफ्का जावा और स्काला में लिखी गई है) लॉग में खत्म नहीं हुई, तो सब कुछ काम कर गया और आप हमारी सेवा का परीक्षण कर सकते हैं।

10.1। कमजोर सर्वर समस्याएँ

अपाचे काफ्का पर प्रयोगों के लिए, मैंने एक कोर और 512 एमबी रैम (सिर्फ 99 रूबल के लिए) के साथ एक कमजोर सर्वर लिया, जो मेरे लिए कई समस्याएं बन गया।

स्मृति से बाहर। बेशक, आप 512 एमबी के साथ ओवरक्लॉक नहीं कर सकते हैं, और सर्वर मेमोरी की कमी के कारण काफ्का को तैनात नहीं कर सकता है। तथ्य यह है कि डिफ़ॉल्ट रूप से, काफ्का 1 जीबी मेमोरी की खपत करता है। कोई आश्चर्य नहीं कि वह गायब था :)

हम kafka-server-start.sh, zookeeper-server-start.sh पर जाते हैं। स्मृति को नियंत्रित करने वाली एक पंक्ति पहले से ही है:

 export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G" 

इसे इसमें बदलें:

 export KAFKA_HEAP_OPTS="-Xmx256M -Xms128M" 

यह काफ्का की भूख को कम करेगा और आपको सर्वर शुरू करने की अनुमति देगा।

एक कमजोर कंप्यूटर के साथ दूसरी समस्या ज़ुकाइपर से कनेक्ट करने के लिए समय की कमी है। डिफ़ॉल्ट रूप से, यह 6 सेकंड दिया जाता है। यदि लोहा कमजोर है, तो यह, निश्चित रूप से, पर्याप्त नहीं है। Server.properties में हम zukipper को कनेक्शन समय बढ़ाते हैं:

 zookeeper.connection.timeout.ms=30000 

मैंने आधा मिनट सेट किया।

11. काफ्का-सर्वर का परीक्षण करें

ऐसा करने के लिए, हम दो टर्मिनल खोलेंगे, एक पर हम निर्माता को लॉन्च करेंगे, दूसरे पर - उपभोक्ता।
पहले कंसोल में, एक पंक्ति दर्ज करें:

 /opt/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test /opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test 

यह आइकन प्रदर्शित होना चाहिए, यह दर्शाता है कि निर्माता स्पैम संदेशों के लिए तैयार है:

 > 

दूसरे कंसोल में, कमांड दर्ज करें:

 /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning 

अब, निर्माता कंसोल में टाइप करते समय, जब आप Enter दबाते हैं, तो यह उपभोक्ता कंसोल में दिखाई देगा।



यदि आप स्क्रीन पर लगभग मेरे जैसे ही दिखते हैं - बधाई, सबसे खराब खत्म हो गया है!


अब हमें बस स्प्रिंग बूट पर कुछ ग्राहकों को लिखना है जो अपाचे काफ्का के माध्यम से एक दूसरे के साथ संवाद करेंगे।

स्प्रिंग बूट पर एक एप्लीकेशन लिखना

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

काफ्का-सर्वर

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

कफ़्का-क्लाइंट में दो घटक होते हैं - निर्माता (वह कफ़्का-सर्वर को संदेश भेजता है) और उपभोक्ता (वह कफ़्का-सर्वर को सुनता है और वहाँ से सब्स्क्राइब किए गए विषयों पर नए संदेश लेता है)। हमारा काम दोनों घटकों को लिखना और उन्हें काम करना है।

उपभोक्ता:

 @Configuration public class KafkaConsumerConfig { @Value("${kafka.server}") private String kafkaServer; @Value("${kafka.group.id}") private String kafkaGroupId; @Bean public KafkaListenerContainerFactory<?> batchFactory() { ConcurrentKafkaListenerContainerFactory<Long, AbstractDto> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); factory.setBatchListener(true); factory.setMessageConverter(new BatchMessagingMessageConverter(converter())); return factory; } @Bean public KafkaListenerContainerFactory<?> singleFactory() { ConcurrentKafkaListenerContainerFactory<Long, AbstractDto> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); factory.setBatchListener(false); factory.setMessageConverter(new StringJsonMessageConverter()); return factory; } @Bean public ConsumerFactory<Long, AbstractDto> consumerFactory() { return new DefaultKafkaConsumerFactory<>(consumerConfigs()); } @Bean public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory() { return new ConcurrentKafkaListenerContainerFactory<>(); } @Bean public Map<String, Object> consumerConfigs() { Map<String, Object> props = new HashMap<>(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaGroupId); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); return props; } @Bean public StringJsonMessageConverter converter() { return new StringJsonMessageConverter(); } } 

हमें kafka.properties से स्थिर डेटा के साथ आरंभिक 2 फ़ील्ड चाहिए।

 kafka.server=localhost:9092 kafka.group.id=server.broadcast 

kafka.server वह पता है जिस पर हमारा सर्वर लटका है, इस मामले में, स्थानीय। डिफ़ॉल्ट रूप से, काफ्का पोर्ट 9092 पर सुनता है।

kafka.group.id उपभोक्ताओं का एक समूह है, जिसके भीतर संदेश का एक उदाहरण दिया जाता है। उदाहरण के लिए, आपके पास एक समूह में तीन कोरियर हैं, और वे सभी एक ही विषय को सुनते हैं। जैसे ही इस विषय के साथ सर्वर पर एक नया संदेश दिखाई देता है, उसे समूह में किसी को दिया जाता है। शेष दो उपभोक्ताओं को संदेश प्राप्त नहीं हो रहा है।

अगला, हम उपभोक्ताओं के लिए एक कारखाना बना रहे हैं - ConsumerFactory।

  @Bean public ConsumerFactory<Long, AbstractDto> consumerFactory() { return new DefaultKafkaConsumerFactory<>(consumerConfigs()); } 

हमें जिन गुणों की आवश्यकता है, उनके साथ आरंभ में, यह भविष्य में उपभोक्ताओं के लिए एक मानक कारखाने के रूप में काम करेगा।

  @Bean public Map<String, Object> consumerConfigs() { Map<String, Object> props = new HashMap<>(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaGroupId); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); return props; } 

ConsumerConfigs सिर्फ मैप कॉन्‍फ़िगर हैं। हम सर्वर का पता, समूह और डिसएरिलाइज़र प्रदान करते हैं।

इसके अलावा, एक उपभोक्ता के लिए सबसे महत्वपूर्ण बिंदुओं में से एक। उपभोक्ता एकल ऑब्जेक्ट और संग्रह दोनों प्राप्त कर सकता है - उदाहरण के लिए, StarshipDto और List दोनों। और अगर हमें JSON के रूप में StarshipDto मिलता है, तो हमें JSON सरणी के रूप में, मोटे तौर पर बोलकर सूची मिलती है। इसलिए, हमारे पास कम से कम दो संदेश कारखाने हैं - एकल संदेशों के लिए और सरणियों के लिए।

  @Bean public KafkaListenerContainerFactory<?> singleFactory() { ConcurrentKafkaListenerContainerFactory<Long, AbstractDto> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); factory.setBatchListener(false); factory.setMessageConverter(new StringJsonMessageConverter()); return factory; } 

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

जब हम संकुल / सरणियों (बैच) के लिए एक कारखाना बनाते हैं, तो मुख्य अंतर (इस तथ्य के अलावा कि हम चिह्नित करते हैं कि हम पैकेज सुन रहे हैं) यह है कि हम एक विशेष पैकेज कनवर्टर के रूप में निर्दिष्ट करते हैं जो पैकेजों को परिवर्तित करेगा। JSON स्ट्रिंग्स से।

  @Bean public KafkaListenerContainerFactory<?> batchFactory() { ConcurrentKafkaListenerContainerFactory<Long, AbstractDto> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); factory.setBatchListener(true); factory.setMessageConverter(new BatchMessagingMessageConverter(converter())); return factory; } @Bean public StringJsonMessageConverter converter() { return new StringJsonMessageConverter(); } 

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

  @Bean public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory() { return new ConcurrentKafkaListenerContainerFactory<>(); } 

उपभोक्ता स्थापित है। हम निर्माता के पास जाते हैं।

 @Configuration public class KafkaProducerConfig { @Value("${kafka.server}") private String kafkaServer; @Value("${kafka.producer.id}") private String kafkaProducerId; @Bean public Map<String, Object> producerConfigs() { Map<String, Object> props = new HashMap<>(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); props.put(ProducerConfig.CLIENT_ID_CONFIG, kafkaProducerId); return props; } @Bean public ProducerFactory<Long, StarshipDto> producerStarshipFactory() { return new DefaultKafkaProducerFactory<>(producerConfigs()); } @Bean public KafkaTemplate<Long, StarshipDto> kafkaTemplate() { KafkaTemplate<Long, StarshipDto> template = new KafkaTemplate<>(producerStarshipFactory()); template.setMessageConverter(new StringJsonMessageConverter()); return template; } } 

स्टैटिक वैरिएबल में से हमें काफ्का सर्वर और प्रोड्यूसर आईडी का पता चाहिए। वह कुछ भी हो सकता है।

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

टेम्प्लेट में, हम चिह्नित करते हैं कि हम JSON में ऑब्जेक्ट्स को अनुक्रमित करेंगे, और यह, शायद, पर्याप्त है।

हमारे पास एक उपभोक्ता और एक निर्माता है, यह एक सेवा लिखने के लिए बना हुआ है जो संदेश भेजेगा और उन्हें प्राप्त करेगा।

 @Service @Slf4j public class StarshipServiceImpl implements StarshipService { private final KafkaTemplate<Long, StarshipDto> kafkaStarshipTemplate; private final ObjectMapper objectMapper; @Autowired public StarshipServiceImpl(KafkaTemplate<Long, StarshipDto> kafkaStarshipTemplate, ObjectMapper objectMapper) { this.kafkaStarshipTemplate = kafkaStarshipTemplate; this.objectMapper = objectMapper; } @Override public void send(StarshipDto dto) { kafkaStarshipTemplate.send("server.starship", dto); } @Override @KafkaListener(id = "Starship", topics = {"server.starship"}, containerFactory = "singleFactory") public void consume(StarshipDto dto) { log.info("=> consumed {}", writeValueAsString(dto)); } private String writeValueAsString(StarshipDto dto) { try { return objectMapper.writeValueAsString(dto); } catch (JsonProcessingException e) { e.printStackTrace(); throw new RuntimeException("Writing value to JSON failed: " + dto.toString()); } } } 

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

  private final KafkaTemplate<Long, StarshipDto> kafkaStarshipTemplate; 

निर्माता विधि:

  @Override public void send(StarshipDto dto) { kafkaStarshipTemplate.send("server.starship", dto); } 

सर्वर को एक संदेश भेजने के लिए सभी आवश्यक है कि टेम्पलेट पर भेजें विधि को कॉल करें और विषय (विषय) और हमारी वस्तु को वहां स्थानांतरित करें। ऑब्जेक्ट को JSON में क्रमबद्ध किया जाएगा और निर्दिष्ट विषय के तहत सर्वर पर उड़ान भरेगा।

सुनने की विधि इस प्रकार है:

  @Override @KafkaListener(id = "Starship", topics = {"server.starship"}, containerFactory = "singleFactory") public void consume(StarshipDto dto) { log.info("=> consumed {}", writeValueAsString(dto)); } 

हम इस विधि को @KafkaListener एनोटेशन के साथ चिह्नित करते हैं, जहां हम किसी भी आईडी को इंगित करते हैं जिसे हम पसंद करते हैं, सुने हुए विषय और एक कारखाना जो हमें प्राप्त संदेश को हमारी आवश्यकता के अनुसार परिवर्तित कर देगा। इस मामले में, चूंकि हम किसी एक वस्तु को स्वीकार करते हैं, इसलिए हमें एक एकल कार्य की आवश्यकता है। सूची के लिए <?>, बैच निर्दिष्ट करें। परिणामस्वरूप, हम ऑब्जेक्ट को भेजने की विधि का उपयोग करके काफ्का सर्वर को भेजते हैं और उपभोग विधि का उपयोग करके इसे प्राप्त करते हैं।

आप 5 मिनट में एक परीक्षण लिख सकते हैं जो काफ्का की पूरी ताकत का प्रदर्शन करेगा, लेकिन हम आगे बढ़ेंगे - 10 मिनट बिताएंगे और एक और एप्लिकेशन लिखेंगे जो सर्वर को संदेश भेजेगा कि हमारा पहला आवेदन सुनेंगे।

काफ्का-परीक्षक

पहला आवेदन लिखने का अनुभव होने के बाद, हम आसानी से दूसरा लिख ​​सकते हैं, खासकर यदि हम पेस्ट और डोटो पैकेज की नकल करते हैं, केवल निर्माता को पंजीकृत करें (हम केवल संदेश भेजेंगे) और सेवा में एकमात्र भेजने की विधि जोड़ें। नीचे दिए गए लिंक का उपयोग करके, आप आसानी से प्रोजेक्ट कोड डाउनलोड कर सकते हैं और सुनिश्चित कर सकते हैं कि वहाँ कुछ भी जटिल नहीं है।

  @Scheduled(initialDelay = 10000, fixedDelay = 5000) @Override public void produce() { StarshipDto dto = createDto(); log.info("<= sending {}", writeValueAsString(dto)); kafkaStarshipTemplate.send("server.starship", dto); } private StarshipDto createDto() { return new StarshipDto("Starship " + (LocalTime.now().toNanoOfDay() / 1000000)); } 

पहले 10 सेकंड के बाद, काफ्का-परीक्षक हर 5 सेकंड (चित्र क्लिक करने योग्य है) को काफ्का सर्वर को स्टारशिप के नाम के साथ संदेश भेजना शुरू कर देता है।



वहां, उन्हें काफ्का-सर्वर द्वारा सुनी और प्राप्त की जाती है (चित्र भी क्लिक करने योग्य है)।



मुझे उम्मीद है कि जो लोग काफ्का पर माइक्रोसर्विसेज लिखना शुरू करने का सपना देखते हैं, वे उतने ही आसानी से सफल होंगे जितना मैंने किया था। और यहाँ परियोजनाओं के लिए लिंक हैं:

→ कफ़्का-सर्वर
→ काफ्का-परीक्षक