👀 📳 🎇 अनुरोधों और उनके समाधानों के बैच प्रसंस्करण की समस्याएं (भाग 2) 🍳 🌦️ 🐯

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

सामग्री

कार्य की संक्षिप्त पुनरावृत्ति
पैकेज बढ़ाएं
रिवर्स इंजीनियरिंग
सभी अनुरोध एकत्र करें
समानता
समस्याओं
निष्कर्ष
व्यापार के आंकड़े
एक अनुबंध समस्या का समाधान
उपयोगी सीमाओं की खोज करें
ह्यूरिस्टिक एरर्स
निष्कर्ष
DDD- शैली इकाइयाँ
एग्रीगेट के अंदर एस्कलेटिंग क्वेश्चन का लॉजिक
कुल के बाहर प्रश्नों को एकत्र करने के लिए तर्क
निष्कर्ष
प्रॉक्सी और डबल कॉल
एक अनुबंध समस्या का समाधान
निष्कर्ष
निष्कर्ष

कार्य की संक्षिप्त पुनरावृत्ति

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

संदेश मॉडल चैट करें

data class ChatMessage (
// nullable persist
val id : Long ? = null ,
/** */
val author : UserReference ,
/** */
val message : String ,
/** */
// - JPA+ null,
val files : List < FileReference > ? = null ,
/** , */
val replyTo : ChatMessage ? = null ,
/** , */
val forwardFrom : ChatMessage ? = null
)

निर्भरता इंजेक्शन के माध्यम से, हम निम्नलिखित बाहरी सेवाओं को लागू कर सकते हैं:

interface ChatMessageRepository {
fun findLast ( n : Int ) : List < ChatMessage >
}

data class FileHeadRemote (
val id : FileReference ,
val name : String
)

interface FileRemoteApi {
fun getHeadById ( id : FileReference ) : FileHeadRemote
   fun getHeadsByIds ( id : Set < FileReference > ) : Set < FileHeadRemote >
fun getHeadsByIds ( id : List < FileReference > ) : List < FileHeadRemote >
}

data class UserRemote (
val id : UserReference ,
val name : String
)

interface UserRemoteApi {
fun getUserById ( id : UserReference ) : UserRemote
   fun getUsersByIds ( id : Set < UserReference > ) : Set < UserRemote >
fun getUsersByIds ( id : List < UserReference > ) : List < UserRemote >
}

हमें एक REST नियंत्रक लागू करने की आवश्यकता है:

interface ChatRestApi {
fun getLast ( n : Int ) : List < ChatMessageUI >
}

जहां:

/** */
data class ReferenceUI (
/** url */
val ref : String ,
/** */
val name : String
)

data class ChatMessageUI (
val id : Long ,
/** */
val author : ReferenceUI ,
/** */
val message : String ,
/** */
val files : List < ReferenceUI >,
/** , */
val replyTo : ChatMessageUI ? = null ,
/** , */
val forwardFrom : ChatMessageUI ? = null
)

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

पैकेज बढ़ाएं

भोली समाधानों की मुख्य समस्या संकुल के छोटे आकार की थी।

बड़े पैकेट में समूह कॉल करने के लिए, आपको किसी तरह अनुरोधों को जमा करना होगा। यह लाइन अनुरोधों के संचय का मतलब नहीं है:

author = userRepository . getUserById ( author ) . toFrontReference () ,

अब हमारे रनटाइम में उपयोगकर्ताओं की सूची को संग्रहीत करने के लिए कोई विशेष स्थान नहीं है - यह धीरे-धीरे बन रहा है। इसे बदलना होगा।

पहले आपको ChatMessage.toFrontModel विधि में मानचित्रण से डेटा अधिग्रहण के तर्क को अलग करने की आवश्यकता है:

private fun ChatMessage . toFrontModel (
getUser : ( UserReference ) -> UserRemote ,
getFile : ( FileReference ) -> FileHeadRemote ,
serializeMessage : ( ChatMessage ) -> ChatMessageUI
) : ChatMessageUI =
ChatMessageUI (
id = id ?: throw IllegalStateException ( " $ this must be persisted" ) ,
author = getUser ( author ) . toFrontReference () ,
message = message ,
files = files ?. let {
       it . map ( getFile ) . map { it . toFrontReference () }
} ?: listOf () ,
forwardFrom = forwardFrom ?. let ( serializeMessage ) ,
replyTo = replyTo ?. let ( serializeMessage )
)

