Исправление "Сonnection storms" при интенсивных нагрузках по входу (и выходу тоже)

2022-05-13 23:21:03 +03:00 · 2022-05-13 23:21:03 +03:00 · db977f2bb7
commit db977f2bb7
parent 66628f5824
6 changed files with 150 additions and 7 deletions
--- a/config.js
+++ b/config.js
@ -40,7 +40,13 @@ let outGoing = {
    //Количество одновременно обрабатываемых исходящих сообщений
    nMaxWorkers: 3,
    //Интервал проверки наличия исходящих сообщений (мс)
-    nCheckTimeout: 1000
+    nCheckTimeout: 1000,
+    //Минимальный размер пула подключений к БД для обработчика исходящих сообщений
+    nPoolMin: 4,
+    //Максимальный размер пула подключений к БД для обработчика исходящих сообщений
+    nPoolMax: 4,
+    //Шаг инкремента подключений к БД в пуле обработчика исходящих сообщений
+    nPoolIncrement: 0
 };

 //Параметры обработки очереди входящих сообщений
@ -50,7 +56,13 @@ let inComing = {
    //Максимальный размер входящего сообщения (мб)
    nMsgMaxSize: 10,
    //Каталог размещения статических ресурсов
-    sStaticDir: "static"
+    sStaticDir: "static",
+    //Минимальный размер пула подключений к БД для обработчика входящих сообщений
+    nPoolMin: 10,
+    //Максимальный размер пула подключений к БД для обработчика входящих сообщений
+    nPoolMax: 10,
+    //Шаг инкремента подключений к БД в пуле обработчика входящих сообщений
+    nPoolIncrement: 0
 };

 //Параметры отправки E-Mail уведомлений
--- a/config_default.js
+++ b/config_default.js
@ -40,7 +40,13 @@ let outGoing = {
    //Количество одновременно обрабатываемых исходящих сообщений
    nMaxWorkers: 1,
    //Интервал проверки наличия исходящих сообщений (мс)
-    nCheckTimeout: 500
+    nCheckTimeout: 500,
+    //Минимальный размер пула подключений к БД для обработчика исходящих сообщений
+    nPoolMin: 4,
+    //Максимальный размер пула подключений к БД для обработчика исходящих сообщений
+    nPoolMax: 4,
+    //Шаг инкремента подключений к БД в пуле обработчика исходящих сообщений
+    nPoolIncrement: 0
 };

 //Параметры обработки очереди входящих сообщений
@ -50,7 +56,13 @@ let inComing = {
    //Максимальный размер входящего сообщения (мб)
    nMsgMaxSize: 10,
    //Каталог размещения статических ресурсов
-    sStaticDir: "static"
+    sStaticDir: "static",
+    //Минимальный размер пула подключений к БД для обработчика входящих сообщений
+    nPoolMin: 10,
+    //Максимальный размер пула подключений к БД для обработчика входящих сообщений
+    nPoolMax: 10,
+    //Шаг инкремента подключений к БД в пуле обработчика входящих сообщений
+    nPoolIncrement: 0
 };

 //Параметры отправки E-Mail уведомлений
--- a/core/app.js
+++ b/core/app.js
@ -199,7 +199,14 @@ class ParusAppServer {
                `Версия сервера приложений: ${prms.config.common.sVersion}, релиз: ${prms.config.common.sRelease}`
            );
            //Создаём подключение к БД
-            this.dbConn = new db.DBConnector({ connectSettings: prms.config.dbConnect });
+            this.dbConn = new db.DBConnector({
+                connectSettings: {
+                    ...prms.config.dbConnect,
+                    nPoolMin: prms.config.inComing.nPoolMin,
+                    nPoolMax: prms.config.inComing.nPoolMax,
+                    nPoolIncrement: prms.config.inComing.nPoolIncrement
+                }
+            });
            //Создаём модуль рассылки уведомлений
            this.notifier = new ntf.Notifier({ logger: this.logger, mail: prms.config.mail });
            //Создаём обработчик очереди исходящих
