Apache Kafka का उपयोग करते हुए सिंक्रोनस अनुरोध-प्रतिक्रिया

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

@Middle_java द्वारा अनुवादित

मूल लेख दिनांक: २६ अक्टूबर २०१ October

अपाचे काफ्का स्वाभाविक रूप से अतुल्यकालिक है। इसलिए, अपाचे काफ्का के लिए अनुरोध-प्रतिक्रिया शब्दार्थ प्राकृतिक नहीं हैं। हालांकि, यह चुनौती नई नहीं है। एंटरप्राइज़ एकीकरण पैटर्न अनुरोध-उत्तर अतुल्यकालिक चैनलों पर तुल्यकालिक संदेश के लिए एक सिद्ध तंत्र प्रदान करता है:



रिटर्न एड्रेस पैटर्न रिक्वेस्ट के लिए एक अनुरोध के साथ एक मैकेनिज्म के साथ रिक्वेस्ट-रिप्लाई पैटर्न का अनुपालन करता है ताकि पता भेजा जा सके।



हाल ही में, स्प्रिंग काफ्का 2.1.3 ने "रिक्वेस्ट रिप्लाई" पैटर्न बॉक्स से समर्थन जोड़ा, और संस्करण 2.2 में इसके कुछ खुरदरेपन को पॉलिश किया गया। आइए देखें कि यह समर्थन कैसे काम करता है:

क्लाइंट साइड: उत्तर देना KafkaTemplate

टेम्प्लेट का सुविख्यात अमूर्त स्प्रिंग में अनुरोध-उत्तर तंत्र के क्लाइंट भाग के लिए आधार बनाता है।

@Bean public ReplyingKafkaTemplate < String, Request, Reply > replyKafkaTemplate( ProducerFactory < String, Request > pf, KafkaMessageListenerContainer < String, Reply > lc) { return new ReplyingKafkaTemplate < > (pf, lc); } 

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

  @Value("${kafka.topic.car.reply}") private String replyTopic; @Bean public Map < String, Object > consumerConfigs() { Map < String, Object > props = new HashMap < > (); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class); props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); return props; } @Bean public Map < String, Object > producerConfigs() { Map < String, Object > props = new HashMap < > (); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); return props; } @Bean public ProducerFactory < String, Request > requestProducerFactory() { return new DefaultKafkaProducerFactory < > (producerConfigs()); } @Bean public ConsumerFactory < String, Reply > replyConsumerFactory() { return new DefaultKafkaConsumerFactory < > (consumerConfigs(), new StringDeserializer(), new JsonSerializer < Reply > ()); } @Bean public KafkaMessageListenerContainer < String, Reply > replyListenerContainer() { ContainerProperties containerProperties = new ContainerProperties(replyTopic); return new KafkaMessageListenerContainer < > (replyConsumerFactory(), containerProperties); } 

इस मामले में, तुल्यकालिक अनुरोध भेजने और प्रतिक्रिया प्राप्त करने के लिए replKafkaTemplate का उपयोग इस प्रकार है:

 @Value("${kafka.topic.car.request}") private String requestTopic; @Value("${kafka.topic.car.reply}") private String replyTopic; @Autowired private ReplyingKafkaTemplate < String, Request, Reply > requestReplyKafkaTemplate; ... RequestReply request = RequestReply.request(...); // producer record ProducerRecord < String, Request > record = new ProducerRecord < String, Request > (requestTopic, request); //       record.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, requestReplyTopic.getBytes())); //     Kafka          RequestReplyFuture < String, Request, Reply > sendAndReceive = requestReplyKafkaTemplate.sendAndReceive(record); sendAndReceive.addCallback(new ListenableFutureCallback < ConsumerRecord < String, Reply >> () { @Override public void onSuccess(ConsumerRecord < String, Reply > result) { //   consumer record Reply reply = result.value(); System.out.println("Reply: " + reply.toString()); } }); 

बहुत सारे बॉयलरप्लेट और एक निम्न-स्तरीय एपीआई भी है, और यहां तक ​​कि आधुनिक कम्पेटिबल सिवनी के बजाय यह अप्रचलित श्रवणयोग्य सीवन एपीआई भी है।

requestReplyKafkaTemplate , KafkaHeaders.CORRELATION_ID शीर्षक बनाने और सेट करने का ध्यान रखता है, लेकिन हमें अनुरोध के लिए स्पष्ट रूप से KafkaHeaders.REPLY_TOPIC शीर्षक सेट करना होगा। कृपया यह भी ध्यान दें कि उत्तर के लिए एक ही विषय उत्तर में बहुत अधिक अनजाने में था । कुछ छी। नहीं काफी मैं वसंत अमूर्त से क्या उम्मीद थी।

सर्वर साइड: @SendTo

