वसंत एकीकरण प्रवाह में प्रथाओं को संभालने में त्रुटि

Question

मेरे पास वसंत एकीकरण प्रवाह है जिसमें एसिंक निष्पादन शामिल है, गेटवे से नियंत्रक तक मूल्य लौटा रहा है, मूल्य लौटने के बाद निरंतर एकीकरण प्रवाह है। यहाँ

@MessagingGateway 
public interface GW { 

    @Gateway(requestChannel = "f.input") 
    Task input(Collection<MessengerIncomingRequest> messages); 

} 

और प्रवाह है:

यहाँ प्रवेश द्वार है

@Bean 
IntegrationFlow jFlow() { 
     return IntegrationFlows.from(
     MessageChannels.executor("f.input", executor())) 
     .split() 
     .channel(MessageChannels.executor(executor())) 
     .transform(transformer) 
     .channel(routerChannel()) 
     .get(); 
} 

@Bean 
ThreadPoolTaskExecutor executor() { 
     ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor(); 
     ... 
     return pool; 
} 

@Bean 
MessageChannel routerChannel() { 
     return MessageChannels 
     .publishSubscribe("routerChannel", executor()) 
     .get(); 
} 

@Bean 
IntegrationFlow routerChannelFlow() { 
     return IntegrationFlows 
     .from(routerChannel()) 
     .publishSubscribeChannel(s -> s 
     .subscribe(f -> f.bridge(null)) 
     .subscribe(process())) 
     .get(); 
} 

@Bean 
IntegrationFlow process() { 
     return f -> 
     f.route(p -> p.getKind().name(), 
     m -> m.suffix("Channel") 
     .channelMapping(TaskKind.CREATE.name(), "create") 
     .... 
} 

@Bean 
IntegrationFlow createFlow() { 
     return IntegrationFlows.from(
     MessageChannels.direct("createChannel")) 
     .handle(routerService) 
     .get(); 
} 

मैं पूरी प्रवाह के लिए एक त्रुटि हैंडलर को कैसे परिभाषित कर सकते हैं? सर्वोत्तम प्रथाएं क्या हैं? मुझे पता है कि मैं गेटवे विधि कॉल के लिए एक कोशिश/पकड़ ब्लॉक डाल सकता हूं, लेकिन यह channel(routerChannel()) से पहले आने वाली सभी चीज़ों के लिए केवल jFlow प्रवाह में अपवादों को पकड़ लेगा।

मैं बाकी प्रवाह के लिए त्रुटियों को कैसे संभाल सकता हूं? या पूरे प्रवाह के लिए?

अद्यतन

मैं publishSubscribeChannel

@Bean 
IntegrationFlow routerChannelFlow() { 
    return IntegrationFlows 
      .from(routerChannel()) 
      .publishSubscribeChannel(s -> s 
        .subscribe(f -> f.bridge(null)) 
        .subscribe(process()) 
        .errorHandler(errorHandler)) 
      .get(); 
} 

के लिए त्रुटि हैंडलर जोड़ा है, लेकिन यह क्योंकि अपवाद मैं निम्नलिखित त्रुटि मिलती है के मामले में, मदद करने के लिए प्रतीत नहीं होता:

cMessagingTemplate$TemporaryReplyChannel : Reply message received but the receiving thread has already received a reply:ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: 

और मेरी त्रुटि हैंडलर को कॉल नहीं किया जाता है।

अद्यतन

गैरी के जवाब के अनुसार मैं इस कोड की कोशिश की:

@Bean 
IntegrationFlow jFLow() { 
    return IntegrationFlows.from(
      MessageChannels.executor("f.input", executor())) 
      .split() 
      .channel(MessageChannels.executor(executor())) 
      .transform(transformer) 
      .channel(routerChannel()) 
      .get(); 
} 

@Bean 
IntegrationFlow exceptionOrErrorFlow() { 
    return IntegrationFlows.from(
      MessageChannels.direct("exceptionChannel")) 
      .handle(errorHandler, "handleError") 
      .get(); 
} 

    @Bean 
MessageChannel exceptionChannel() { 
    return MessageChannels.direct("exceptionChannel") 
      .get(); 
} 

@Bean 
IntegrationFlow process() { 
     return f -> 
     f.enrichHeaders((spec) -> 
        spec.header("errorChannel", "exceptionChannel", true)) 
     f.route(p -> p.getKind().name(), 
     m -> m.suffix("Channel") 
     .channelMapping(TaskKind.CREATE.name(), "create") 
     .... 
} 

@MessagingGateway(errorChannel = "exceptionChannel") 

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

Answer 1

सबसे पहले, मुझे बताएं कि गेटवे कैसे काम करता है - इसे नीचे दिए गए समाधान को समझने में मदद करनी चाहिए।

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

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

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

तो, एक समाधान स्वतंत्र प्रवाह पर भेजे गए संदेश पर errorChannel शीर्षलेख को ठीक करना है। गेटवे द्वारा स्थापित errorChannel शीर्षलेख को प्रतिस्थापित करने के लिए .enrichHeaders का उपयोग करें (सत्य पर ओवरराइट सेट करना सुनिश्चित करें)। इसे प्रवाह में जितनी जल्दी हो सके किया जाना चाहिए ताकि किसी भी अपवाद को उस चैनल पर भेजा जा सके (और फिर आप वहां अपने त्रुटि हैंडलर की सदस्यता ले सकें)।

एक वैकल्पिक समाधान अपनी खुद की त्रुटि से निपटने निष्पादक अप तार, स्पष्ट रूप से अपने MessagePublishingErrorHandler पर एक defaultErrorChannel की स्थापना और errorChannel हैडर दूर करने के लिए है।

एसिंक त्रुटि रूटिंग पहले हेडर की तलाश में है; यदि मौजूद है, तो त्रुटि संदेश वहां रूट किया गया है; अगर कोई हेडर नहीं है और MPEH में कोई डिफ़ॉल्ट त्रुटि चैनल नहीं है; संदेश को डिफ़ॉल्ट errorChannel पर भेजा जाएगा, जिसमें (सामान्यतः) LoggingChannelAdapter सदस्यता ली गई है। डिफ़ॉल्ट errorChannel एक पब/उप चैनल है ताकि आप इसके अन्य अंतराल की सदस्यता ले सकें।

संपादित

आप पब/उप से पहले चैनल बदल रहे हैं।

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

यदि आप अपने exceptionChannel पर गेटवे पर errorChannel सेट करते हैं तो दोनों पैरों पर अपवाद वहां जाएंगे।