NodeJS 中的异步/等待

Posted 2023-02-25

【中文标题】NodeJS 中的异步/等待

【英文标题】：Async/Await In NodeJS

【发布时间】：2020-03-31 22:02:30

【问题描述】：

这在 NodeJS 中有效，但我必须在创建它的代码块中使用 recordObject。 即使在函数的开头声明了recordObject，它在代码块之外也是不可见的，因为函数在recordObject被检索之前继续。我想使用 async/await 来执行此操作，但在从 helpers.getMostRecent 返回后，我不知道在哪里获取 recordObject。

这是整个函数。 有人可以告诉我如何使用 Async/Await 执行此操作吗？ 谢谢，约翰

// Define a function which builds a webpage and all supporting client/server code for 
// adding records to a json table in the database.
meta.build.AddWebpage = function(tableId)

  let dataObject = ;
  dataObject.uniqueField01Value = "";   
  dataObject.uniqueField01Name = "table.tableName";           
  dataObject.path = '/dbMetadata/metadata.json';
  dataObject.queryString = 'WHERE:;tableId:;MatchesExactly:;' + tableId + ':;';  

  // Collect information about the webpage from the metadata.
  // 1. Look in metadata.json - Read the object for the given tableId.
  helpers.getMostRecent(dataObject, function(errorFromGetMostRecent, payload)
  
    if(!errorFromGetMostRecent) // Got the most recent record from getMostRecent
    
      // Used to decode the payload buffer into readable text.
      let decoder = new StringDecoder('utf8');    

      // This instance of the Writable object gives us a place for a callback to run when the payload is received.
      const writable = new Writable();

      // Called by pipeline below. Does something useful with the payload
      writable.write = function(payload)     
      
        let stringContainer = '';                 
        stringContainer = stringContainer + decoder.write(payload);
        let recordObject = JSON.parse(stringContainer);

        // recordObject is all the metadata for the table.
        // I can use recordObject to build the webpage and all supporting code for adding records to a table
        ///////////////My problem is I have to do everything else in this code block////////////////////
        ?????????????????????????????How can I change this to Async/Await???????????????????????????????

      ; // End of: writable.write = function(payload)...

      // Passes the payload stream to the writable object which calls writable.write 
      // which does something useful with the payload.
      pipeline
      (
        payload,
        writable,
        function(error)if(error)console.log('There was an error.');
      );

     // End of: if(!error) Got the most recent record from gitHashedPass
    else // There was indeed an error returned by getMostRecent when attempting to get the most current record.
    
      helpers.log // Log the error.
      (
        7,
        'bxpa2p2t7ps3wrd1dqu7' + '\n' + 
        'The following was the error message from getMostRecent:' + '\n' +                                             
        errorFromGetMostRecent + '\n'                                                 
      ); // End of: helpers.log // Log the error.
     // End of: Else // There was indeed an error returned by getHashedPass when attempting to get the most current record.



  ); //End of: helpers.getMostRecent(dataObject, function(errorFromGetMostRecent, payload)

  // Assemble the webpage string from the metadata in recordObject

// End of: meta.build.AddWebpage = function(tableId)...

这是上面的函数调用的函数getMostRecent，以获取我们希望添加到会计系统的表的json记录。