सर्वर की ओर से, अनुरोध के लिए विषय पर सुनने वाले सामान्य काफ़्कालिस्टनर को प्रतिक्रिया संदेश प्रदान करने के लिए @SendTo एनोटेशन के साथ अतिरिक्त रूप से सजाया जाता है। श्रोता विधि द्वारा लौटाई गई वस्तु स्वचालित रूप से प्रतिक्रिया संदेश में लपेटी जाती है, CORRELATION_ID को जोड़ा जाता है , और प्रतिक्रिया REPLY_TOPIC शीर्ष लेख में निर्दिष्ट विषय में प्रकाशित होती है।

  @Bean public Map < String, Object > consumerConfigs() { Map < String, Object > props = new HashMap < > (); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonSerializer.class); props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); return props; } @Bean public Map < String, Object > producerConfigs() { Map < String, Object > props = new HashMap < > (); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); return props; } @Bean public ConsumerFactory < String, Request > requestConsumerFactory() { return new DefaultKafkaConsumerFactory < > (consumerConfigs(), new StringDeserializer(), new JsonSerializer < Request > ()); } @Bean public KafkaListenerContainerFactory < ConcurrentMessageListenerContainer < String, Request >> requestListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory < String, Request > factory = new ConcurrentKafkaListenerContainerFactory < > (); factory.setConsumerFactory(requestConsumerFactory()); factory.setReplyTemplate(replyTemplate()); return factory; } @Bean public ProducerFactory < String, Reply > replyProducerFactory() { return new DefaultKafkaProducerFactory < > (producerConfigs()); } @Bean public KafkaTemplate < String, Reply > replyTemplate() { return new KafkaTemplate < > (replyProducerFactory()); } 

यहाँ कुछ विन्यास भी आवश्यक है, लेकिन श्रोता विन्यास सरल है:

  @KafkaListener(topics = "${kafka.topic.car.request}", containerFactory = "requestListenerContainerFactory") @SendTo() public Reply receive(Request request) { Reply reply = ...; return reply; } 

लेकिन उपभोक्ता के कई उदाहरणों के बारे में क्या?

जब तक हम उपभोक्ता के कई उदाहरणों का उपयोग करते हैं, तब तक सब कुछ काम करता है। यदि हमारे पास कई ग्राहक उदाहरण हैं, तो हमें यह सुनिश्चित करने की आवश्यकता है कि प्रतिक्रिया सही ग्राहक उदाहरण के लिए भेजी जाए। स्प्रिंग काफ्का प्रलेखन मानता है कि प्रत्येक उपभोक्ता एक अद्वितीय विषय का उपयोग कर सकता है या कि एक अतिरिक्त काफ्काहेडर्स हेडर मूल्य अनुरोध के साथ भेजा जाता है। RESPONSE_PARTITION एक चार-बाइट फ़ील्ड है जिसमें पूर्णांक अनुभाग संख्या का BIG-ENDRIT प्रतिनिधित्व है। विभिन्न ग्राहकों के लिए अलग-अलग विषयों का उपयोग करना स्पष्ट रूप से बहुत लचीला नहीं है, इसलिए हम स्पष्ट REPLY_PARTITION सेटिंग चुनते हैं। फिर क्लाइंट को पता होना चाहिए कि उसे किस पार्टीशन को सौंपा गया है। प्रलेखन एक विशेष विभाजन का चयन करने के लिए एक स्पष्ट विन्यास का उपयोग करने का सुझाव देता है। आइए इसे हमारे उदाहरण में जोड़ें:

  @Value("${kafka.topic.car.reply.partition}") private int replyPartition; ... @Bean public KafkaMessageListenerContainer < String, RequestReply > replyListenerContainer() { ContainerProperties containerProperties = new ContainerProperties(replyTopic); TopicPartitionInitialOffset initialOffset = new TopicPartitionInitialOffset(replyTopic, replyPartition); return new KafkaMessageListenerContainer < > (replyConsumerFactory(), containerProperties, initialOffset); } private static byte[] intToBytesBigEndian(final int data) { return new byte[] { (byte)((data >> 24) & 0xff), (byte)((data >> 16) & 0xff), (byte)((data >> 8) & 0xff), (byte)((data >> 0) & 0xff), }; } ... record.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, requestReplyTopic.getBytes())); record.headers().add(new RecordHeader(KafkaHeaders.REPLY_PARTITION, intToBytesBigEndian(replyPartition))); RequestReplyFuture < String, RequestReply, RequestReply > sendAndReceive = requestReplyKafkaTemplate.sendAndReceive(record); ... 

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

प्रतिक्रिया और विभाजन के लिए विषय प्रसंस्करण encapsulating