--- a/core/out_queue.js
+++ b/core/out_queue.js
@ -151,7 +151,12 @@ class OutQueue extends EventEmitter {
            //Отдаём команду дочернему процессу обработчика на старт исполнения
            prms.proc.send({
                nQueueId: prms.queue.nId,
-                connectSettings: this.dbConn.connectSettings,
+                connectSettings: {
+                    ...this.dbConn.connectSettings,
+                    nPoolMin: this.outGoing.nPoolMin,
+                    nPoolMax: this.outGoing.nPoolMax,
+                    nPoolIncrement: this.outGoing.nPoolIncrement
+                },
                service: _.find(this.services, { nId: prms.queue.nServiceId }),
                function: _.find(_.find(this.services, { nId: prms.queue.nServiceId }).functions, {
                    nId: prms.queue.nServiceFnId
@ -354,7 +359,7 @@ class OutQueue extends EventEmitter {
    async restartDetectingLoop() {
        //Включаем опрос очереди только если установлен флаг работы
        if (this.bWorking) {
-            this.nDetectingLoopTimeOut = await setTimeout(async () => {
+            this.nDetectingLoopTimeOut = setTimeout(async () => {
                await this.outDetectingLoop();
            }, this.outGoing.nCheckTimeout);
        }
--- a/models/obj_config.js
+++ b/models/obj_config.js
@ -28,6 +28,24 @@ const validateInComingPort = val => val >= 0 && val <= 65535 && Number.isInteger
 //Функция проверки значения порта сервера обслуживания входящих сообщений
 const validateMsgMaxSize = val => val >= 1 && val <= 1000 && Number.isInteger(val);

+//Функция проверки значения минимального размера пула подключений к БД для обработчика исходящих сообщений
+const validatePoolMinOutGoing = val => val >= 0 && val <= 10;
+
+//Функция проверки значения максимального размера пула подключений к БД для обработчика исходящих сообщений
+const validatePoolMaxOutGoing = val => val >= 1 && val <= 10;
+
+//Функция проверки значения шага инкремента подключений к БД в пуле обработчика исходящих сообщений
+const validatePoolIncrementOutGoing = val => val >= 0 && val <= 10;
+
+//Функция проверки значения минимального размера пула подключений к БД для обработчика входящих сообщений
+const validatePoolMinInComing = val => val >= 0 && val <= 1000;
+
+//Функция проверки значения максимального размера пула подключений к БД для обработчика входящих сообщений
+const validatePoolMaxInComing = val => val >= 1 && val <= 1000;
+
+//Функция проверки значения шага инкремента подключений к БД в пуле обработчика входящих сообщений
+const validatePoolIncrementInComing = val => val >= 0 && val <= 1000;
+
 //Схема валидации общих параметров сервера приложений
 const common = new Schema({
    //Версия сервера приложений
@ -160,6 +178,48 @@ const outGoing = new Schema({
            validateCheckTimeout: path =>
                `Значение интервала проверки наличия исходящих сообщений (${path}) должно быть целым числом в диапазоне от 100 до 60000`
        }
+    },
+    //Минимальный размер пула подключений к БД для обработчика исходящих сообщений
+    nPoolMin: {
+        type: Number,
+        required: true,
+        use: { validatePoolMinOutGoing },
+        message: {
+            type: path =>
+                `Минимальный размер пула подключений к БД для обработчика исходящих сообщений (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            required: path =>
+                `Не указан минимальный размер пула подключений к БД для обработчика исходящих сообщений (${path})`,
+            validatePoolMinOutGoing: path =>
+                `Значение минимального размера пула подключений к БД для обработчика исходящих сообщений (${path}) должно быть целым числом в диапазоне от 0 до 10`
+        }
+    },
+    //Максимальный размер пула подключений к БД для обработчика исходящих сообщений
+    nPoolMax: {
+        type: Number,
+        required: true,
+        use: { validatePoolMaxOutGoing },
+        message: {
+            type: path =>
+                `Максимальный размер пула подключений к БД для обработчика исходящих сообщений (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            required: path =>
+                `Не указан максимальный размер пула подключений к БД для обработчика исходящих сообщений (${path})`,
+            validatePoolMaxOutGoing: path =>
+                `Значение максимального размера пула подключений к БД для обработчика исходящих сообщений (${path}) должно быть целым числом в диапазоне от 1 до 10`
+        }
+    },
+    //Шаг инкремента подключений к БД в пуле обработчика исходящих сообщений
+    nPoolIncrement: {
+        type: Number,
+        required: true,
+        use: { validatePoolIncrementOutGoing },
+        message: {
+            type: path =>
+                `Шаг инкремента подключений к БД в пуле обработчика исходящих сообщений (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            required: path =>
+                `Не указан шаг инкремента подключений к БД в пуле обработчика исходящих сообщений (${path})`,
+            validatePoolIncrementOutGoing: path =>
+                `Значение шага инкремента подключений к БД в пуле обработчика исходящих сообщений (${path}) должно быть целым числом в диапазоне от 0 до 10`
+        }
    }
 });

@ -200,6 +260,48 @@ const inComing = new Schema({
                `Каталог размещения статических ресурсов (${path}) имеет некорректный тип данных (ожидалось - String)`,
            required: path => `Не указан каталог размещения статических ресурсов (${path})`
        }
+    },
+    //Минимальный размер пула подключений к БД для обработчика входящих сообщений
+    nPoolMin: {
+        type: Number,
+        required: true,
+        use: { validatePoolMinInComing },
+        message: {
+            type: path =>
+                `Минимальный размер пула подключений к БД для обработчика входящих сообщений (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            required: path =>
+                `Не указан минимальный размер пула подключений к БД для обработчика входящих сообщений (${path})`,
+            validatePoolMinInComing: path =>
+                `Значение минимального размера пула подключений к БД для обработчика входящих сообщений (${path}) должно быть целым числом в диапазоне от 0 до 1000`
+        }
+    },
+    //Максимальный размер пула подключений к БД для обработчика входящих сообщений
+    nPoolMax: {
+        type: Number,
+        required: true,
+        use: { validatePoolMaxInComing },
+        message: {
+            type: path =>
+                `Максимальный размер пула подключений к БД для обработчика входящих сообщений (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            required: path =>
+                `Не указан максимальный размер пула подключений к БД для обработчика входящих сообщений (${path})`,
+            validatePoolMaxInComing: path =>
+                `Значение максимального размера пула подключений к БД для обработчика входящих сообщений (${path}) должно быть целым числом в диапазоне от 1 до 1000`
+        }
+    },
+    //Шаг инкремента подключений к БД в пуле обработчика входящих сообщений
+    nPoolIncrement: {
+        type: Number,
+        required: true,
+        use: { validatePoolIncrementInComing },
+        message: {
+            type: path =>
+                `Шаг инкремента подключений к БД в пуле обработчика входящих сообщений (${path}) имеет некорректный тип данных (ожидалось - Number)`,
+            required: path =>
+                `Не указан шаг инкремента подключений к БД в пуле обработчика входящих сообщений (${path})`,
+            validatePoolIncrementInComing: path =>
+                `Значение шага инкремента подключений к БД в пуле обработчика входящих сообщений (${path}) должно быть целым числом в диапазоне от 0 до 1000`
+        }
    }
 });

--- a/modules/parus_oracle_db.js
+++ b/modules/parus_oracle_db.js
@ -28,9 +28,11 @@ const readCursorData = cursor => {
            rows.push(row);
        });
        queryStream.on("error", err => {
+            queryStream.destroy();
            reject(new Error(err.message));
        });
        queryStream.on("close", () => {
+            queryStream.destroy();
            resolve(rows);
        });
    });
@ -44,6 +46,9 @@ const connect = async prms => {
            password: prms.sPassword,
            connectString: prms.sConnectString,
            queueTimeout: 600000,
+            poolMin: prms.nPoolMin ? prms.nPoolMin : 4,
+            poolMax: prms.nPoolMax ? prms.nPoolMax : 4,
+            poolIncrement: prms.nPoolIncrement ? prms.nPoolIncrement : 0,
            sessionCallback: (connection, requestedTag, callback) => {
                if (prms.sSessionAppName) connection.module = prms.sSessionAppName;
                connection.execute(`ALTER SESSION SET CURRENT_SCHEMA=${prms.sSchema}`).then(