स्प्रिंग क्लाउड स्ट्रीम कफका और डाटाबेस प्रति सेवा

Question

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

एक उदाहरण:

Order service -------> | Kafka |------->Payment Service 
     |          | 
Orders MariaDB DB     Payment MariaDB Database 

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

निजी ऑर्डर बिजनेस ऑर्डर बिजनेस;

@PostMapping 
public Order createOrder(@RequestBody Order order){ 
    logger.debug("createOrder()"); 
    //a.- Save the order in the DB 
    orderBusiness.createOrder(order); 
    //b. Publish in the topic so that Payment Service charges for the item. 
    try{ 
     orderSource.output().send(MessageBuilder.withPayload(order).build()); 
    }catch(Exception e){ 
     logger.error("{}", e); 
    } 
    return order; 
} 

ये मेरी संदेह कर रहे हैं:

1. कदम (आदेश डीबी में बचाने के लिए) ए और b.- (संदेश प्रकाशित), एक सौदे में किया जाना चाहिए atomically। मैं उसे कैसे प्राप्त कर सकता हूं?
2. यह पिछले एक से संबंधित है: मैं संदेश भेजता हूं: orderSource.output()। भेजें (MessageBuilder.withPayload (ऑर्डर) .build()); यह ऑपरेशन अतुल्यकालिक है और हमेशा सत्य लौटाता है, भले ही काफ्का ब्रोकर नीचे न हो। मुझे कैसे पता चलेगा कि संदेश काफ्का ब्रोकर तक पहुंच गया है?

Answer 1

कदम ए (आदेश डीबी में बचाने के लिए) और b.- (संदेश प्रकाशित) atomically, एक सौदे में किया जाना चाहिए। मैं उसे कैसे प्राप्त कर सकता हूं?

काफ्का वर्तमान में लेनदेन का समर्थन नहीं करता है (और इस प्रकार कोई रोलबैक या प्रतिबद्ध नहीं), जिसे आपको इस तरह कुछ सिंक्रनाइज़ करने की आवश्यकता होगी। तो संक्षेप में: आप वह नहीं कर सकते जो आप करना चाहते हैं। यह निकट-आश भविष्य में बदल जाएगा, जब KIP-98 विलय हो गया है, लेकिन इसमें कुछ समय लग सकता है। इसके अलावा, कफका में लेन-देन के साथ भी, दो प्रणालियों में एक परमाणु लेनदेन करना बहुत कठिन काम है, जो कुछ भी है, वह केवल कफका में लेनदेन के समर्थन से बेहतर होगा, यह अभी भी पूरी तरह से आपके मुद्दे को हल नहीं करेगा। इसके लिए आपको अपने सिस्टम में two phase commit के कुछ रूपों को लागू करने की आवश्यकता होगी।

आप कुछ हद तक निर्माता गुण कॉन्फ़िगर करके पास प्राप्त कर सकते हैं, लेकिन अंत में आप अपने सिस्टम (MariaDB या काफ्का) में से एक के लिए कम से कम एक बार या में अधिकतम एक बार के बीच चुना करना होगा।

चलो शुरू करें कि आप कफका में क्या कर सकते हैं, एक संदेश की डिलीवरी सुनिश्चित करें और आगे नीचे हम समग्र प्रक्रिया प्रवाह के लिए अपने विकल्पों में जायेंगे और परिणाम क्या हैं।

गारंटी वितरण

आप कॉन्फ़िगर कर सकते हैं कि कितने दलालों, आपके संदेशों के प्राप्त होने की पुष्टि करने के लिए है से पहले अनुरोध acks पैरामीटर के साथ आप के लिए वापस आ गया है: को यह निर्धारित करके सभी आपको बता दलाल तब तक प्रतीक्षा करें जब तक कि सभी प्रतिकृतियां आपको जवाब देने से पहले आपके संदेश को स्वीकार नहीं कर लेतीं। यह अभी भी 100% गारंटी नहीं है कि आपका संदेश खोया नहीं जाएगा, क्योंकि यह अभी तक पृष्ठ कैश पर लिखा गया है और ब्रोकर के साथ सैद्धांतिक परिदृश्य हैं जो डिस्क पर बने रहने से पहले विफल रहता है, जहां संदेश अभी भी खो सकता है। लेकिन यह उतना ही अच्छा गारंटी है जितना आप प्राप्त करने जा रहे हैं। आप अंतराल को कम करके डेटा हानि को कम कर सकते हैं, जिस पर दलाल डिस्क पर fsync को मजबूर करते हैं (टेक्स्ट और/या flush.ms पर जोर दिया) लेकिन कृपया ध्यान रखें कि ये मान उनके साथ भारी प्रदर्शन ला सकते हैं दंड।

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

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

तो बुनियादी तौर पर आप एक ऐसी प्रणाली है जिसमें आप बेहतर डुप्लिकेट/लापता मूल्यों (या नहीं, आप आंशिक विफलताओं को पुन: भेजने के आधार पर) के साथ सौदा करने में सक्षम हैं चुनने की आवश्यकता है और उस आदेश आप में बातें करते हैं को प्रभावित करती है।

विकल्प 1

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

विकल्प 2

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

बेशक आप इन दोनों दृष्टिकोणों को कम करने और इन्हें बाद में पुनः प्रयास करने का ट्रैक रखकर दोनों दृष्टिकोणों को कम कर सकते हैं, लेकिन यह सब बड़े मुद्दे पर एक बैंडएड का अधिक है।

व्यक्तिगत रूप से मैं दृष्टिकोण 1 के साथ जाऊंगा, क्योंकि विफलता की संभावना खुद को भेजने से कुछ हद तक छोटी होनी चाहिए और काफ्का के दूसरी तरफ कुछ प्रकार की डुप्लिकेट जांच लागू करनी चाहिए।

---

यह पिछले एक से संबंधित है: मैं के साथ संदेश भेजने:। orderSource.output() भेजने के लिए (MessageBuilder.withPayload (क्रम) .build()); यह ऑपरेशन असीमित है और हमेशा सत्य लौटाता है, भले ही काफ्का ब्रोकर नीचे हो। मुझे कैसे पता चलेगा कि संदेश कफ्का ब्रोकर तक पहुंच गया है?

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

Producer<String, String> producer = new KafkaProducer<>(myConfig); 
final ArrayList<Exception> exceptionList = new ArrayList<>(); 

for(MessageType message : messages){ 
    producer.send(new ProducerRecord<String, String>("myTopic", message.getKey(), message.getValue()), new Callback() { 
    @Override 
    public void onCompletion(RecordMetadata metadata, Exception exception) { 
     if (exception != null) { 
     exceptionList.add(exception); 
     } 
    } 
    }); 
} 

producer.flush(); 

if (!exceptionList.isEmpty()) { 
    // do stuff 
} 

Answer 2

मुझे लगता है कि घटना सोर्सिंग लागू करने के लिए उचित तरीके से काफ्का सीधे कि मिला हुआ बोतलबंद पानी (https://www.confluent.io/blog/bottled-water-real-time-integration-of-postgresql-and-kafka/) या अधिक सक्रिय Debezium (http://debezium.io/) का उपयोग कर आरडीबीएमएस binlog e.g से पढ़ता है एक प्लगइन द्वारा धक्का घटनाओं से भरा जाना होने से है। फिर माइक्रोस्कोर्सेस का उपभोग उन घटनाओं को सुन सकता है, उन्हें उपभोग कर सकता है और आरडीबीएमएस डेटाबेस के साथ अंततः अपने संबंधित डेटाबेस पर कार्य कर सकता है। , https://stackoverflow.com/a/43607887/986160