如何在Node.js中通过队列管理大量PostgreSQL数据库调用？

First, let's address a critical issue in your current code: you're calling pool3.end() after every query. This destroys the entire connection pool immediately, defeating the purpose of using a pool—connections should be reused, not created and destroyed for every single request. We'll fix that as part of the queue implementation.

Here's a step-by-step solution to build a query queue that processes one request per second, eliminating connection pool exhaustion:

1. Initialize the Connection Pool Globally #

Set up your PostgreSQL pool once at the start of your application (not inside each query handler):

const { Pool } = require('pg');
const credentialsPostGres = { /* Your DB credentials here */ };

// Global connection pool - initialize once, reuse indefinitely
const pool3 = new Pool(credentialsPostGres);

2. Build the Query Queue & Consumer #

We'll create an in-memory queue to hold pending queries, plus a consumer function that processes one task per second:

// Queue to store pending query tasks (each includes SQL and callback logic)
const queryQueue = [];
// Flag to prevent overlapping consumer runs
let isProcessing = false;

// Consumer function that processes one task from the queue
async function processNextQuery() {
  if (isProcessing || queryQueue.length === 0) return;
  
  isProcessing = true;
  const { sql, onSuccess, onError } = queryQueue.shift();
  
  try {
    // Execute the query using the global connection pool
    const results = await pool3.query(sql);
    // Format results to match your original code's output
    const formattedData = { data: Object.values(JSON.parse(JSON.stringify(results.rows))) };
    // Trigger success callback with the result
    onSuccess(formattedData);
  } catch (err) {
    const errorMsg = `${err} ERROR IN QUERY EXECUTION`;
    console.error(errorMsg);
    // Trigger error callback if provided
    if (onError) onError(err);
  } finally {
    isProcessing = false;
    // Wait 1 second before processing the next query
    setTimeout(processNextQuery, 1000);
  }
}

// Helper function to add queries to the queue
function addQueryToQueue(sql, onSuccess, onError) {
  queryQueue.push({ sql, onSuccess, onError });
  // Start processing if we aren't already
  if (!isProcessing) {
    processNextQuery();
  }
}

3. Rewrite Your Query Calls to Use the Queue #

Replace your original query logic with calls to addQueryToQueue:

// Your original query becomes this
const sql_call = "select colum1 from table2 where x = y"; // Keep your actual complex query here
addQueryToQueue(
  sql_call,
  (formattedResult) => {
    // Match your original success callback flow
    const res = [formattedResult];
    return callback(res, data); // Pass results to your existing callback
  },
  (err) => {
    // Optional: Add custom error handling here if needed
    console.error("Query failed:", err);
  }
);

Key Implementation Details #

• Connection Pool Reuse: By keeping the pool alive globally, we let PostgreSQL manage connection reuse efficiently, reducing overhead and avoiding unnecessary connection creation.
• Controlled Rate Limiting: The 1-second delay between queries ensures you never exceed your pool's 100-connection limit (since we only run one query at a time).
• Error Handling: We've preserved your error logging and added optional error callbacks to handle failures gracefully.
• Lightweight Queue: This in-memory solution works perfectly for your use case. If you need advanced features like persistent queues (to survive app restarts), retries, or adjustable concurrency, consider libraries like bull or bee-queue—but this simple implementation will solve your immediate overflow problem.

内容的提问来源于stack exchange，提问作者MBrownG