Node.js generic-pool如何设置请求排队超时？

Hey there! You’re totally right that generic-pool doesn’t include a built-in configuration for setting a timeout on pending connection requests (even with maxWaitingClients set). But don’t worry—we can add this logic ourselves using Promise.race() to wrap the pool’s acquire() method.

Here’s how to modify your code to add timeout handling: #

First, create a reusable helper function to generate a timeout promise:

// Helper function to reject after a specified timeout
function createAcquireTimeout(ms) {
  return new Promise((_, reject) => {
    setTimeout(() => {
      reject(new Error(`Failed to acquire connection: timed out after ${ms}ms`));
    }, ms);
  });
}

Then, update the connection acquisition part of your processPendingBlocks function. We’ll use Promise.race() to race the pool’s acquire() promise against our timeout promise—whichever resolves/rejects first wins:

async function processPendingBlocks(ProcessingMap, channelid, configPath) {
  setTimeout(async () => {
    let nextBlockNumber = fs.readFileSync(configPath, "utf8");
    let processBlock;
    do {
      processBlock = ProcessingMap.get(channelid, nextBlockNumber);
      if (processBlock == undefined) {
        break;
      }
      try {
        // Race the acquire request against a timeout (e.g., 5000ms = 5 seconds)
        const sock = await Promise.race([
          pool.acquire(),
          createAcquireTimeout(5000)
        ]);
        
        try {
          // Process the block with the acquired socket
          await blockProcessing.processBlockEvent(channelid, processBlock, sock, configPath, folderLog);
        } finally {
          // Always release the socket back to the pool, even if processing fails
          await pool.release(sock);
        }
      } catch (error) {
        console.error(`Failed to process block: ${error}`);
        // Optional: Handle timeout-specific errors separately
        if (error.message.includes("timed out")) {
          console.warn("Connection queue is overloaded—request timed out waiting for a socket");
        }
      }
      
      ProcessingMap.remove(channelid, nextBlockNumber);
      fs.writeFileSync(configPath, parseInt(nextBlockNumber, 10) + 1);
      nextBlockNumber = fs.readFileSync(configPath, "utf8");
    } while (true);
    processPendingBlocks(ProcessingMap, channelid, configPath);
  }, blockProcessInterval)
}

Key Notes: #

• Timeout Flexibility: Adjust the 5000 value in createAcquireTimeout() to match your desired waiting period (in milliseconds).
• Resource Safety: The finally block ensures the socket is always released back to the pool, even if block processing throws an error—this prevents connection leaks.
• Error Handling: We added a check to identify timeout errors separately, so you can add custom logic (like retries or alerting) if needed.

This approach works with all recent versions of generic-pool and plays nicely with your existing pool configuration (including maxWaitingClients).

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