👮 🦂 🤟🏽 PubSub लगभग मुफ्त है: PostgreSQL में NOTIFY सुविधाएँ 👨🏽‍✈️ 🧚🏾 😀

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

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

हम NOTIFY / LISTEN तंत्र का उपयोग करके डेटा संचारित और प्राप्त करेंगे, और हम Node.js. के लिए एक मॉडल कार्यान्वयन एकत्र करेंगे

लेकिन इस रास्ते पर एक रेक है जिसे सावधानी से परिधि में लाना होगा।

प्रोटोकॉल सुविधाएँ

सुनें

LISTEN

Libpq लाइब्रेरी का उपयोग करने वाला एप्लिकेशन लिस्टेन कमांड को एक नियमित एसक्यूएल कमांड के रूप में निष्पादित करता है, और फिर इसे समय-समय पर नई सूचनाओं की जांच के लिए PQnotify फ़ंक्शन को कॉल करना होगा।

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

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

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

उदाहरण के लिए, psql के संदर्भ में, यह इस तरह दिखता है:

 _tmp=# LISTEN test; LISTEN _tmp=# SELECT 1; ?column? ---------- 1 (1 row) Asynchronous notification "test" with payload "abc123" received from server process with PID 63991.

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

NOTIFY

 NOTIFY  [ ,  ]

NOTIFY कमांड उन सभी क्लाइंट अनुप्रयोगों के लिए एक अतिरिक्त "संदेश" लाइन के साथ एक अधिसूचना घटना भेजता है जो पहले मौजूदा लिस्टेन डेटाबेस में निर्दिष्ट चैनल नाम के साथ एक चैनल को निष्पादित कर चुके हैं।
...
"संदेश" लाइन जिसे अधिसूचना के साथ प्रेषित किया जाएगा ... एक सरल पाठ स्थिरांक होना चाहिए। एक मानक कॉन्फ़िगरेशन में, इसकी लंबाई 8000 बाइट्स से कम होनी चाहिए ।

यही है, अगर हमारे "संदेश" में अचानक ASCII से कुछ बहुत अलग है, तो हमें इसे स्क्रीन करना होगा , और यदि यह 8000 बाइट्स (वर्ण नहीं!) के आकार से अधिक है, तो इसे ब्लॉकों में काट लें और फिर इसे गोंद करें । उसी समय, हमें ऐसे ब्लॉकों के हस्तांतरण को संसाधित करने के लिए चैनल बैंडविड्थ और सर्वर संसाधनों दोनों को बचाना चाहिए - अर्थात्, संभव सामग्री के रूप में छोटी सेवा "बाइंडिंग" को जोड़ें, लेकिन क्लाइंट एप्लिकेशन को "गला" न दें, इसे पैक करने के लिए मजबूर करें। gzip -9 ।

तंत्र के अतिरिक्त लाभों में से, कोई भी संदेश के "स्रोत" के लिए बाध्यकारी को नोट कर सकता है ...

... अतिरिक्त कार्य को जाँच से रोका जा सकता है यदि संकेतन प्रक्रिया (घटना डेटा में संकेतित) का पीआईडी सत्र के स्वयं के पीआईडी से मेल खाता है (आप इसे libpq से संपर्क करके पा सकते हैं)। यदि वे मेल खाते हैं, तो सत्र को अपने स्वयं के कार्यों की सूचना मिली, ताकि इसे अनदेखा किया जा सके।

... और गारंटी वितरण आदेश:

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

हम कुछ भी विशेष रूप से संयोजित नहीं करेंगे, इसलिए हमारा प्रत्येक अनुरोध एक अलग लेनदेन के अनुरूप होगा।

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

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

इसलिए, एक कनेक्शन का उपयोग करना बेहतर है जहां स्पष्ट रूप से कोई लेनदेन या लंबी क्वेरी नहीं है।

AccessExclusiveLock डेटाबेस 0 के कक्षा 1262 के ऑब्जेक्ट 0 पर

यदि अचानक आपके NOTIFYs ने ऐसे लॉक की अपेक्षा को कुंद और लॉग करना शुरू कर दिया, तो आप अभी भी "शॉर्ट पैंट से बाहर हो गए हैं", और यह "वयस्क" एमक्यू के बारे में सोचने का समय है।

आखिरकार, अधिसूचना कतार, हालांकि काफी बड़ी (मानक बिल्ड में 8 जीबी), अभी भी परिमित है। टॉम लेन के जवाब के अनुसार:

यह लॉक लेनदेन के अधिसूचित संदेश (ओं) को सम्मिलित करते समय आयोजित किया जाता है, जिसके बाद लेनदेन लॉक हो जाता है और जारी करता है।

यही है, चारों ओर काम करने के लिए बहुत अधिक विकल्प नहीं हैं:

कम बार भेजें
यही है, भेजे गए संकेतकों को एकत्र करने के लिए, अगर ये कुछ काउंटर हैं, एक लंबे अंतराल पर।
कम भेजें
उदाहरण के लिए, ट्रांसमिट किए गए JSON से एप्लिकेशन कुंजी मानों के दृष्टिकोण से "डिफ़ॉल्ट" को हटाने के लिए।
केवल संकेत भेजें , कोई सामग्री नहीं
एक विकल्प के रूप में - कई चैनल शुरू करने के लिए, प्रत्येक का नाम पहले से ही कुछ लागू अर्थों में होगा।
अभी भी डेटाबेस से एक शिपमेंट बनाते हैं

जटिल संदेश भेजना

शरीर की एन्कोडिंग

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

 NOTIFY test, E'\x20\x21'

 Asynchronous notification "test" with payload " !" received from server process with PID 63991.

लेकिन फिर से प्रलेखन की ओर मुड़ते हैं:

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

इसलिए, यहां तक कि win1251 से उद्धरण चिह्न-पंजा के सामान्य चिह्न के साथ, हम दुख में रोटी लेते हैं:

 NOTIFY test, E'\x98' -- ERROR: invalid byte sequence for encoding "UTF8": 0x98

चूँकि हम UTF-8 वर्णों को मैन्युअल रूप से एनकोड करना और उन्हें बाइट्स द्वारा लिखना चाहते हैं ", हम बेस 64 में पैक किए गए संदेश बॉडी को भेजने के लिए तुरंत सहमत होंगे, यदि सीमा \x20-\x7E बाहर कोई वर्ण हैं या, यदि आवश्यक हो, विभाजन। एक तरफ, पैकेजिंग की यह विधि अतिरेक को बहुत अधिक नहीं बढ़ाती है (4: 3 गुणांक), दूसरी तरफ, यह किसी भी भाषा में सिस्टम लाइब्रेरी के स्तर पर लागू किया जाता है और न्यूनतम अतिरिक्त भार प्रदान करेगा।

लेकिन भले ही हमारे पास "अजीब" अक्षर न हों, और संदेश एक सेगमेंट में फिट बैठता है, फिर भी एक विशेषता है - एपोस्ट्रोफ़ से बचना :