// Define a function to retrieve the most current record in a table for a given primary key.
// Serves as table validation for put handlers (editing existing records)
// Checks that a record with the supplied primary key exists to modify.
// Also checks that a candidate values supplied for a unique fields are not already in the table. Enforces uniqueness.
// Streams the most current record back to the calling function.
helpers.getMostRecent = function(data, callback)

  // No value set but declared here because we need access to these thoughout the entire function.
  let queryArray, queryString;

  // typical example of a queryString: 'WHERE:;userId:;MatchesExactly:;' + userId + ':;'
  queryString = data.queryString

  // Make an array out of the queryString where each phrase of the query is an element.
  queryArray = queryString.split(":;");       

  // Create an empty map data structure which will be used to merge records with the same primary key. 
  // Hard if not impossible do that with objects.
  let tableMap = new Map();

  // Create a variable to track whether or not candidate values for any unique fields have already been used.
  let uniqueValuesAlreadyUsed = false;

  // Create a variable to track the primary key of a record that we may encounter which 
  // is holding a candidate unique value (perhaps and email address or phone number).
  // If we encounter this primary key again as we proceed through the records then we will 
  // check to see if the candidate unique value has been changed or if the record has been deleted. 
  // If so we will set this variable to false signifying that the candidate unique value is available again.
  let primaryKeyOfRecordHoldingCandidateUniqueValue = false

  // This function sets up a stream where each chunk of data is a complete line in the table file.
  let readInterface = readline.createInterface
  (
     // specify the file to be read.
      input: fs.createReadStream(helpers.baseDir + data.path)
    
  );

  // Look at each record in the file.
  readInterface.on('line', function(line) 
  
    // Convert the JSON string (a single line from the table file) into lineValueObject.
    let lineValueObject = JSON.parse(line);    

    // Declare a variable to serve as a key in the map to manage the lineValueObject.
    let primaryKey = lineValueObject[queryArray[1]];  

    let shouldDeleteThisRecord = false; 

    if(lineValueObject.deleted === true) // if the record in the table file had the delete field set to true:
    
      // If this record was holding our candidate unique field value:
      if(lineValueObject[queryArray[1]] === primaryKeyOfRecordHoldingCandidateUniqueValue) 
      
        // The record holding our candidate unique field value has been deleted so...

        // The candidate unique field value is available.
        uniqueValuesAlreadyUsed = false;

        // There is no more reason to track this record.
        primaryKeyOfRecordHoldingCandidateUniqueValue = false;          
      

      // This is not the record we are trying to modify.
      // Remove this record from the map. 
      shouldDeleteThisRecord = true;      
    
    else // The record was not deleted
     
      // If the current record does not have a primary key matching the record we wish to change:
      if(lineValueObject[queryArray[1]] != queryArray[3])
      
        // Check if this record has the same unique field value we wish to write. 
        // In other words: Has the unique field value already been used?
        if(lineValueObject[data.uniqueField01Name] === data.uniqueField01Value)
        
          // Currently this unique field value is taken. 
          // As we proceed, we may encounter a record with this same primary key where the unique field value has been changed.
          // Or, with this same primary key that has been deleted. 
          // Either case will make the unique field value available again as we proceed through the records
          // So we need to compare the primary key of this record to other records that we encounter.

          // Flag that the unique field value is already being used.
          uniqueValuesAlreadyUsed = true;

          // Take note of the primary key so that we can check as we proceed if this record gets 
          // deleted or if the unique field value gets changed.
          primaryKeyOfRecordHoldingCandidateUniqueValue = lineValueObject[queryArray[3]];
         // End of: Check if this record has the same unique field value we wish to write. Is the unique field value already taken?

        // Well then - Not deleted, not the same key as the record we want to change, not the candidate unique field value so...
        // Check if this record was previously holding the candidate unique field value but is now changed.
        else if
        (
          primaryKeyOfRecordHoldingCandidateUniqueValue === lineValueObject[queryArray[1]]
          &&
          lineValueObject[data.uniqueField01Name] != data.uniqueField01Value 
        )
        
          // This record was tying up the candidate unique field value but is no longer holding it.
          // The candidate unique field value is available again.
          uniqueValuesAlreadyUsed = false;

          // There is no more reason to track this record.
          primaryKeyOfRecordHoldingCandidateUniqueValue = false;           
        

        // This is not the record we are trying to modify.
        // Remove this record from the map.
        shouldDeleteThisRecord = true;
       // End of: If the current record does not have a primary key matching the record we wish to change:

     //End of: else - The record was not deleted

    // If the record was not marked for deletion and has the primary key of the record we wish to change:
    // This is the record we are going to send back to the calling function.
    if(shouldDeleteThisRecord == false)
              
      // Update this record in the map.
      tableMap.set(primaryKey, lineValueObject);
    
    else // shouldDeleteThisRecord is true. This is not the record we are trying to modify. Don't send it back.
    
      tableMap.delete(primaryKey);
    

  ); // End of: readInterface.on('line', function(line)...
  // End of: Look at each record...


  // This listener fires after we have looked through all the records in the table file.
  // The callback function defined here will stream one record back to the clients browser.
  readInterface.on('close', function() 
            
    // This readable stream will be used to write the result of the merge to a new file.
    const sourceStream = new Readable(); 

    // Check that the specified record was found.
    if(tableMap.size === 0)
    
      helpers.log
      (
        4,
        'xwmv16fc90bzrbnvhbg5' + '\n' +
        'No record found for primary key' + '\n'
      );
            //params:  error 
      return callback("No record found for primary key");
    

    if(uniqueValuesAlreadyUsed === true)
    
      helpers.log
      (
        4,
        'xwmv16fc90bzrbnvhbg5' + '\n' +
        'This ' +  data.uniqueField01Name + ' value already exists' + '\n'
      );
            //params:  error 
      return callback('This ' +  data.uniqueField01Name + ' value already exists');
    

    for (const [key, valueObject] of tableMap)
    
      // Convert the data object to a string.
      let stringData = JSON.stringify(valueObject);     

      // Load the readable stream with data.
      sourceStream.push(stringData + '\n');                  
                    

    // Tell the stream no more data is coming.
    sourceStream.push(null);     

    //params:  error, json record
    callback(false, sourceStream);             

  ); // End of: readInterface.on('close', function()...   

; // End of: helpers.getMostRecent = function(data, callback)...
// End of: // Define a function to retrieve the most current record in a table for a given primary key.

【参考方案1】：

我建议阅读https://alligator.io/js/async-functions/ 或类似教程，以更好地理解async/await。有了这个，您可能会重写您的函数并使您的流程更易于使用。

使用async/await 将回调函数重写为函数的简单示例如下所示：

没有async/await：

function consumerFunction() 
  myFunction(1, function(error, value) 
    console.log(error, value); // outputs: null, 2
  );


function myFunction(value, callback) 
  setTimeout(() =>  // mimick async behaviour
    callback(null, value + 1);
  , 1000);

与async/await:

async function consumerFunction() 
  try 
    const value = myFunction(1);
    console.log(value); // outputs 2
   catch (error) 
    // Code to handle potential errors
  


function myFunction(value) 
  return new Promise(function (resolve, reject) 
    setTimeout(() =>  // mimick async behavoir
      resolve(value + 1); // success case
      // reject('Something bad happened'); // error case
    , 1000);
  )

【讨论】：

谢谢@tomahaug。通过您的示例开始看到它。我已经在使用 async/await 将流从 NodeJS 传递到带有 fetch 的浏览器。现在我从您的示例中看到，必须有一个承诺，而我的获取就是承诺。现在我想我只需要弄清楚上面代码中的哪些内容需要包含在一个 Promise 中，以及最终的 await 需要去哪里。感谢您为我指明正确的方向。

听起来您走在正确的道路上。祝你好运！