javascript - RXJS PostgreSQL 在 nodejs 中的背壓

原始標題：如何在 JavaScript 中減慢 PgSQL 結果列的流動？

我在使用 Node.js v4.5.0 並搭配 RXJS(5.4.0) 和 PostgreSQL（驅動程式 "pg": "6.1.4"）時遇到記憶體不足的問題。

我手動這樣創建一個 PgSQL 行的可觀察對象：

return Rx.Observable.create((subscriber) => {
    pool.connect().then((client: pg.Client) => {
        const stream:any = client.query(query.toParam());
        stream.on('row', (row) => {
            subscriber.next(row);
        });

        stream.on('end', () => {
            subscriber.complete();
            client.release();
        });
    });
});

然後我將一些運算子附加到 rx 可觀察對象上並進行一些處理。請注意，從資料庫返回的行有點重。

調查讓我得出一個結論：從資料庫返回的行比處理它們的速度快得多。必須為那些大量資料保留記憶體以等待處理，這導致了記憶體不足的問題：

致命錯誤：CALL_AND_RETRY_LAST 分配失敗 - 程序內存不足

中止陷阱：6

我在 PostgreSQL 驅動程式中沒有看到任何暫停資料流的選項。有什麼方法可以解決這個問題嗎？

答案＃1

如果你使用同一位作者的 pg-cursor 函式庫，應該相對簡單明瞭：

return Rx.Observable.defer(() => pg.connect())
  .flatMap(client => {
    const cursor = client.query(new Cursor('SELECT * FROM some_table WHERE prop > $1', [100]))

    const observableCursor = Rx.Observable.bindNodeCallback(cursor.read.bind(cursor));

    // Get the first 100 items
    observableCursor(100)
      .map(processRows)
      // This will only emit after the first one completes
      // and will recursively call this for each result
      .expand(_ => 
        observableCursor(100)
          .map(processRows)
      )
      // Unsubscribes once we don't get any more results
      .takeWhile(rows => rows.length > 0)
  });

答案＃2

我正在嘗試把 @paulpdaniels 建議的解決方案寫得盡可能易讀，但我不確定這是否真的屬於反應式風格。

我也有不同的處理功能，這取決於上下文變數。我希望有一個架構可以輕鬆使用不同的 rxjs 運算子鏈，這取決於上下文條件，並隱藏 cursor.read 的邏輯，但我擔心這是不可能的。

必須有一個地方，在我明確告訴資料庫抓取下一批資料（cursor.read）時，並且這個地方必須是在我釋放大量資料內存之後。

以下解決方案使用 TypeScript：

const db = require('./libs/db.js');
import * as Cursor from 'pg-cursor';
import * as pg from 'pg';
import * as Rx from 'rxjs';

let cursor, readHandler;

const source = Rx.Observable.create((subscriber) => {
    db.getClient().then((client: pg.Client) => {

        cursor = client.query(new Cursor('SELECT * FROM some_table WHERE prop > $1', [100]));
        readHandler = function (err, rowsBulk) {
            if (err) {
                subscriber.error(err);
                return client.release();
            }
            if (!rowsBulk.length) {
                subscriber.complete();
                return client.release();
            }
            subscriber.next(rowsBulk);
        };
    }).then(() => {
        // fetch first 100 records
        cursor.read(100, readHandler);
    });
});

source.flatMap((item) => {
    return new Promise((resolve) => {
        console.log('processing batch of items');
        setTimeout(() => { // timeout simulates processing
            item = null;
            console.log('release memory');
            resolve({});
        }, 5000);
    });
}).map(() => {
    // fetch every next 100 records
    console.log('allocate new memory');
    cursor.read(100, readHandler);
    return '.';
}).subscribe(console.log, (err) => {
    console.log('Error occurred');
    console.error(err);
});