Node.js और Redis पर उपयोगकर्ता सूचनाएं लंबित

कार्य विवरण

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


यदि इसी तरह के अनुप्रयोग सामाजिक में स्थित हैं। नेटवर्क, फिर सीधे सूचनाएं भेजने के लिए, VKontakte नेटवर्क इसके लिए safe.sendNotification विधि प्रदान करता है;
सहपाठियों के लिए: सूचनाएं ।
और Apple और Android उपकरणों के लिए, इस तरह की सूचनाएं क्रमशः Apple Push Notification Service और Android Cloud से Device Messaging (C2DM) तक भेजी जा सकती हैं।

हल करने के लिए उपकरणों का विकल्प

इस समस्या को हल करने के लिए, दो अचल संपत्तियों का चयन किया गया था:
Node.js (0.6.x) और रेडिस। प्लस प्राप्त और भेजे गए नोटिफिकेशन और परीक्षण के लिए प्रतिज्ञा दिखाने के लिए एक आदिम UI को लागू करने के लिए व्यक्त करते हैं।

पहले क्लस्टर मॉड्यूल (0.6 में शामिल) के माध्यम से आसानी से सभी कोर का उपयोग करने की क्षमता शामिल थी, Node.js संस्करण 0.4 क्लस्टर के लिए एक अलग क्लस्टर पैकेज के रूप में बनाया गया है।

रेडिस को विभिन्न डेटा प्रकारों का समर्थन करने के लिए चुना गया था, जो कि बॉक्स के ठीक बाहर है (उसी मेमेचे की तुलना में)। साथ ही कार्य कतारों के आयोजन के लिए सुविधाजनक आदेशों की उपलब्धता। और यह भी क्योंकि Youporn.com अब 100% रेडिस साइट है

कोर एपीआई

server_name / send_delayed - विलंबित अधिसूचना (GET और POST दोनों मापदंडों को स्वीकार करता है)

• देरी - अंतराल जिसके बाद भेजना आवश्यक है (सेकंड में सेट)
• url - सीधे वह पता जिसे आपको समय आने पर अनुरोध करने की आवश्यकता है। सामाजिक नेटवर्क के लिए, यह sendNotification या सूचनाओं के माध्यम से भेज रहा है। SendSimple, पुश भेजने के लिए इसे एक अतिरिक्त सेवा पर भेजा जा सकता है जो सीधे APNS या C2DN को भेजने से संबंधित है।
• यूआईडी - उपयोगकर्ता आईडी।
• प्रकार - आंकड़ों के लिए पैरामीटर (नियुक्ति 1, नियोजित_कोल)
• recheck_url - एक अतिरिक्त पता जिसे अधिसूचना भेजने से पहले अनुरोध किया जाना चाहिए। यह किसी कार्य की वर्तमान स्थिति को ट्रैक करने के लिए उपयोगी हो सकता है। उदाहरण के लिए, निर्माण को तेज किया गया था या रद्द कर दिया गया था। url को "ठीक" या "त्रुटि" का उत्तर देना चाहिए, सकारात्मक उत्तर के मामले में, एक अधिसूचना भेजी जाएगी। (पैरामीटर वैकल्पिक है और यदि यह अनुपस्थित है तो कोई जाँच नहीं की जाएगी)
• send_than_online - एक अतिरिक्त पैरामीटर, यदि मौजूद है, तो घटनाओं को कतार में जोड़ा जाता है, लेकिन भेजने को केवल तभी किया जाएगा जब / user_online अनुरोध कहा जाता है। इस पैरामीटर की शुरूआत इस तथ्य के कारण है कि कई नेटवर्क प्रति दिन प्रति उपयोगकर्ता पैकेज की संख्या को सीमित करते हैं।

server_name / user_online - एक विशिष्ट उपयोगकर्ता के लिए एक घटना ले लो (यदि समय आ गया है) और भेजें

• यूआईडी - उपयोगकर्ता आईडी

ऐप डिजाइन

कतार के लिए मुख्य संरचनाएं:
संदेश कतार - SORTEDSET जिसमें रैंक घटना का समय होगा;
ऑनलाइन संदेशों की कतार - प्रत्येक यूआईडी के लिए SORTEDSET अलग है, जिसमें रैंक घटना का समय होगा;
प्रसंस्करण के लिए कतार - सूची - यदि यह संदेश इस सूची में है, तो संदेश संसाधित किया जाएगा।

मास्टर प्रक्रिया

1. सर्वर पर कोर की संख्या के बराबर कई श्रमिकों को शुरू करता है;
2. आरईपीएल शुरू होता है;
3. एक संदेश कतार स्कैनर लॉन्च करता है ( ऑनलाइन संदेश कतार स्कैन नहीं की जाती है);
4. श्रमिकों के पतन पर नज़र रखता है और गिरावट की स्थिति में उन्हें पुनः आरंभ करता है।
5. पूर्ण गिरावट की स्थिति में व्यक्तिगत घटनाओं से संदेश कतार को पुनर्स्थापित करता है;
6. आँकड़ों को ट्रैक करने के लिए यूआई सर्वर शुरू करता है।

स्कैनर वर्तमान समय से रिवर्स ऑर्डर में 100 के बराबर राशि में संदेश कतार से ZREVRANGEBYSCORE के माध्यम से घटनाओं का एक पैकेट प्राप्त करता है। लेन-देन में, उन्हें प्रसंस्करण के लिए एक प्रसंस्करण कतार में रखता है और उन्हें संदेश कतार से हटा देता है।

अलग काम करनेवाला

1. send_delayed या user_online कमांड को स्वीकार करता है। ये घटनाएँ JSON.stringify के माध्यम से एचएएसएच में जेएसएन प्रारूप में दर्ज की जाती हैं। Msgpack में इवेंट स्टोरेज की भी कोशिश की गई है। यद्यपि उपरोक्त परीक्षणों के अनुसार यह लिखा गया है कि यह तेज होना चाहिए, लेकिन सरल वस्तुओं के लिए गति में कोई वृद्धि नहीं हुई थी। घटना पहचानकर्ता एक यादृच्छिक घर द्वारा बनाया गया है और एक संदेश कतार में रखा गया है। घटना की प्रत्येक कुंजी के लिए, सेटिंग्स से प्रतिक्रिया समय + देरी के बराबर एक जीवनकाल निर्धारित किया जाता है।

2. प्रसंस्करण के लिए एक कतार स्कैनर लॉन्च किया
स्कैनर BLPOP के माध्यम से LIST से ईवेंट प्राप्त करता है (केवल उन्हें लेने के लिए जब वे कतार में होते हैं)। यदि आप सर्वर में चलने वाले सर्वर के साथ MONITOR कमांड चलाते हैं, तो लॉग कुछ इस तरह दिख सकता है:

1336484301.674405 "zrevrangebyscore" "actual_queue_test-queue" "1336484301673" "-inf" "limit" "0" "100"
1336484301.675102 "zrevrangebyscore" "actual_queue_test-queue" "1336484301674" "-inf" "limit" "0" "100"
1336484301.675387 "zrevrangebyscore" "actual_queue_test-queue" "1336484301675" "-inf" "limit" "0" "100"

// fill processing queue
Worker.prototype.fetch = function(fn) {
var self = this;
this.client.zrevrangebyscore(this.queue_key, Date.now(),
"-inf", "limit", 0, self.pack, function(err, reply) {
process.nextTick(function() {
self.fetch();
}); // re-schedule fetch
...

इस मामले में, हर बार जब हम अगले टिक पर लाते हैं। जैसा कि आप देख सकते हैं, अगर कतार में कोई घटना नहीं है, तो प्रत्येक टिक पर इसे जांचने का कोई मतलब नहीं है।
JSON.parse करता है, अगर अनुपस्थित या (वर्तमान समय - कुंजी> USER_POLL_THRESHOLD - 30 मिनट) - हम अंतिम भेजने की कुंजी की जांच करते हैं - हम उस इवेंट-डेटा ईवेंट को ट्रिगर करते हैं, जिस पर UrlSender एक अनुरोध भेज सकता है, recheck_url की जाँच करने के बाद, यदि कोई हो। यदि http अनुरोध पर 200 कोड प्राप्त होता है, तो हम ईवेंट 'इवेंट- सेंड ' की शुरुआत करते हैं, डेटाबेस में ईवेंट को मिटा देते हैं, वर्तमान समय तक उपयोगकर्ता के लिए अंतिम भेजने की कुंजी सेट करते हैं।

ऑनलाइन घटनाओं को संसाधित करने का क्रम इस प्रकार है: हम एक विशिष्ट उपयोगकर्ता के लिए ऑनलाइन संदेशों की एक कतार प्राप्त करते हैं, ZREVRANGEBYSCORE के माध्यम से निकटतम घटना प्राप्त करते हैं। हम USET_POLL_THRESHOLD_ONLINE सेट करके जांचते हैं कि क्या भेजने का समय आ गया है और उन्हें प्रसंस्करण कतार में डाल दिया जाता है और उन्हें संदेश कतार से निकाल दिया जाता है।

परीक्षण

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

logger.spec('event-processed'); //only test env
…
logger.spec('event-fetched'); //only test env

चल रहे परीक्षण:
1. config / settings.test.js कॉपी करें -> config / settings.js और सेटिंग्स को अपने स्वयं में बदलें;
2. सर्वर प्रारंभ: APP_ENV = परीक्षण npm प्रारंभ
3. चल रहे परीक्षण: APP_ENV = परीक्षण प्रतिज्ञा --spec परीक्षण / विलंबित-अनुरोध-test.js

टेस्ट स्क्रिप्ट उदाहरण (माणिक)

require 'uri'
require 'net/http'

api_url = "http://127.0.0.1:8000/send_delayed"
url = "http://127.0.0.1:8000/health"
recheck_url = "http://127.0.0.1:8000/recheck_url_correct"
# recheck_url = "http://127.0.0.1:8000/recheck_url_error"
uid=rand(1000)
type="test_common"
delay=1
post_url = URI.parse(api_url)
# res = Net::HTTP.post_form(post_url,{"url" => url, "delay" => delay,
# "uid" => uid, "type" => type, "recheck_url" => recheck_url, "send_than_online" => true})
res = Net::HTTP.post_form(post_url,{"url" => url, "delay" => delay,
"uid" => uid, "type" => type, "recheck_url" => recheck_url})
p res.inspect

सांख्यिकी इंटरफ़ेस उदाहरण देखें

सर्वर स्रोत कोड

पिंकी द्वारा सह-लिखित
https://github.com/catz/actual_queue