#प्लगइन विकास

#पृष्ठभूमि

एकल-नोड वातावरण में, प्लगइन आमतौर पर इन-प्रोसेस स्थिति, इवेंट या कार्यों के माध्यम से अपनी आवश्यकताओं को पूरा कर सकते हैं। हालाँकि, क्लस्टर मोड में, वही प्लगइन एक साथ कई इंस्टेंस पर चल सकता है, जिससे निम्नलिखित विशिष्ट समस्याएँ सामने आती हैं:

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

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

#समाधान

#कैश घटक (Cache)

मेमोरी में संग्रहीत किए जाने वाले डेटा के लिए, प्रबंधन हेतु सिस्टम के बिल्ट-इन कैश घटक का उपयोग करने की सलाह दी जाती है।

• app.cache के माध्यम से डिफ़ॉल्ट कैश इंस्टेंस प्राप्त करें।
• Cache set/get/del/reset जैसे बुनियादी ऑपरेशन प्रदान करता है, और कैशिंग लॉजिक को इनकैप्सुलेट करने के लिए wrap और wrapWithCondition के साथ-साथ mset/mget/mdel जैसी बैच विधियों का भी समर्थन करता है।
• क्लस्टर में डिप्लॉय करते समय, साझा डेटा को एक स्थायी स्टोरेज (जैसे Redis) में रखने और इंस्टेंस रीस्टार्ट होने पर कैश के नुकसान को रोकने के लिए एक उचित ttl सेट करने की सलाह दी जाती है।

उदाहरण: plugin-auth में कैश इनिशियलाइज़ेशन और उपयोग

// packages/plugins/@nocobase/plugin-auth/src/server/plugin.ts
async load() {
  this.cache = await this.app.cacheManager.createCache({
    name: 'auth',
    prefix: 'auth',
    store: 'redis',
  });

  await this.cache.wrap('token:config', async () => {
    const repo = this.app.db.getRepository('tokenPolicies');
    return repo.findOne({ filterByTk: 'default' });
  }, 60 * 1000);
}

#सिंक मैसेज मैनेजर (SyncMessageManager)

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

• प्लगइन बेस क्लास ने sendSyncMessage को लागू किया है, जो आंतरिक रूप से app.syncMessageManager.publish को कॉल करता है और टकराव से बचने के लिए स्वचालित रूप से चैनल में एक एप्लिकेशन-लेवल प्रीफिक्स जोड़ता है।
• publish एक transaction निर्दिष्ट कर सकता है, और डेटाबेस ट्रांजेक्शन कमिट होने के बाद संदेश भेजा जाएगा, जिससे स्थिति और संदेश सिंक्रनाइज़ेशन सुनिश्चित होगा।
• अन्य इंस्टेंस से संदेशों को प्रोसेस करने के लिए handleSyncMessage का उपयोग करें। beforeLoad चरण के दौरान सब्सक्राइब करना कॉन्फ़िगरेशन परिवर्तनों और स्कीमा सिंक्रनाइज़ेशन जैसे परिदृश्यों के लिए बहुत उपयुक्त है।

उदाहरण: plugin-data-source-main कई नोड्स में स्कीमा की निरंतरता बनाए रखने के लिए सिंक संदेशों का उपयोग करता है

export class PluginDataSourceMainServer extends Plugin {
  async handleSyncMessage(message) {
    if (message.type === 'syncCollection') {
      await this.app.db.getRepository('collections').load(message.collectionName);
    }
  }

  private sendSchemaChange(data, options) {
    this.sendSyncMessage(data, options); // स्वचालित रूप से app.syncMessageManager.publish को कॉल करता है
  }
}

#पब/सब मैनेजर (PubSubManager)

संदेश प्रसारण सिंक संकेतों का अंतर्निहित घटक है और इसे सीधे भी उपयोग किया जा सकता है। जब आपको इंस्टेंस के बीच संदेश प्रसारित करने की आवश्यकता हो, तो आप इस घटक का उपयोग कर सकते हैं।

• app.pubSubManager.subscribe(channel, handler, { debounce }) का उपयोग इंस्टेंस के बीच एक चैनल को सब्सक्राइब करने के लिए किया जा सकता है; debounce विकल्प का उपयोग बार-बार प्रसारण के कारण होने वाले लगातार कॉलबैक को रोकने के लिए किया जाता है।
• publish skipSelf (डिफ़ॉल्ट true है) और onlySelf का समर्थन करता है ताकि यह नियंत्रित किया जा सके कि संदेश वर्तमान इंस्टेंस पर वापस भेजा जाता है या नहीं।
• एप्लिकेशन शुरू होने से पहले एक एडाप्टर (जैसे Redis, RabbitMQ, आदि) को कॉन्फ़िगर किया जाना चाहिए; अन्यथा, यह डिफ़ॉल्ट रूप से एक बाहरी मैसेजिंग सिस्टम से कनेक्ट नहीं होगा।

उदाहरण: plugin-async-task-manager कार्य रद्द करने वाले इवेंट को प्रसारित करने के लिए PubSub का उपयोग करता है

const channel = `${plugin.name}.task.cancel`;

await this.app.pubSubManager.subscribe(channel, async ({ id }) => {
  this.logger.info(`Task ${id} cancelled on other node`);
  await this.stopLocalTask(id);
});