चलो रिटर्न एड्रेस पैटर्न को एनकैप्सुलेट करके शुरू करें, प्रतिक्रिया और विभाजन के लिए विषय के साथ गुजरना। प्रतिक्रिया के लिए विषय को RequestReplyTemplate में इंजेक्ट किया जाना चाहिए और इसलिए, एपीआई में बिल्कुल भी मौजूद नहीं होना चाहिए। जब किसी उत्तर के लिए विभाजन की बात आती है, तो हम इसके विपरीत करेंगे: उत्तर के लिए विषय श्रोता को सौंपे गए विभाजन को निकालें, और इस विभाजन को स्वतः स्थानांतरित करें। इससे क्लाइंट को इन हेडर की देखभाल करने की आवश्यकता समाप्त हो जाती है।
उसी समय, आइए मानक KafkaTemplate की तरह एपीआई को भी देखें ( सरलीकृत मापदंडों के साथ SendAndReceive () विधि को ओवरलोड करें और डिफ़ॉल्ट विषय का उपयोग करके संबंधित अतिभारित तरीकों को जोड़ें):

 public class PartitionAwareReplyingKafkaTemplate < K, V, R > extends ReplyingKafkaTemplate < K, V, R > { public PartitionAwareReplyingKafkaTemplate(ProducerFactory < K, V > producerFactory, GenericMessageListenerContainer < K, R > replyContainer) { super(producerFactory, replyContainer); } private TopicPartition getFirstAssignedReplyTopicPartition() { if (getAssignedReplyTopicPartitions() != null && getAssignedReplyTopicPartitions().iterator().hasNext()) { TopicPartition replyPartition = getAssignedReplyTopicPartitions().iterator().next(); if (this.logger.isDebugEnabled()) { this.logger.debug("Using partition " + replyPartition.partition()); } return replyPartition; } else { throw new KafkaException("Illegal state: No reply partition is assigned to this instance"); } } private static byte[] intToBytesBigEndian(final int data) { return new byte[] { (byte)((data >> 24) & 0xff), (byte)((data >> 16) & 0xff), (byte)((data >> 8) & 0xff), (byte)((data >> 0) & 0xff), }; } public RequestReplyFuture < K, V, R > sendAndReceiveDefault(@Nullable V data) { return sendAndReceive(getDefaultTopic(), data); } public RequestReplyFuture < K, V, R > sendAndReceiveDefault(K key, @Nullable V data) { return sendAndReceive(getDefaultTopic(), key, data); } ... public RequestReplyFuture < K, V, R > sendAndReceive(String topic, @Nullable V data) { ProducerRecord < K, V > record = new ProducerRecord < > (topic, data); return doSendAndReceive(record); } public RequestReplyFuture < K, V, R > sendAndReceive(String topic, K key, @Nullable V data) { ProducerRecord < K, V > record = new ProducerRecord < > (topic, key, data); return doSendAndReceive(record); } ... @Override public RequestReplyFuture < K, V, R > sendAndReceive(ProducerRecord < K, V > record) { return doSendAndReceive(record); } protected RequestReplyFuture < K, V, R > doSendAndReceive(ProducerRecord < K, V > record) { TopicPartition replyPartition = getFirstAssignedReplyTopicPartition(); record.headers() .add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, replyPartition.topic().getBytes())) .add(new RecordHeader(KafkaHeaders.REPLY_PARTITION, intToBytesBigEndian(replyPartition.partition()))); return super.sendAndReceive(record); } } 

