调整后的mysql逐条数据处理

data-handler.js

/**
 * 获取所有原始数据
 */
function getIds() {
  var ids = [];
  var errList = [];
  return new Promise(function(resolve, reject) {
    connection
      .query(
        "select id,fans_id,fans_nickname,buyer_id,fans_type,fans_info from yztrade"
      )
      .stream()
      .pipe(
        stream.Transform({
          objectMode: true,
          transform: function(row, encoding, callback) {
            try {
              var fans = JSON.parse(row.fans_info);
              ids.push({
                id: row.id,
                fans
              });
              console.log(
                `完成id: ${row.id} ${decodeURIComponent(fans.fans_nickname)}`
              );
              setTimeout(function() {
                callback();
              }, 30);
            } catch (err) {
              console.log(`解析${row.id}出粗,已经记录,继续查询`);
              callback();
            }
          }
        })
      )
      .on("finish", function() {
        console.log("所有更新数据已经获取");
        if (errList.length) {
          console.log("未正常处理的id为:");
          console.log(errList);
        }
        resolve(ids);
      });
  });
}

/**
 * 进行处理
 */
function handler(ids,errList) {
  var trade = ids.shift();
  var errList = errList||[];
  var fans = trade.fans;
  var query = `UPDATE yztrade SET fans_id='${fans.fans_id}',fans_nickname='${fans.fans_nickname}',buyer_id='${fans.buyer_id}',fans_type='${fans.fans_type}' where id=${trade.id}`;
 
  connection.query(query, function(err) {
    if (err) {
      console.log(err)
      errList.push(trade.id);
      console.log(`${decodeURIComponent(fans.fans_nickname)} 分析出错`);
    } else {
      console.log(`${decodeURIComponent(fans.fans_nickname)} 分析完成`);
    }
    
    if (ids.length) {
      setTimeout(function() {
        handler(ids,errList);
      }, 50);
    } else {
      console.log("全部更新完毕");
      connection.end();
      if (errList.length) {
        console.log("没有更新成功的数据,请自己重新检查一下:");
        console.log(errList);
      }
    }
  });
}

getIds()
  .then(function(ids) {
    handler(ids);
  })
  .catch(function() {
    console.log("获取原始数据出错");
  });