await this.app.pubSubManager.publish(channel, { id: taskId }, { skipSelf: true });

#इवेंट क्यू घटक (EventQueue)

मैसेज क्यू का उपयोग एसिंक्रोनस कार्यों को शेड्यूल करने के लिए किया जाता है, जो लंबे समय तक चलने वाले या पुनः प्रयास करने योग्य ऑपरेशनों को संभालने के लिए उपयुक्त है।

• app.eventQueue.subscribe(channel, { idle, process, concurrency }) के साथ एक उपभोक्ता घोषित करें। process एक Promise लौटाता है, और आप टाइमआउट को नियंत्रित करने के लिए AbortSignal.timeout का उपयोग कर सकते हैं।
• publish स्वचालित रूप से एप्लिकेशन नाम प्रीफिक्स जोड़ता है और timeout और maxRetries जैसे विकल्पों का समर्थन करता है। यह डिफ़ॉल्ट रूप से एक इन-मेमोरी क्यू एडाप्टर का उपयोग करता है, लेकिन आवश्यकतानुसार RabbitMQ जैसे विस्तारित एडाप्टर पर स्विच किया जा सकता है।
• एक क्लस्टर में, सुनिश्चित करें कि सभी नोड्स एक ही एडाप्टर का उपयोग करें ताकि नोड्स के बीच कार्य विखंडन से बचा जा सके।

उदाहरण: plugin-async-task-manager कार्यों को शेड्यूल करने के लिए EventQueue का उपयोग करता है

this.app.eventQueue.subscribe(`${plugin.name}.task`, {
  concurrency: this.concurrency,
  idle: this.idle,
  process: async (payload, { signal }) => {
    await this.runTask(payload.id, { signal });
  },
});

await this.app.eventQueue.publish(`${plugin.name}.task`, { id: taskId }, { maxRetries: 3 });

#डिस्ट्रीब्यूटेड लॉक मैनेजर (LockManager)

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

• डिफ़ॉल्ट रूप से, यह एक प्रोसेस-आधारित local एडाप्टर प्रदान करता है। आप Redis जैसे डिस्ट्रीब्यूटेड इम्प्लीमेंटेशन को रजिस्टर कर सकते हैं। समवर्तीता को नियंत्रित करने के लिए app.lockManager.runExclusive(key, fn, ttl) या acquire/tryAcquire का उपयोग करें।
• ttl का उपयोग लॉक को जारी करने के लिए एक सुरक्षा उपाय के रूप में किया जाता है, जिससे असाधारण मामलों में इसे अनिश्चित काल तक रोके जाने से रोका जा सके।
• सामान्य परिदृश्यों में शामिल हैं: स्कीमा परिवर्तन, डुप्लिकेट कार्यों को रोकना, रेट लिमिटिंग, आदि।

उदाहरण: plugin-data-source-main फ़ील्ड हटाने की प्रक्रिया को सुरक्षित रखने के लिए डिस्ट्रीब्यूटेड लॉक का उपयोग करता है

const lockKey = `${this.name}:fields.beforeDestroy:${collectionName}`;
await this.app.lockManager.runExclusive(lockKey, async () => {
  await fieldModel.remove(options);
  this.sendSyncMessage({ type: 'removeField', collectionName, fieldName });
});

#विकास संबंधी सुझाव

• इन-मेमोरी स्थिति की निरंतरता: विकास के दौरान इन-मेमोरी स्थिति का उपयोग करने से बचने का प्रयास करें। इसके बजाय, स्थिति की निरंतरता बनाए रखने के लिए कैशिंग या सिंक्रोनस संदेशों का उपयोग करें।
• बिल्ट-इन इंटरफेस का पुन: उपयोग करें: प्लगइन में क्रॉस-नोड संचार लॉजिक को फिर से लागू करने से बचने के लिए app.cache और app.syncMessageManager जैसी एकीकृत क्षमताओं का उपयोग करें।
• ट्रांजेक्शन सीमाओं पर ध्यान दें: डेटा और संदेश की निरंतरता सुनिश्चित करने के लिए ट्रांजेक्शन वाले ऑपरेशनों को transaction.afterCommit का उपयोग करना चाहिए (syncMessageManager.publish में यह बिल्ट-इन है)।
• बैकऑफ़ रणनीति विकसित करें: क्यू और प्रसारण कार्यों के लिए, असाधारण स्थितियों में नए ट्रैफिक स्पाइक्स को रोकने के लिए उचित timeout, maxRetries, और debounce मान सेट करें।
• पूरक मॉनिटरिंग और लॉगिंग का उपयोग करें: क्लस्टर में रुक-रुक कर आने वाली समस्याओं के निवारण को सुविधाजनक बनाने के लिए चैनल नाम, संदेश पेलोड, लॉक कुंजी आदि को रिकॉर्ड करने के लिए एप्लिकेशन लॉग का अच्छी तरह से उपयोग करें।

इन क्षमताओं के साथ, प्लगइन विभिन्न इंस्टेंस के बीच सुरक्षित रूप से स्थिति साझा कर सकते हैं, कॉन्फ़िगरेशन को सिंक्रनाइज़ कर सकते हैं और कार्यों को शेड्यूल कर सकते हैं, जिससे क्लस्टर डिप्लॉयमेंट परिदृश्यों की स्थिरता और निरंतरता आवश्यकताओं को पूरा किया जा सके।