यह पता चला है कि यह फ़ंक्शन केवल तीन बाहरी कार्यों पर निर्भर करता है (और संपूर्ण कक्षाओं पर नहीं, जैसा कि यह शुरुआत में था)।

इस तरह की बहाली के बाद, फ़ंक्शन का शरीर कम स्पष्ट नहीं हो गया, और अनुबंध कठिन हो गया (इसमें पेशेवरों और विपक्ष दोनों शामिल हैं)।

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

चूंकि serializeMessage फ़ंक्शन एक पुनरावर्ती फ़ंक्शन के समान है, इसलिए इसे रिफैक्टरिंग के पहले चरण में एक स्पष्ट पुनरावृत्ति के रूप में किया जा सकता है:

class ChatRestController (
private val messageRepository : ChatMessageRepository ,
private val userRepository : UserRemoteApi ,
private val fileRepository : FileRemoteApi
) : ChatRestApi {
override fun getLast ( n : Int ) =
messageRepository . findLast ( n )
. map { it . toFrontModel () }

मैंने toFrontModel मेथड के लिए एक स्टब बनाया, जो अब तक ठीक वैसे ही काम करता है जैसे हमारे पहले भोले-भाले कार्यान्वयन (तीनों बाहरी कार्यों का कार्यान्वयन समान रहता है)।

private fun ChatMessage . toFrontModel () : ChatMessageUI =
toFrontModel (
getUser = userRepository ::getUserById ,
getFile = fileRepository ::getHeadById ,
serializeMessage = { it . toFrontModel () }
   )

लेकिन हमें getUser, getFile, और serializeMessage फ़ंक्शन को कुशलतापूर्वक कार्य करने की आवश्यकता है, अर्थात, सही आकार के पैकेट में उचित सेवाओं के लिए अनुरोध भेजें (सैद्धांतिक रूप से, यह आकार प्रत्येक सेवा के लिए अलग हो सकता है) या आम तौर पर प्रति सेवा एक असीमित अनुरोध की अनुमति है।

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

आप अनुरोधों के संचय और उनके क्रमिक निष्पादन के साथ योजना भी आज़मा सकते हैं। हालांकि, ऐसी योजनाओं के लिए अतुल्यकालिक निष्पादन की आवश्यकता होगी, लेकिन अब हम समकालिक लोगों पर ध्यान केंद्रित करेंगे।

इसलिए, बैच प्रोसेसिंग का उपयोग शुरू करने के लिए, एक या दूसरे तरीके से हमें पहले से अधिक से अधिक अनुरोधों का पता लगाना होगा (अधिमानतः सभी) जो हमें बनाने होंगे। अगर हम REST कंट्रोलर के बारे में बात कर रहे हैं, तो पूरे सत्र में प्रत्येक सेवा के लिए अनुरोधों को संयोजित करना अच्छा होगा।

सभी कॉल समूह

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

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

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

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

चूंकि मेरे व्यवहार में इस तरह का दृष्टिकोण सबसे अधिक मांग में है, इसलिए आगे के उदाहरण इस विकल्प की चिंता करेंगे।

हम प्रश्नों के बड़े सेट प्राप्त करने के ऐसे तरीकों को अलग कर सकते हैं:

रिवर्स इंजीनियरिंग;
व्यवसाय के आंकड़े;
DDD की शैली में समुच्चय;
प्रॉक्सी और डबल कॉल।

आइए हमारी परियोजना के उदाहरण पर सभी विकल्पों को देखें।

रिवर्स इंजीनियरिंग

सभी अनुरोध एकत्र करें

चूँकि हमारे पास जानकारी एकत्र करने और इसे दृश्यपटल के लिए परिवर्तित करने में शामिल सभी कार्यों को लागू करने के लिए एक कोड है, हम रिवर्स इंजीनियरिंग कर सकते हैं और इस कोड से हम समझ सकते हैं कि अनुरोध क्या होंगे:

class ChatRestController (
private val messageRepository : ChatMessageRepository ,
private val userRepository : UserRemoteApi ,
private val fileRepository : FileRemoteApi
) : ChatRestApi {
override fun getLast ( n : Int ) =
messageRepository . findLast ( n )
. let { messages ->
         // , forward reply
val allMessages = messages . asSequence () . flatMap {
           sequenceOf ( it , it . forwardFrom , it . replyTo ) . filterNotNull ()
} . toSet ()
val allUserReq = allMessages . map { it . author }
         val allFileReq = allMessages . flatMap { it . files ?: listOf () } . toSet ()

सभी अनुरोध एकत्र किए जाते हैं, अब आपको वास्तविक बैच प्रसंस्करण करने की आवश्यकता है।

AllUserReq और allFileReq के लिए, हम बाहरी प्रश्न बनाते हैं और उन्हें आईडी द्वारा समूहित करते हैं। यदि पैकेज के आकार पर कोई प्रतिबंध नहीं है, तो यह कुछ इस तरह दिखाई देगा:

userRepository . getUsersByIds ( allMessages . map { it . author } . toSet ())
. associateBy { it . id } ::get
fileRepository . getHeadsByIds ( allMessages . flatMap { it . files ?: listOf () } . toSet ())
. associateBy { it . id } ::get

यदि कोई प्रतिबंध है, तो कोड निम्नलिखित रूप लेगा:

val userApiChunkLimit = 100
allMessages . map { it . author } . asSequence () . distinct ()
. chunked ( userApiChunkLimit , userRepository ::getUsersByIds )
. flatten ()
. associateBy { it . id } ::get

दुर्भाग्य से, स्ट्रीम के विपरीत, अनुक्रम आसानी से एक समानांतर पैकेट अनुरोध पर स्विच नहीं कर सकता है।

यदि आप एक समानांतर क्वेरी को मान्य और आवश्यक मानते हैं, तो आप कर सकते हैं, उदाहरण के लिए, यह:

allMessages . map { it . author } . parallelStream () . distinct ()
. chunked ( userApiChunkLimit , userRepository ::getUsersByIds )
. flatten ()
. associateBy { it . id } ::get

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

fun < T , R > Stream < out T >. chunked ( size : Int , transform : ( List < T > ) -> R ) : Stream < out R > =
batches ( this , size ) . map ( transform )

fun < T > Stream < out Collection < T >>. flatten () : Stream < T > =
flatMap { it . stream () }

fun < T , K > Stream < T >. associateBy ( keySelector : ( T ) -> K ) : Map < K , T > =
collect ( Collectors . toMap ( keySelector , { it } ))

अब सब कुछ एक साथ रखना बाकी है:

override fun getLast ( n : Int ) =
messageRepository . findLast ( n )
. let { messages ->
       // , forward reply
val allMessages = messages . asSequence () . flatMap { message ->
         sequenceOf ( message , message . forwardFrom , message . replyTo )
. filterNotNull ()
} . toSet ()

messages . map ( ValueHolder < ( ChatMessage ) -> ChatMessageUI > () . apply {
         value = memoize { message : ChatMessage ->
           message . toFrontModel (
// ,
getUser = allMessages . map { it . author } . parallelStream () . distinct ()
. chunked ( userApiChunkLimit , userRepository ::getUsersByIds )
. flatten ()
. associateBy { it . id } ::get . orThrow { IllegalArgumentException ( "User $ it" ) } ,
//
getFile = fileRepository . getHeadsByIds ( allMessages . flatMap { it . files ?: listOf () } . toSet ())
. associateBy { it . id } ::get . orThrow { IllegalArgumentException ( "File $ it" ) } ,
//
serializeMessage = value
)
}
} . value )
}

स्पष्टीकरण और सरलीकरण

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

fun < A , R > memoize ( func : ( A ) -> R ) = func as? Memoize2 ?: Memoize2 ( func )
class Memoize2 < A , R > ( val func : ( A ) -> R ) : ( A ) -> R , java . util . function . Function < A , R > {
private val cache = hashMapOf < A , R > ()
override fun invoke ( p1 : A ) = cache . getOrPut ( p1 , { func ( p1 ) } )
override fun apply ( t : A ) : R = invoke ( t )
}

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

inline fun < A , R > recursiveMemoize ( crossinline func : ( A , ( A ) -> R ) -> R ) : ( A ) -> R =
ValueHolder < ( A ) -> R > () . apply {
     value = memoize { a -> func ( a , value ) }
} . value

यदि आप पहली बार इस तीर के नारे को समझ सकते हैं - बधाई हो, तो आप एक कार्यात्मक प्रोग्रामर हैं :-)

अब कोड इस तरह दिखेगा:

override fun getLast ( n : Int ) =
messageRepository . findLast ( n )
. let { messages ->
       // , forward reply
val allMessages = messages . asSequence () . flatMap { message ->
         sequenceOf ( message , message . forwardFrom , message . replyTo )
. filterNotNull ()
} . toSet ()

// ,
val getUser = allMessages . map { it . author } . parallelStream () . distinct ()
. chunked ( userApiChunkLimit , userRepository ::getUsersByIds )
. flatten ()
. associateBy { it . id } ::get . orThrow { IllegalArgumentException ( "User $ it " ) }

       //
val getFile = fileRepository . getHeadsByIds ( allMessages . flatMap { it . files ?: listOf () } . toSet ())
. associateBy { it . id } ::get . orThrow { IllegalArgumentException ( "File $ it" ) }

       messages . map ( recursiveMemoize { message , memoized : ( ChatMessage ) -> ChatMessageUI ->
         message . toFrontModel (
getUser = getUser ,
getFile = getFile ,
//
serializeMessage = memoized
         )
} )

आप यह भी देख सकते हैं कि यह कैसे परिभाषित किया जाता है:

/** [exception] , null */
fun < P , R > (( P ) -> R ? ) . orThrow ( exception : ( P ) -> Exception ) : ( P ) -> R =
{ p -> invoke ( p ) . let { it ?: throw exception ( p ) } }

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

इस फिक्स के बाद, getLast रनटाइम लगभग 300 एमएस होने की उम्मीद है। इसके अलावा, यह समय थोड़ा बढ़ेगा, भले ही अनुरोध पैकेट आकार पर प्रतिबंधों में फिट न हों (क्योंकि पैकेट समानांतर में अनुरोध किए जाते हैं)। आपको याद दिलाता हूं कि हमारा न्यूनतम लक्ष्य 500 एमएस है, और 250 एमएस को सामान्य काम माना जा सकता है।

समानता

लेकिन आपको आगे बढ़ने की जरूरत है। UserRepository और fileRepository के कॉल पूरी तरह से स्वतंत्र हैं, और उन्हें आसानी से समानांतर किया जा सकता है, सिद्धांत में 200 एमएस।

उदाहरण के लिए, हमारे जॉइन फंक्शन के माध्यम से:

override fun getLast ( n : Int ) =
messageRepository . findLast ( n )
. let { messages ->
       // , forward reply
val allMessages = messages . asSequence () . flatMap { message ->
         sequenceOf ( message , message . forwardFrom , message . replyTo )
. filterNotNull ()
} . toSet ()

join ( {
         // ,
allMessages . map { it . author } . parallelStream () . distinct ()
. chunked ( userApiChunkLimit , userRepository ::getUsersByIds )
. flatten ()
. associateBy { it . id }
} , {
         //
fileRepository . getHeadsByIds ( allMessages . flatMap { it . files ?: listOf () } . toSet ())
. associateBy { it . id }
} ) . let { ( users , files ) ->
         messages . map ( recursiveMemoize { message , memoized : ( ChatMessage ) -> ChatMessageUI ->
           message . toFrontModel (
getUser = users ::get . orThrow { IllegalArgumentException ( "User $ it" ) } ,
getFile = files ::get . orThrow { IllegalArgumentException ( "File $ it " ) } ,
//
serializeMessage = memoized
           )
} )
}
}

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

समस्याओं

सामान्य तौर पर, कोड, निश्चित रूप से, हमारे भोले पहले संस्करण की तुलना में कम पठनीय है, लेकिन यह अच्छा है कि क्रमबद्धता ही (toFrontModel के कार्यान्वयन) लगभग नहीं बदली और पूरी तरह से पठनीय बनी रही। बाहरी सेवाओं के साथ चालाक काम का पूरा तर्क एक ही स्थान पर रहता है।

इस दृष्टिकोण का नकारात्मक पक्ष यह है कि हमारा अमूर्त कार्य आगे बढ़ रहा है।

यदि हमें toFrontModel में परिवर्तन करने की आवश्यकता है, तो हमें लगभग निश्चित रूप से getLast फ़ंक्शन में परिवर्तन करने होंगे, जो कि प्रतिस्थापन बारबरा लिस्कॉव (Liskov प्रतिस्थापन सिद्धांत) के सिद्धांत का उल्लंघन करता है।

उदाहरण के लिए, हम संलग्न फ़ाइलों को केवल मुख्य संदेशों में डिक्रिप्ट करने के लिए सहमत हुए, लेकिन उत्तरों में और आगे (उत्तर / आगे) में नहीं, या केवल प्रतिक्रियाओं में और पहले स्तर के आगे। इस स्थिति में, toFrontModel कोड में परिवर्तन करने के बाद, आपको फ़ाइलों के लिए अनुरोध संग्रह कोड में संबंधित सुधार करना होगा। इसके अलावा, सुधार nontrivial होगा:

fileRepository . getHeadsByIds (
allMessages . flatMap { it . files ?: listOf () } . toSet ()
)

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

निष्कर्ष

पेशेवरों:

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

विपक्ष:

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

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

व्यापार के आंकड़े

एक अनुबंध का समाधान

क्या होगा अगर हम एक सटीक अनुबंध के साथ काम नहीं करते हैं और इसलिए, अनुरोधों के सटीक सेट के साथ, लेकिन लगभग एक के साथ? इसके अलावा, हम एक अनुमानित सेट बनाएंगे ताकि इसमें सटीक सेट शामिल हो और यह विषय क्षेत्र की विशेषताओं पर आधारित हो।

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

उपयोगी सीमाओं की खोज करें

आइए हमारे उदाहरण के साथ ऐसा करने का प्रयास करें।
हमारे मामले में, निम्नलिखित व्यवसाय विशेषताएं हैं:

चैट प्रतिभागियों की सूची पूर्वनिर्धारित है;
चैट एक दूसरे से पूरी तरह से अलग हैं;
उत्तर / आगे की जंजीरों का घोंसला छोटा होता है (~ 2–3 संदेश)।

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

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

इसके अलावा, अगर चैट उपयोगकर्ताओं की सूची पूर्व निर्धारित है, लेकिन यह जानकारी उपयोगकर्ता सेवा में नहीं है (जो बहुत संभावना है), तो ऐसी जानकारी से कोई मतलब नहीं होगा। हम चैट उपयोगकर्ता सूची के लिए एक अतिरिक्त अनुरोध करेंगे, और फिर आपको अभी भी उपयोगकर्ता सेवा के लिए अनुरोध करना होगा।

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

यहाँ यह आश्चर्यजनक लग सकता है कि हम किसी भी चैट आइडेंटिफ़ायर को पास नहीं कर रहे हैं। मैंने जानबूझकर ऐसा किया था ताकि नमूना कोड को अव्यवस्थित न करूं।

पहली नज़र में, दूसरे प्रतिबंध से कुछ भी नहीं होता है। किसी भी स्थिति में, मुझे अभी भी उससे उपयोगी कुछ भी नहीं मिला।

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

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

दूसरी ओर, हमें यह याद रखना चाहिए कि हमें सभी चैट संदेश नहीं मिलते हैं, लेकिन केवल अंतिम एन, और यह आसानी से पता कर सकते हैं कि उनमें कोई भी फाइल नहीं है।

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

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

चूँकि हम अभी भी किसी न किसी तरह से अपनी पद्धति को लागू करना चाहते हैं, तो मान लीजिए कि आँकड़े निम्नलिखित दर्शाते हैं:

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

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

fun getHeadsByChat () : List < FileHeadRemote >

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

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

कोड निम्नानुसार है:

override fun getLast ( n : Int ) =
messageRepository . findLast ( n )
. let { messages ->
       join (
{ userRepository . getUsersByChat () . associateBy { it . id } } ,
{ fileRepository . getHeadsByChat () . associateBy { it . id } }
       )
. let { //
}
}

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

प्रीफ़ैच कोड बहुत छोटा और स्पष्ट हो गया है।

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

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

यह विकल्प एक स्थिर 100 ms प्रति अनुरोध देता है।

ह्यूरिस्टिक एरर्स

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

निष्कर्ष

पेशेवरों:

व्यापार के आंकड़ों के तर्क को पढ़ना आसान है और आमतौर पर तुच्छ है। यह प्रयोज्यता की सीमाओं को समझने, प्रीफैच अनुबंध को सत्यापित करने और संशोधित करने के लिए अच्छा है।
स्केलेबिलिटी रिवर्स इंजीनियरिंग जितनी ही अच्छी है।
डेटा के अनुसार कोड का सामंजस्य कम हो जाता है, जिससे कोड का बेहतर समानांतरकरण हो सकता है।
प्रीस्टेक तर्क, REST नियंत्रक के मुख्य तर्क की तरह, आवश्यकताओं पर आधारित है। यह एक कमजोर प्लस है अगर आवश्यकताएं अक्सर बदलती हैं।

विपक्ष:

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

, . - - , . , , .

. - , .

DDD

DDD, . DDD — . DDD, . , ( , ).

, — . , :

, prefetch . prefetch -, . , (-), , .

, «». ChatMessage, UserReference, FileReference , - , , . Chat, — ChatPage ChatPageMessage:

interface Chat {
fun getLastPage ( n : Int ) : ChatPage
}

interface ChatPage {
val messages : List < ChatPageMessage >
}

data class ChatPageMessage (
val id : Long ,
val author : UserRemote ,
val message : String ,
val files : List < FileHeadRemote >,
val replyTo : ChatPageMessage ? ,
val forwardFrom : ChatPageMessage ?
)

. , .

FileHeadRemote UserRemote , , .

, REST- :

class ChatRestController (
private val chat : Chat
) : ChatRestApi {
override fun getLast ( n : Int ) =
chat . getLastPage ( n ) . toFrontModel ()

private fun ChatPage . toFrontModel () =
messages . map { it . toFrontModel () }

   private fun ChatPageMessage . toFrontModel () : ChatMessageUI =
ChatMessageUI (
id = id ,
author = author . toFrontReference () ,
message = message ,
files = files . toFrontReference () ,
forwardFrom = forwardFrom ?. toFrontModel () ,
replyTo = replyTo ?. toFrontModel ()
)
}

, : , , , . , prefetch . , (Single Responsibility Principle, SRP).

, .

, -.

class ChatImpl (
private val messageRepository : ChatMessageRepository ,
private val userRepository : UserRemoteApi ,
private val fileRepository : FileRemoteApi
) : Chat {
override fun getLastPage ( n : Int ) = object : ChatPage {
override val messages : List < ChatPageMessage >
get () =
runBlocking ( IO ) {
           val prefetch = async (
{ userRepository . getUsersByChat () . associateBy { it . id } } ,
{ fileRepository . getHeadsByChat () . associateBy { it . id } }
           )

withContext ( IO ) { messageRepository . findLast ( n ) }
             . map (
prefetch . await () . let { ( users , files ) ->
                 recursiveMemoize { message , memoized : ( ChatMessage ) -> ChatPageMessage ->
                   message . toDomainModel (
getUser = users ::get . orThrow { IllegalArgumentException ( "User $ it " ) } ,
getFile = files ::get . orThrow { IllegalArgumentException ( "File $ it " ) } ,
//
serializeMessage = memoized
                   )
}
}
             )
}
   }
}

private fun ChatMessage . toDomainModel (
getUser : ( UserReference ) -> UserRemote ,
getFile : ( FileReference ) -> FileHeadRemote ,
serializeMessage : ( ChatMessage ) -> ChatPageMessage
) = ChatPageMessage (
id = id ?: throw IllegalStateException ( " $ this must be persisted" ) ,
author = getUser ( author ) ,
message = message ,
files = files ?. map ( getFile ) ?: listOf () ,
forwardFrom = forwardFrom ?. let ( serializeMessage ) ,
replyTo = replyTo ?. let ( serializeMessage )
)

, getLastPage , prefetch, toDomainModel .

userRepository, fileRepository messageRepository Kotlin . , - .

, , - -.

: ChatPage Chat, getLast(), . :

interface Chat {
fun getPage () : ChatPage
}

, Chat ChatPage:

class ChatPageImpl (
private val messageData : List < ChatMessage >,
private val userData : List < UserRemote >,
private val fileData : List < FileHeadRemote >
) : ChatPage {
override val messages : List < ChatPageMessage >
get () =
messageData . map (
( userData . associateBy { it . id } to fileData . associateBy { it . id } )
. let { ( users , files ) ->
             recursiveMemoize { message , self : ( ChatMessage ) -> ChatPageMessage ->
               message . toDomainModel (
getUser = users ::get . orThrow () ,
getFile = files ::get . orThrow () ,
//
serializeMessage = self
               )
}
}
       )
}

prefetch, :

fun chatPagePrefetch (
pageSize : Int ,
messageRepository : ChatMessageRepository ,
userRepository : UserRemoteApi ,
fileRepository : FileRemoteApi
) =
runBlocking ( IO ) {
     async (
{ userRepository . getUsersByChat () } ,
{ fileRepository . getHeadsByChat () } ,
{ messageRepository . findLast ( pageSize ) }
     )
}

, , prefetch. DDD Application Services.

class ChatService (
private val messageRepository : ChatMessageRepository ,
private val userRepository : UserRemoteApi ,
private val fileRepository : FileRemoteApi
) {
private fun chatPagePrefetch ( pageSize : Int ) =
runBlocking ( IO ) {
       async (
{ messageRepository . findLast ( pageSize ) } ,
{ userRepository . getUsersByChat () } ,
{ fileRepository . getHeadsByChat () }
       ) . await ()
}

   fun getLastPage ( n : Int ) : ChatPage =
chatPagePrefetch ( n )
. let { ( messageData , userData , fileData ) ->
         ChatPageImpl ( messageData , userData , fileData )
}
}

, Chat::getLastPage ChatService::getLastPage. :

class ChatRestController (
private val chat : ChatService
) : ChatRestApi

निष्कर्ष

prefetch , .
prefetch , , . prefetch , .
, prefetch .

prefetch, .

, prefetch , , . , , . , - ? , . , Mockito, , . .

, .

: . , - , .

, messageRepository, . , messageRepository , .

UserReference->UserRemote FileReference->FileHeadRemote, .

class ChatRestController (
private val messageRepository : ChatMessageRepository ,
private val userRepository : UserRemoteApi ,
private val fileRepository : FileRemoteApi
) : ChatRestApi {
override fun getLast ( n : Int ) : List < ChatMessageUI > {
val messages = messageRepository . findLast ( n )

//
fun transform (
getUser : ( UserReference ) -> UserRemote ,
getFile : ( FileReference ) -> FileHeadRemote
     ) : List < ChatMessageUI > =
messages . map (
recursiveMemoize { message , self ->
           message . toFrontModel ( getUser , getFile , self )
}
       )

//
val userIds = mutableSetOf < UserReference > ()
val fileIds = mutableSetOf < FileReference > ()
transform (
{ userIds += it ; UserRemote ( 0L , "" ) } ,
{ fileIds += it ; FileHeadRemote ( 0L , "" ) }
     )

return runBlocking ( IO ) {
       //
async (
{ userRepository . getUsersByIds ( userIds ) . associateBy { it . id } ::get . orThrow () } ,
{ fileRepository . getHeadsByIds ( fileIds ) . associateBy { it . id } ::get . orThrow () }
       ) . await () . let { ( getUser , getFile ) ->
         transform ( getUser , getFile )
}
}
   }
}

, , , -, . , , ( ).

-, . , . , , , , - .

निष्कर्ष

पेशेवरों:

prefetch - .
prefetch.
, -.

विपक्ष:

, , . , -: , ( , ).

, prefetch .

, - , prefetch .

निष्कर्ष

, . , ( ).

, . : ( , ), , , .

— - . . — - , , . , , , , .

— . , . .

अनुरोधों और उनके समाधानों के बैच प्रसंस्करण की समस्याएं (भाग 2)

कार्य की संक्षिप्त पुनरावृत्ति

पैकेज बढ़ाएं

रिवर्स इंजीनियरिंग

व्यापार के आंकड़े

DDD

निष्कर्ष

More articles: