🤫 👇🏽 🍏 Node.js में एक लाख लाइक्स या टास्क कतारों को इकट्ठा करें 🥥 ✉️ 🗾

पिछले हफ्ते हमने एक दौर की तारीख मनाई - एक करोड़ उपयोगकर्ता लाइकस्टोर डेटाबेस में जमा हुए पसंद करते हैं, न तो अधिक और न ही कम।

हम जावास्क्रिप्ट का उपयोग करते हैं, सभी वर्तमान सेवाएं जावास्क्रिप्ट / नोड में लिखे गए हैं। सामान्य तौर पर, मुझे हमारे प्रोजेक्ट में Node.js के उपयोग का कोई अफसोस नहीं है, यह खुद को HTTP API को लागू करने के लिए सबसे अच्छा उपकरण साबित हुआ है। लेकिन "पसंद" इकट्ठा करने के लिए, यह एक daemon होना चाहिए जो लगातार काम करता है। Node.js के लिए संभवतः सबसे विशिष्ट कार्य नहीं है - कार्यान्वयन की बारीकियों और कुछ नुकसानों के बारे में, हम आगे पढ़ते हैं।

Sollector

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

अधिकतम दक्षता के लिए, हम कलेक्टर के 2 उदाहरण चलाते हैं, विभिन्न मोड में काम करते हैं: प्रारंभिक, सामान्य। प्रारंभिक मोड में, कलेक्टर केवल नए जुड़े नेटवर्क की प्रक्रिया करता है। इस प्रकार, उपयोगकर्ता नेटवर्क को जोड़ने के बाद जितनी जल्दी हो सके "पसंद" प्राप्त करता है। सभी "पसंद" अपलोड किए जाने के बाद, नेटवर्क सामान्य मोड में जाता है, एक अन्य उदाहरण द्वारा संसाधित किया जाता है, जहां शुल्क के बीच का समय बहुत अधिक होता है।

 var argv = require('optimist').argv; var env = process.env.NODE_ENV = process.env.NODE_ENV || 'development'; var mode = process.env.COLLECTOR_MODE = process.env.COLLECTOR_MODE || argv.mode || 'normal'; var scheduler = require('./source/engine/scheduler'); scheduler(mode).run();

समयबद्धक

शेड्यूलर अनिवार्य रूप से Node.js. के लिए लिखे गए while(true) के समान ही है सच कहूँ तो, "सिंक्रोनस" से "एसिंक्रोनस" मोड में सोच को बदलना मेरे लिए सबसे आसान प्रक्रिया नहीं थी। Node.js में अनंत संख्या में कार्यों को शुरू करना एक कठिन काम की तरह लग रहा था, विचार के परिणामस्वरूप यह प्रश्न SO पर पैदा हुआ था।

एक विकल्प async.queue का उपयोग करना था , जो स्पष्ट लग रहा था लेकिन इस कार्य के लिए सबसे अच्छा नहीं था। कई प्रयासों के बाद, सेटटाइमआउट असिंक्रोनस के लिए सबसे अच्छा विकल्प बन गया, जबकि (सच)।

 function scheduler (mode) { function schedulerLoop() { runCollectingTasks(restartScheduler); } function restartScheduler (err, results) { if (err) { logger.error(err); } // http://stackoverflow.com/questions/16072699/nodejs-settimeout-memory-leak var timeout = setTimeout(schedulerLoop, config.collector.schedulerRestart); } // ... return { run: function () { schedulerLoop(); } }; }

यहाँ यह Node.js - मेमोरी लीक पर डेमन के बहुत नुकसान को नोट किया जाना चाहिए। मैंने देखा कि कलेक्टर के लंबे समय तक काम करने के बाद, उन्होंने उस समय बड़ी मात्रा में मेमोरी का उपभोग करना शुरू कर दिया और सबसे अप्रत्याशित क्षण बस रुक गया। एसओ पर सवाल के साथ कोड में टिप्पणी पर ध्यान दें । जब मैंने var timeout = जोड़ा, उसके बाद स्थिति में सुधार हुआ, लेकिन मौलिक रूप से नहीं।

मेमोरी लीक के बारे में एक महाकाव्य पोस्ट और जॉयंट और वॉलमार्ट इंजीनियरों द्वारा एक जांच के बाद एक और कारण सामने आया। Node.js v0.10.22 के संक्रमण के साथ, कलेक्टर ने और भी अधिक स्थिर काम करना शुरू कर दिया। फिर भी, स्वतःस्फूर्त ठहराव होता है, इसका कारण समझना अत्यंत कठिन है।

नेटवर्क और राज्य

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

 function runCollectingTasks(callback) { prepareCollectingTasks(function (err, tasks) { if (err) { return callback(err); } runTasks(tasks, 'collecting', callback); }); } function prepareCollectingTasks(callback) { networks.findByMode(mode, function (err, states) { if (err) { return callback({message: 'error during networks query', err: err}); } if (!states) { return callback({message: 'failed to read networks states'}); } callback(null, createCollectingTasks(states)); }); }

कार्य

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

केवल उन कार्यों को जो वर्तमान बिंदु के बाद निर्धारित किए गए थे, उन्हें चलाने की अनुमति है।

 function createCollectingTasks(states) { var tasks = states.map(function (state) { return allowedToExecute(state) ? collectingTask(state) : null; }).filter(function (task) { return task !== null; }); return tasks; } function allowedToExecute (state) { return moment().diff(state.scheduledTo) > 0; } function collectingTask(state) { return function (callback) { return executor(state, connectors, callback); }; }

डेटा का एक सरणी फ़ंक्शन के एक सरणी में परिवर्तित हो जाता है जो runTasks इनपुट पर जाता है। क्रमिक रूप से async.series माध्यम से प्रत्येक कार्य करता है।

 function runTasks(tasks, type, callback) { async.series(tasks, function (err) { if (err) { // report error but continue execution to do not break execution chain.. logger.error(err); } callback(null, tasks.length); }); }

निर्वाहक

एक सामान्यीकृत फ़ंक्शन जो वर्तमान स्थिति लेता है, मौजूदा कनेक्टर्स की एक सूची और एक कॉलबैक फ़ंक्शन (मैंने इसका सार दिखाने के लिए सभी त्रुटि हैंडलिंग और लॉगिंग को हटा दिया है)।

 function executor(state, connectors, callback) { var service = state.service; var connector = connectors[service]; var connectorStarted = moment(); connector(state, user, connectorExecuted); function connectorExecuted(err, updatedState, results) { saveConnectorState(state, connectorStateSaved); function saveConnectorState (state, callback) { // ... } function connectorStateSaved (err) { // ... saveConnectorResults(results, connectorResultsSaved); } function saveConnectorResults(results, callback) { // ... connectorResultsSaved(results, connectorResultsSaved); } function connectorResultsSaved (err, saveDuration) { // ... callback(null); } } }

कनेक्टर्स

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

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

मैं GitHub के लिए कोड का एक उदाहरण दूंगा (फिर, मैं सार दिखाने के लिए विवरण को छोटा करूंगा)।

 var API = 'https://api.github.com'; function connector(state, user, callback) { var accessToken = state.accessToken; if (!accessToken) { return callback('missing accessToken for user: ' + state.user); } initState(state); var uri = formatRequestUri(accessToken, state); var headers = { 'Content-Type': 'application/json', 'User-Agent': 'likeastore/collector'}; request({uri: uri, headers: headers, timeout: config.collector.request.timeout, json: true}, function (err, response, body) { if (err) { return handleUnexpected(response, body, state, err, function (err) { callback (err, state); }); } return handleResponse(response, body); }); function initState(state) { // intialize state fields (page, errors, mode etc.) ... } function formatRequestUri(accessToken, state) { // create request URI based on values from state ... } function handleResponse(response, body) { var rateLimit = +response.headers['x-ratelimit-remaining']; var stars = body.map(function (r) { return { // transforms response into likeastore item }; }); return callback(null, scheduleTo(updateState(state, stars, rateLimit, false)), stars); } function updateState(state, data, rateLimit, failed) { state.lastExecution = moment().toDate(); // update other state fields (next page, mode) ... return state; } }

scheduleTo

अंत में, जब कनेक्टर पूरा हो गया है और स्थिति अपडेट हो गई है, तो आपको अगले प्रारंभ समय की गणना करने की आवश्यकता है। यह एपीआई सीमाओं और कलेक्टर ऑपरेशन मोड के आधार पर गणना की जाती है (प्रारंभिक मोड के लिए, ठहराव न्यूनतम है, सामान्य मोड के लिए यह सामान्य है, आमतौर पर 15 मिनट, यदि कनेक्टर दर सीमा पर जाता है तो ठहराव अधिकतम है)।

 function scheduleTo(state) { var currentMoment = moment(); var service = state.service; var scheduleForMode = { initial: config.collector.quotes[service].runAfter, normal: config.collector.nextNormalRunAfter, rateLimit: config.collector.nextRateLimitRunAfter }; var next = scheduleForMode[state.mode]; var scheduledTo = currentMoment.add(next, 'milliseconds'); state.scheduledTo = scheduledTo.toDate(); return state; }

यहां ऐसा सीधा कोड है जो पिछले साल सितंबर के आसपास "बस गया" और हम तब से सभी जोड़ते हैं, नए कनेक्टर हैं, इंजन स्वयं अपरिवर्तित रहता है। मैं Node.js में कार्य कतार चलाने के लिए एक अलग पुस्तकालय आवंटित करने के बारे में सोच रहा हूं, लेकिन मुझे यह सुनिश्चित नहीं है कि यह एक सामान्यीकृत कार्य कितना है।

समय समाप्त हो रहा है, उपयोगकर्ताओं की संख्या बढ़ रही है, और इस समय, 3 हजार कार्यों को संसाधित करने में लगभग 30 मिनट लगते हैं, जो काफी लंबा समय है (हम चक्र के समय को 15 मिनट से अधिक नहीं रखने का प्रयास करते हैं)। मुझे लगता है कि भविष्य में, कलेक्टर की वास्तुकला संदेश कतारों की दिशा में बदल जाएगी और कलेक्टरों के अलगाव ऑपरेटिंग मोड के अनुसार नहीं, बल्कि आसान क्षैतिज स्केलेबिलिटी के लिए एक और मानदंड (नेटवर्क प्रकार, उपयोगकर्ता क्लस्टर) के अनुसार होगा।

Node.js में एक लाख लाइक्स या टास्क कतारों को इकट्ठा करें