主页 / 问题 / 1542815
Accepted
Александр Беляко
Asked:2023-09-27 17:01:40 +0000 UTC2023-09-27 17:01:40 +0000 UTC

为什么 COMMIT 在 Postgres 中不起作用？

  • 772

请帮我弄清楚为什么 COMMIT 在 Postgres 中不起作用？

我正在解决这个问题。有一个带有分页功能的API。我对 1000 个对象发出同步页面请求。我将这些对象异步插入到数据库表中。我在 Node.js + pg 模块中有执行此操作的代码。该脚本在单独的用户下使用数据库。但是，我不理解以下行为：我运行脚本，同时通过 pgAdmin 中的另一个用户使用对象请求中的 select count(*)，我查看加载到表中的对象数量。尽管我在脚本中显式打开一个事务并在其完成时进行提交，但它并没有增长。只有在脚本完全完成其工作后，我才设法看到更改。如果您使用 ctrl-c 手动中断脚本的执行，则已经进行的提交也不会保存。我想了解为什么会发生这种情况，我尝试了不同的选项，查看了 Postgres 日志，

接下来是代码：db.js 文件

const Pool = require('pg').Pool
module.exports = Pool

索引.js 文件

const Pool = require('./db')
const itemsCount = 1000

const apiList = [
    'https://api1.example.com'
    'https://api2.example.com'
    'https://api3.example.com'
]

const dbOptions = {
    host: 'localhost',
    port: 5432,
    database: 'db_name',
    user: 'user',
    password: 'password',
    max: 20
}


apiList.forEach(api => {
    const url = api
    let page = 1
    let objects = []
    do {
        objects = getObjects(url, page) // эта функция постранично получает данные из API по 1000 штук.
        processObjects(objects, api) // эта функция загружает полученные объекты в БД.
        page++
    } while (objects.length === itemsCount)       
})


async function processObjects(objects, api) {
    const pool = new Pool(dbOptions)
    for (const object of objects) {
        const ID = object.ID
        const client = await pool.connect()
        const selectObjects = await client.query('select * from objects where id=$1', [ID])
        if (selectObjects.rows.length === 0) {
            try {
                await client.query('BEGIN;')
                const insertObjects = await client.query(`insert into objects ("id", "json", "api", "loaded")  
                VALUES ($1, $2, $3, $4)`, [ID, object, api, 'now()'])
                await client.query('COMMIT;')
            } catch (error) {
                console.error(error.stack)
            }
        }
        await client.end()
        client.release()
    }
    await pool.end()
}
javascript

1 个回答

  1. Best Answer

    这就是您需要编写的方式（getObjects-例如存根）：

    // @ts-check
const { setTimeout } = require('node:timers/promises');
const { randomUUID } = require('node:crypto');
const { Pool } = require('pg');

const pool = new Pool({
    host: 'localhost',
    port: 5432,
    database: 'db_name',
    user: 'user',
    password: 'password',
    max: 20
});

run().catch(console.error);

async function run() {
  const itemsCount = 1000;

  const apiList = [
    'https://api1.example.com',
    'https://api2.example.com',
    'https://api3.example.com',
  ];

  const res = await Promise.allSettled(
    apiList.map(async (url) => {
      console.log(`[START] ${url}`);
      await handleApiItem(url, itemsCount);
      console.log(`[DONE] ${url}`);
    }),
  );
  res.forEach((x) => {
    if (x.status === 'rejected') {
      console.error(x.reason);
    }
  });

  await pool.end().catch(console.error);
}

/**
 * Фyнкция, которая загрузит все данные из endpoint
 * @param {string} url
 * @param {number} [requestedItemsCount]
 */
async function handleApiItem(url, requestedItemsCount = 1000) {
  let page = 1;
  while (1) {
    try {
      const objects = await getObjects(
        url,
        { pageNumber: page, itemsCount: requestedItemsCount }
      );
      if (objects?.length > 0) {
        await processObjects(objects, url);
      }
      if (!objects || objects.length < requestedItemsCount) {
        break;
      }
      page++;      
    } catch (err) {
      console.error(err);
    }
  }
}

/**
 * Эта функция загружает полученные объекты в БД.
 * @param {Awaited<ReturnType<getObjects>>} objects 
 * @param {string} api 
 */
async function processObjects(objects, api) {
  const client = await pool.connect();

  for (const object of objects) {
    const ID = object.ID;
    const selectObjects = await client.query('select id from objects where id=$1', [ID]);
    if (selectObjects.rows.length === 0) {
      try {
        await client.query('BEGIN;');
        await client.query(
          `insert into objects ("id", "json", "api", "loaded") VALUES ($1, $2, $3, $4)`,
          [ID, object, api, 'now()']
        );
        await client.query('COMMIT;')
      } catch (error) {
        console.error(error);
      }
    }
  }

  client.release();
}

/**
 * Эта функция постранично получает данные из API по 1000 штук.
 * @param {string} url 
 * @param {{ pageNumber: number, itemsCount?: number }} options 
 */
async function getObjects(url, options) {
  const timeout = Math.trunc(10 * Math.random() + 1);
  await setTimeout(timeout);

  return options.pageNumber > 10 && timeout % 5 === 0
    ? []
    : Array.from(
        { length: options.itemsCount || 1000 },
        () => ({ ID: randomUUID() })
      );
}

    我会提供const selectObjects = await client.query('select id from objects where id=$1', [ID]);添加建议 and api=$2。

    • 0

相关问题