अगला चरण: अधिक आधुनिक कंप्लीटेबल फ़ॉरेस्ट के लिए सुने जाने योग्य सिवनी को अपनाना।

 public class CompletableFutureReplyingKafkaTemplate < K, V, R > extends PartitionAwareReplyingKafkaTemplate < K, V, R > { public CompletableFutureReplyingKafkaTemplate(ProducerFactory < K, V > producerFactory, GenericMessageListenerContainer < K, R > replyContainer) { super(producerFactory, replyContainer); } public CompletableFuture < R > requestReplyDefault(V value) { return adapt(sendAndReceiveDefault(value)); } public CompletableFuture < R > requestReplyDefault(K key, V value) { return adapt(sendAndReceiveDefault(key, value)); } ... public CompletableFuture < R > requestReply(String topic, V value) { return adapt(sendAndReceive(topic, value)); } public CompletableFuture < R > requestReply(String topic, K key, V value) { return adapt(sendAndReceive(topic, key, value)); } ... private CompletableFuture < R > adapt(RequestReplyFuture < K, V, R > requestReplyFuture) { CompletableFuture < R > completableResult = new CompletableFuture < R > () { @Override public boolean cancel(boolean mayInterruptIfRunning) { boolean result = requestReplyFuture.cancel(mayInterruptIfRunning); super.cancel(mayInterruptIfRunning); return result; } }; //       requestReplyFuture.getSendFuture().addCallback(new ListenableFutureCallback < SendResult < K, V >> () { @Override public void onSuccess(SendResult < K, V > sendResult) { // NOOP } @Override public void onFailure(Throwable t) { completableResult.completeExceptionally(t); } }); //     requestReplyFuture.addCallback(new ListenableFutureCallback < ConsumerRecord < K, R >> () { @Override public void onSuccess(ConsumerRecord < K, R > result) { completableResult.complete(result.value()); } @Override public void onFailure(Throwable t) { completableResult.completeExceptionally(t); } }); return completableResult; } } 

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

  @Autowired private CompletableFutureReplyingKafkaTemplate < String, Request, Reply > requestReplyKafkaTemplate; ... requestReplyKafkaTemplate.requestReply(request).thenAccept(reply - > System.out.println("Reply: " + reply.toString()); ); 

संक्षेप में देना

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

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

चेतावनी 2:
यदि एक सिंक्रोनस रिक्वेस्ट-रिस्पांस की आवश्यकता होती है, तो अपाचे काफ्का जैसे एसिंक्रोनस चैनल का उपयोग करने की तुलना में HTTP- आधारित प्रोटोकॉल बहुत सरल और अधिक कुशल है।
हालांकि, ऐसे परिदृश्य हो सकते हैं जहां काफ्का के माध्यम से एक तुल्यकालिक अनुरोध-प्रतिक्रिया समझ में आती है। समझदारी से काम के लिए सबसे अच्छा उपकरण चुनें।

पूरी तरह से काम करने वाला उदाहरण github.com/callistaenterprise/blog-synchronous-kafka पर पाया जा सकता है।

टिप्पणियाँ

फेडेरिको • 7 महीने पहले
और जब हमारे पास हाइब्रिड ज़रूरतें होती हैं, उदाहरण के लिए, 50% मामलों में हमें एक रिक्वेस्ट-रिस्पांस की आवश्यकता होती है और 50% में हमें इवेंट मैनेजमेंट की आवश्यकता होती है? हम यह कैसे करते हैं? स्प्रिंग काफ्का द्वारा आवश्यक विन्यास बहुत भयानक लग रहा है।

जहानज़ेब कय्यूम • 6 महीने पहले
वसंत में अब प्रतिक्रिया के लिए एक सामान्य विषय में विभाजन का उपयोग करके डिफ़ॉल्ट समर्थन है।

संस्करण 2.2 से शुरू, टेम्पलेट कॉन्फ़िगर किए गए प्रतिक्रिया कंटेनर (उत्तर कंटेनर) से प्रतिक्रिया या विभाजन के लिए विषय को निर्धारित करने की कोशिश करता है।

https://docs.spring.io/spring-kafka/reference/html/#replying-template

nir rozenberg • 8 महीने पहले
हैलो,
अंतिम पैराग्राफ में, आपने लिखा है कि ऐसे परिदृश्य हो सकते हैं जब काफ्का के माध्यम से एक तुल्यकालिक अनुरोध-प्रतिक्रिया HTTP की तुलना में समझ में आता है।
क्या आप ऐसे परिदृश्यों का उदाहरण दे सकते हैं?
आपका धन्यवाद
निर

Janne Keskitalo nir rozenberg • 8 महीने पहले
हम एक बड़ी मात्रा में लेनदेन प्रसंस्करण प्रणाली को कई माइक्रोसर्विसेज में विभाजित करने जा रहे हैं, और मुझे एक समान प्रसंस्करण आत्मीयता प्राप्त करने के लिए काफ्का के रिक्वेस्ट-रिस्पॉन्स मैसेजिंग का उपयोग करने का विचार है। मूल रूप से, काफ्का का उपयोग एक कॉल से एक ही लेनदेन प्रोसेसर प्रक्रिया के लिए सभी क्लाइंट को रूट करने के लिए किया जाता है, जो तब एक समय में क्रमिक रूप से उन्हें निष्पादित करता है। इस प्रकार का प्रसंस्करण रैखिकता को सुनिश्चित करता है ( https://stackoverflow.com/a/19515375/7430325 ), कारण कनेक्टिविटी और कुशल कैशिंग की भी अनुमति देता है। अनिवार्य रूप से, समन्वय प्रयासों को डेटाबेस से काफ्का में स्थानांतरित किया जाएगा, और हम डेटाबेस को धारावाहिक परिवर्तनशील अलगाव मोड में शुरू कर सकते हैं।
मुझे अभी तक हमारे लेन-देन शब्दार्थों के विवरण में यह देखना है कि यह कहां कम हुआ है, इसलिए यह केवल एक विचार है।

@Middle_java द्वारा अनुवादित