Я нашел много вещей, связанных с использованием pg-обещания и await / async, но ничего, что вполне отвечает моей проблеме с async (пакет node / npm) и, в частности, взаимодействие между async.queue и pg-запросы на обещание.
Моя проблема: Мне нужно асинхронно выполнить несколько миллионов вычислений (совпадений) и зафиксировать их результаты в одном асинхронном процессе в postgres db.Мой основной процесс - это обещание, которое сначала вычисляет все возможные различные комбинации двух записей из таблицы и сегментирует их на порции по тысяче пар за раз.
Эти порции по тысяче пар (то есть [[0,1], [0,2], ..., [0, 1000]] - это мой массив содержимого «первый индекс чанков»), который передается в экземпляр async.queue, который сначала выполняет вычисление совпадениязатем запись в БД.
То, что заставляло меня часами чесать голову, это то, что фиксация БД не работает, использует ли она операторы вставки или транзакции.Я точно знаю, какие функции я использую для работы с БД, поскольку я написал для них ручные тесты.
Мой основной код выглядит следующим образом:
'use strict';
const promise = require('bluebird');
const initOptions = {
promiseLib: promise
};
const pgp = require('pg-promise')(initOptions);
const cn = {connexion parameters...};
const db = pgp(cn);
const async = require('async');
var mainPromise = (db, php, param) => {
return new Promise(resolve => {
//some code computing the chunksArray using param
...
var q = async.queue((chunk, done) => {
var scores = performScoresCalculations(chunk);
//scores is an array containing the 1000 scores for any chunk of a 1000 pairs
performDbCommitting(db, scores);
//commit those scores to the db using pg-promise
done();
}, 10);
q.drain = () => {
resolve(arr);
//admittedly not quite sure about that part, haven't used async.queue much so far
}
q.push(chunksArray);
)}.catch(err => {
console.error(err);
});
};
Теперь мой массив результатов выглядит следующим образом:
[{column1: 'value1_0', column2: 'value2_0', ..., columnN: 'valueN_0'}, ...,{column1: 'value1_999', column2: 'value2_999', column3: 'value3_999'}] с тысячами записей в нем.
Моя функция executeDbCommitting выглядит следующим образом:
var performDbCommitting = (db, pgp, scores) => {
console.log('test1');
//displays 'test1', as expected
var query = pgp.helpers.insert(scores, ['column1', 'column2', 'column3'], 'myScoreTable');
console.log(query);
//display the full content of the query, as expected
db.any(query).then(data => {
console.log('test2');
//nothing is displayed
console.log(data);
//nothing is displayed
return;
}).catch(err => {
console.error(err);
});
}
Итак, вот моя проблема:
- при тестировании "вручную" executeDbCommitting работает отлично, я даже пробовал версию с транзакциями, такая же работает без нареканий,
- , когдаиспользуется в async.queue все, что в executeDbCommitting, кажется, работает до вызова db.any (query), о чем свидетельствует console.log, отображающий информацию правильно до этой точки,
- ошибка не возникает, вычисленияover chunksArray продолжайте работать группами по 1000, как и ожидалось,
- if Я проверяю любой из массивов (chunk, chunksArray, партитуры и т. д.), все в порядке, длины правильные, их содержимое тоже.
pg-обещание просто не нужнопихать мои 1000 записей за раз в базу данных при использовании с async.queue, и вот где я застрял.У меня нет проблем с представлением, что ошибка лежит на мне, это первый раз, когда я использую async.queue, особенно в сочетании с обещаниями bluebird и pg-обещанием.
Большое спасибо заранее, что нашли времяпрочитать это и пролить свет на эту проблему, если можете.