एक एपोस्ट्रोफ को एक पंक्ति में शामिल करने के लिए, इसके बगल में दो एपोस्ट्रोफिस लिखें, उदाहरण के लिए: 'जोन ऑफ आर्क'। ध्यान दें कि यह दोहरे उद्धरण (") के समान नहीं है।

खंड पहचान

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

वास्तव में, हम पहले से ही दो "निर्देशांक" जानते हैं - यह भेजने की प्रक्रिया का पीआईडी और प्रत्येक अधिसूचना में आने वाले चैनल का नाम है । और प्रोटोकॉल ही हमें खंडों के आगमन के आदेश की गारंटी देता है।

पड़ोसी लेखन

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

कंटेनर की सीमा

कई खंडों से एक अभिन्न कंटेनर को इकट्ठा करने के लिए, हमें इसके पूरा होने के क्षण को जानने की आवश्यकता है। इसके लिए दो विशिष्ट तरीके हैं:

उनमें से पहले में लक्ष्य आकार (बाइट्स या खंडों में) का स्थानांतरण
उनमें से प्रत्येक में अंतिम खंड [नहीं] के संकेत का संचरण

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

ऑब्जेक्ट ट्रांसफर

सादे टेक्स्ट स्ट्रिंग्स और JSON ऑब्जेक्ट्स को "संदेश" के रूप में प्रसारित करने के लिए, हम प्राप्तकर्ता पक्ष पर उलटा परिवर्तन के लिए एक और प्रतीक-चिह्न जोड़ते हैं।

चूंकि हमने बेस 64 में "नॉन-फॉर्मेट" को एनकोड करने का फैसला किया था, इसलिए झंडे के लिए हम किसी भी अनुमति वाले पात्रों को ले सकते हैं जो इस सेट में नहीं हैं।

कुल मिलाकर, हमें प्रेषित खंडों के लिए निम्नलिखित विकल्प मिले:

 -- ""   !simple string -- "  "  @{"a":1} --    base64 #<segment> --    base64 $<segment>

जैसा कि आप देख सकते हैं, यह विश्लेषण करने के लिए पर्याप्त है कि खंड को केवल पहला चरित्र प्राप्त करने के लिए यह समझने के लिए कि इसके साथ आगे क्या करने की आवश्यकता है।

एक पबबस कार्यान्वयन लागू करना

हमारा आवेदन Node.js पर होगा, इसलिए हम PostgreSQL के साथ काम करने के लिए नोड-पोस्टग्रेज मॉड्यूल का उपयोग करेंगे।

हम शुरुआती फ्रेम लिखते हैं

शुरू करने के लिए, चलिए PubSub को EventEmitter के वारिस के रूप में बनाते हैं , ताकि विशिष्ट चैनलों के लिए सदस्यता लेने वालों के लिए घटनाओं को उत्पन्न करने में सक्षम हो सकें:

 const util = require('util'); const EventEmitter = require('events').EventEmitter; const PubSub = function(connection, interval, skipSelf) { //     this.connection = connection; //        this.connection.on('notification', p._onmessage.bind(this)); //         this.skipSelf = skipSelf; //  "" setInterval(() => { this.connection.query('SELECT 1'); }, interval); //     ""  this.slices = {}; }; util.inherits(PubSub, EventEmitter); const p = PubSub.prototype;

हम चैनलों के साथ काम करते हैं

चूंकि LISTEN / UNLISTEN किसी भी तरह से कसम नहीं खाते हैं, जब किसी चैनल की पुनः सदस्यता लेते हैं या उस सदस्यता से जिसे हम सब्सक्राइब नहीं किया गया था, से सदस्यता नहीं लेते हैं, हम कुछ भी जटिल नहीं करेंगे।

 //     - "",      //     -        const quot = str => /^[_a-z][0-9a-z_\$]*$/.test(str) ? str : `"${str}"`; p.subscribe = function(channel) { this.connection.query(`LISTEN ${quot(channel)}`); return this; }; p.unsubscribe = function(channel) { this.connection.query(`UNLISTEN ${quot(channel)}`); return this; };

संदेश भेजना और प्राप्त करना

 const PAYLOAD_LIMIT = 8000 - 1; const PAYLOAD_FL_STR = '!'; const PAYLOAD_FL_OBJ = '@'; const PAYLOAD_FL_SEQ = '#'; const PAYLOAD_FL_FIN = '$'; const PAYLOAD_SZ_HEAD = 1; const PAYLOAD_SZ_DATA = PAYLOAD_LIMIT - PAYLOAD_SZ_HEAD; //  ""  const reASCII = /^[\x20-\x7E]*$/; //  p.publish = function(channel, payload) { let query = `NOTIFY ${quot(channel)}`; if (payload !== null && payload !== undefined) { //    -    let str = typeof payload == 'string' ? PAYLOAD_FL_STR + payload : PAYLOAD_FL_OBJ + JSON.stringify(payload); if (str.length > PAYLOAD_LIMIT || !reASCII.test(str)) { //   base64- const b64 = Buffer.from(str).toString('base64'); for (let pos = 0, len = b64.length; pos < len; pos += PAYLOAD_SZ_DATA) { let fin = pos + PAYLOAD_SZ_DATA; let seg = fin >= len ? PAYLOAD_FL_FIN + b64.slice(pos) : PAYLOAD_FL_SEQ + b64.slice(pos, fin); this.connection.query(`${query}, '${seg}'`); } } else { //        ? //     str = str.replace(/'/g, "''"); this.connection.query(`${query}, '${str}'`); } } else { //       this.connection.query(query); } return this; }; //    p._onmessage = function(msg) { const {processId, channel, payload} = msg; //  "" if (processId == this.connection.processID && this.skipSelf) { return; } // ""  const id = `${processId}:${channel}`; let rv; //   let fl = payload.charAt(0); if (fl == PAYLOAD_FL_SEQ || fl == PAYLOAD_FL_FIN) { // base64 const str = payload.slice(PAYLOAD_SZ_HEAD); const slices = this.slices; let b64; if (fl == PAYLOAD_FL_FIN) { //   if (slices[id]) { slices[id].push(str); b64 = slices[id].join(''); delete slices[id]; } else { b64 = str; } } else { //     if (slices[id]) { slices[id].push(str); } else { slices[id] = [str]; } } if (b64) { rv = Buffer.from(b64, 'base64').toString(); fl = rv.charAt(0); } } else { //  / rv = payload; } if (rv !== undefined) { //   '' let res = { processId , channel }; if (rv) { //       let data = rv.slice(1); res.payload = fl == PAYLOAD_FL_OBJ ? JSON.parse(data) : data; } this.emit(channel, res); } };

कुछ परीक्षण

 const pg = require('pg'); const pgsql = new pg.Client({ host : 'example-db' , port : 5432 , user : 'postgres' , password : 'postgres' , database : '_tmp' }); pgsql.connect(err => { let psA = new PubSub(pgsql, 1000); let psB = new PubSub(pgsql, 1000); let chA = 'channel:A'; let chB = 'channel:B'; psA.subscribe(chA); psB.subscribe(chB); psA.on(chA, (msg) => { console.log('A:rcv', msg); }); psB.on(chB, (msg) => { console.log('B:rcv', msg); }); psB.publish(chA); psB.publish(chA, 'simple string'); psB.publish(chA, '  '); psB.publish(chA, {a : 1}); psA.publish(chB, '   100  '.repeat(100)); });

सब कुछ काफी सरल है, इसलिए आप इसे आसानी से अपनी परियोजना में इस्तेमाल किए गए किसी अन्य पीएल पर लागू कर सकते हैं, उदाहरण के लिए अतुल्यकालिक सूचनाओं के साथ काम करने का आधार:

PubSub लगभग मुफ्त है: PostgreSQL में NOTIFY सुविधाएँ