कार्य समानांतर लाइब्रेरी - कस्टम कार्य शेड्यूलर

Question

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

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

1. प्रति बार निर्धारित कार्यों की संख्या सीमित करें। मुझे लगता है कि अनुरोधों की संख्या इस सीमा से अधिक हो गई है तो उसे कार्य को दूर करने या संभवतः ब्लॉक करने की आवश्यकता होगी? (कार्यों के बैक लॉग को रोकने के लिए)
2. पता लगाएं कि क्या अनुरोध पहले ही शेड्यूलर में निष्पादित किया गया है, लेकिन अभी तक नहीं हुआ है और यदि ऐसा नहीं है तो दूसरे कार्य को कतारबद्ध करें लेकिन इसके बजाय पहले वापस करें।

क्या लोगों को लगता है कि ये जिम्मेदारियों के प्रकार हैं जिन पर कार्य शेड्यूलर से निपटना चाहिए या क्या मैं गलत पेड़ को भड़क रहा हूं? यदि आपके पास विकल्प हैं तो मैं सुझावों के लिए खुला हूं।

Answer 1

मैं दूसरों कि TPL Dataflow एक की तरह लगता है के साथ सहमत लगता थ्रॉटल तुलना में एक बेहतर फिट होने के लिए प्रतीत होता है इसके लिए अच्छा समाधान।

प्रसंस्करण सीमित करने के लिए, आप एक TransformBlock बना सकते हैं कि वास्तव में किसी भी तरह से डेटा को बदलने नहीं है, यह सिर्फ यह विलंब अगर यह पिछले डेटा के बाद बहुत जल्द ही आ:

static IPropagatorBlock<T, T> CreateDelayBlock<T>(TimeSpan delay) 
{ 
    DateTime lastItem = DateTime.MinValue; 
    return new TransformBlock<T, T>(
     async x => 
       { 
        var waitTime = lastItem + delay - DateTime.UtcNow; 
        if (waitTime > TimeSpan.Zero) 
         await Task.Delay(waitTime); 

        lastItem = DateTime.UtcNow; 

        return x; 
       }, 
     new ExecutionDataflowBlockOptions { BoundedCapacity = 1 }); 
} 

फिर एक बनाने विधि है कि डेटा (0 से शुरू उदाहरण पूर्णांकों के लिए) का उत्पादन:

static async Task Producer(ITargetBlock<int> target) 
{ 
    int i = 0; 
    while (await target.SendAsync(i)) 
     i++; 
} 

यह एसिंक्रोनस लिखा है ताकि अगर लक्ष्य ब्लॉक अभी आइटम प्रोसेस करने में सक्षम नहीं है, यह इंतजार करेंगे। ,

static void Consumer(int i) 
{ 
    Console.WriteLine(i); 
} 

और अंत में यह सब एक साथ लिंक और यह शुरू:

फिर एक उपभोक्ता विधि लिखने

var delayBlock = CreateDelayBlock<int>(TimeSpan.FromMilliseconds(500)); 

var consumerBlock = new ActionBlock<int>(
    (Action<int>)Consumer, 
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded }); 

delayBlock.LinkTo(consumerBlock, new DataflowLinkOptions { PropagateCompletion = true }); 

Task.WaitAll(Producer(delayBlock), consumerBlock.Completion); 

यहाँ, delayBlock सबसे में एक आइटम हर 500 एमएस और स्वीकार करेंगे Consumer() विधि समानांतर में कई बार चला सकती है। प्रसंस्करण को समाप्त करने के लिए, delayBlock.Complete() पर कॉल करें।

यदि आप अपने # 2 प्रति कुछ कैशिंग जोड़ना चाहते हैं, तो आप वहां एक और TransformBlock बना सकते हैं और इसे अन्य ब्लॉक से लिंक कर सकते हैं।

Answer 2

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

अद्यतन:

उस मामले में, TPL Dataflow अपने पसंदीदा समाधान अगर आप CTP का लाभ उठा सकते है। एक थ्रॉटल बफरब्लॉक समाधान है।

यह उदाहरण documentation provided by Microsoft से आता है:

// Hand-off through a bounded BufferBlock<T> 
private static BufferBlock<int> m_buffer = new BufferBlock<int>(
    new DataflowBlockOptions { BoundedCapacity = 10 }); 

// Producer 
private static async void Producer() 
{ 
    while(true) 
    { 
     await m_buffer.SendAsync(Produce()); 
    } 
} 

// Consumer 
private static async Task Consumer() 
{ 
    while(true) 
    { 
     Process(await m_buffer.ReceiveAsync()); 
    } 
} 

// Start the Producer and Consumer 
private static async Task Run() 
{ 
    await Task.WhenAll(Producer(), Consumer()); 
} 

अद्यतन:

चेक बाहर RX के Observable.Throttle।

Answer 3

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

1. BufferBlock<T> से शुरू करें जो कि लॉजिकल ब्लॉक है जिसे आप पोस्ट करेंगे।
2. BufferBlock<T> को एक कस्टम ब्लॉक में लिंक करें जिसमें अनुरोध/सेक और थ्रॉटलिंग तर्क का ज्ञान है।
3. कस्टम ब्लॉक को 2 से ActionBlock<T> पर लिंक करें।

मेरे पास इस सेकेंड के लिए # 2 के लिए कस्टम ब्लॉक लिखने का समय नहीं है, लेकिन मैं बाद में जांच करूँगा और आपके लिए एक कार्यान्वयन भरने का प्रयास करूंगा यदि आपने इसे पहले से ही नहीं निकाला है।

Answer 4

मैंने आरएक्स का अधिक उपयोग नहीं किया है, लेकिन एएफएआईसीटी अवलोकन योग्य। विन्डो विधि इसके लिए ठीक काम करेगी।

http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.window(VS.103).aspx

यह जो तत्वों फेंक, जो मेरा अनुमान है कि नहीं है कि आप क्या चाहते